1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3
4/** Every ingested event lands here first. */
5export const Events = GCP.PubSub.Topic("Events", {});
6
7/** The warehouse the drain job writes into. */
8export const Analytics = GCP.BigQuery.Dataset("Analytics", {
9 forceDestroy: true,
10});
11
12/**
13 * A pull subscription. Pub/Sub holds messages until they are acked, so
14 * the drain can run on a schedule instead of keeping a consumer warm.
15 *
16 * Dependent resources go in an `Effect.gen` because the topic has to
17 * exist before its name can be referenced. Yielding it twice is free —
18 * resources are keyed by logical id, so both hosts get the same
19 * subscription.
20 */
21export const Inbox = Effect.gen(function* () {
22 const topic = yield* Events;
23 return yield* GCP.PubSub.Subscription("Inbox", {
24 topic: topic.name,
25 ackDeadlineSeconds: 60,
26 });
27});
28
29/**
30 * `payload` is JSON rather than a column per attribute so producers can
31 * add fields without a schema migration.
32 */
33export const EventsTable = Effect.gen(function* () {
34 const dataset = yield* Analytics;
35 return yield* GCP.BigQuery.Table("EventsTable", {
36 datasetId: dataset.datasetId,
37 tableId: "events",
38 schema: [
39 { name: "id", type: "STRING", mode: "REQUIRED" },
40 { name: "type", type: "STRING", mode: "REQUIRED" },
41 { name: "occurredAt", type: "TIMESTAMP", mode: "REQUIRED" },
42 { name: "payload", type: "JSON" },
43 ],
44 });
45});
46
47/**
48 * The envelope producers publish and the drain writes to BigQuery.
49 *
50 * A type alias rather than an interface: BigQuery's row payload is an
51 * index-signature type, and only aliases get an implicit index signature.
52 */
53export type EventRow = {
54 id: string;
55 type: string;
56 occurredAt: string;
57 payload: string;
58};