1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3import { HttpServerRequest } from "effect/http/HttpServerRequest";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import Drain from "./Drain.ts";
6import { Events, EventsTable, type EventRow } from "./resources.ts";
7
8/**
9 * The front door of an analytics pipeline.
10 *
11 * Producers post events; the service publishes them to Pub/Sub and
12 * returns. Nothing touches BigQuery on the request path, so a slow
13 * warehouse or a schema change cannot take the ingest endpoint down —
14 * the messages just queue up until {@link Drain} runs.
15 *
16 * - `POST /events` — accept an event and publish it.
17 * - `POST /drain` — start a drain now, instead of waiting for a schedule.
18 * - `GET /events/count` — count what has landed in BigQuery.
19 */
20export default class Ingest extends GCP.Function<Ingest>()(
21 "Ingest",
22 {
23 main: import.meta.url,
24 invokerIamDisabled: true,
25 },
26 Effect.gen(function* () {
27 const topic = yield* Events;
28 const table = yield* EventsTable;
29 const drain = yield* Drain;
30
31 // pubsub.publisher on the topic only.
32 const publisher = yield* GCP.PubSub.WriteTopic(topic);
33 // bigquery.dataViewer on the table, plus bigquery.jobUser on the
34 // project — BigQuery only grants running query jobs there.
35 const warehouse = yield* GCP.BigQuery.ReadTable(table);
36 // Binding a Job to a Service grants run.jobsExecutorWithOverrides on
37 // the job — this is how one host triggers another.
38 const runDrain = yield* GCP.Run.RunJob(drain);
39
40 // An accessor: the table id is bound at deploy time and read back
41 // inside the handler.
42 const tableId = yield* table.tableId;
43
44 return {
45 fetch: Effect.gen(function* () {
46 const request = yield* HttpServerRequest;
47 const url = new URL(request.originalUrl);
48
49 if (request.method === "GET" && url.pathname === "/") {
50 return HttpServerResponse.text("ok");
51 }
52
53 if (request.method === "POST" && url.pathname === "/events") {
54 const body = (yield* request.json) as {
55 type?: string;
56 payload?: unknown;
57 };
58 if (!body.type) {
59 return yield* HttpServerResponse.json(
60 { error: "type is required" },
61 { status: 400 },
62 );
63 }
64
65 const event: EventRow = {
66 id: crypto.randomUUID(),
67 type: body.type,
68 occurredAt: new Date().toISOString(),
69 payload: JSON.stringify(body.payload ?? {}),
70 };
71
72 yield* publisher
73 .publish({
74 data: JSON.stringify(event),
75 // Attributes are queryable without decoding the body,
76 // which lets a filtered subscription fan out by type.
77 attributes: { type: event.type },
78 })
79 .pipe(Effect.orDie);
80
81 return yield* HttpServerResponse.json(
82 { id: event.id },
83 { status: 202 },
84 );
85 }
86
87 // Cloud Run Jobs are asynchronous: this returns as soon as the
88 // execution is created, not when it finishes.
89 if (request.method === "POST" && url.pathname === "/drain") {
90 const operation = yield* runDrain().pipe(Effect.orDie);
91 return yield* HttpServerResponse.json(
92 { execution: operation.name ?? null },
93 { status: 202 },
94 );
95 }
96
97 if (request.method === "GET" && url.pathname === "/events/count") {
98 const type = url.searchParams.get("type");
99 const events = yield* tableId;
100 // Unqualified table names resolve against the bound table's
101 // dataset; `params` become named `@type` parameters.
102 const rows = yield* warehouse
103 .query(
104 type
105 ? `SELECT COUNT(*) AS n FROM \`${events}\` WHERE type = @type`
106 : `SELECT COUNT(*) AS n FROM \`${events}\``,
107 type ? { type } : undefined,
108 )
109 .pipe(Effect.orDie);
110
111 return yield* HttpServerResponse.json({
112 count: Number(rows[0]?.n ?? 0),
113 });
114 }
115
116 return yield* HttpServerResponse.json(
117 { error: "not found" },
118 { status: 404 },
119 );
120 }),
121 };
122 }).pipe(
123 Effect.provide([
124 GCP.PubSub.WriteTopicHttp,
125 GCP.BigQuery.ReadTableHttp,
126 GCP.Run.RunJobHttp,
127 ]),
128 ),
129) {}