1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3
4/**
5 * Every order event is published here once. Each consumer gets its own
6 * subscription (created for it by `GCP.Run.TopicEventSource`), so each
7 * sees every event it subscribes to, independently of the others.
8 */
9export const OrderEvents = GCP.PubSub.Topic("OrderEvents", {});
10
11/**
12 * Where Pub/Sub parks events the email consumer keeps failing on, after
13 * `maxDeliveryAttempts` tries.
14 */
15export const DeadOrderEvents = GCP.PubSub.Topic("DeadOrderEvents", {});
16
17/**
18 * A topic without a subscription drops what is published to it, so the
19 * dead letters need one to be kept. It is a pull subscription: an
20 * operator (or the test) pulls from it to inspect or replay failures.
21 */
22export const DeadOrderEventsInbox = Effect.gen(function* () {
23 const topic = yield* DeadOrderEvents;
24 return yield* GCP.PubSub.Subscription("DeadOrderEventsInbox", {
25 topic: topic.name,
26 // Keep dead letters for the maximum 7 days.
27 messageRetentionDuration: "604800s",
28 });
29});
30
31/** The email consumer's outbox: one object per confirmation it sends. */
32export const Outbox = GCP.Storage.Bucket("Outbox", {
33 forceDestroy: true,
34});
35
36/** The analytics warehouse. */
37export const Warehouse = GCP.BigQuery.Dataset("Warehouse", {
38 forceDestroy: true,
39});
40
41/** One row per event the analytics consumer receives, of every type. */
42export const OrderEventsTable = Effect.gen(function* () {
43 const dataset = yield* Warehouse;
44 return yield* GCP.BigQuery.Table("OrderEventsTable", {
45 datasetId: dataset.datasetId,
46 tableId: "order_events",
47 schema: [
48 { name: "eventId", type: "STRING", mode: "REQUIRED" },
49 { name: "type", type: "STRING", mode: "REQUIRED" },
50 { name: "orderId", type: "STRING", mode: "REQUIRED" },
51 { name: "email", type: "STRING", mode: "REQUIRED" },
52 { name: "total", type: "FLOAT", mode: "REQUIRED" },
53 { name: "occurredAt", type: "TIMESTAMP", mode: "REQUIRED" },
54 { name: "receivedAt", type: "TIMESTAMP", mode: "REQUIRED" },
55 ],
56 });
57});
58
59export type OrderEventType = "order.created" | "order.cancelled";
60
61/**
62 * What the API publishes. The type is also set as the message's `type`
63 * attribute, which is what subscription filters match on.
64 */
65export interface OrderEvent {
66 eventId: string;
67 type: OrderEventType;
68 orderId: string;
69 email: string;
70 total: number;
71 occurredAt: string;
72}
73
74/** Outbox object for the confirmation of one event. */
75export const emailObjectFor = (eventId: string) => `emails/${eventId}.json`;
76
77/** Decode a pushed message's base64 JSON body. */
78export const decodeOrderEvent = (data: string | undefined) =>
79 Effect.try(
80 () =>
81 JSON.parse(
82 Buffer.from(data ?? "", "base64").toString("utf8"),
83 ) as OrderEvent,
84 );