1import * as GCP from "@/GCP";
2import type { TopicMessage } from "@/GCP/PubSub/TopicEventSource.ts";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import * as HttpServerResponse from "effect/http/HttpServerResponse";
6
7/** Marker object a consumer writes for one delivered message. */
8export const markerFor = (messageId: string) => `markers/${messageId}.json`;
9
10export const PushOrders = GCP.PubSub.Topic("PushOrders", {});
11export const PullOrders = GCP.PubSub.Topic("PullOrders", {});
12export const Markers = GCP.Storage.Bucket("TopicMarkers", {
13 location: "US-CENTRAL1",
14 forceDestroy: true,
15});
16
17const recordMessages =
18 (
19 putObject: (request: {
20 name: string;
21 body: string;
22 }) => Effect.Effect<unknown, unknown, any>,
23 ) =>
24 (messages: Stream.Stream<TopicMessage>) =>
25 messages.pipe(
26 Stream.runForEach(({ message, subscription }) =>
27 putObject({
28 name: markerFor(message.messageId ?? "unknown"),
29 body: JSON.stringify({
30 data: atob(message.data ?? ""),
31 attributes: message.attributes ?? {},
32 subscription,
33 }),
34 }).pipe(Effect.asVoid, Effect.orDie),
35 ),
36 );
37
38/**
39 * Effect-native Cloud Run service consuming a topic over push. The
40 * service keeps the default IAM invoker check, so deliveries must carry a
41 * token Cloud Run accepts.
42 * Deployed from {@link ../TopicEventSource.test.ts}.
43 */
44export class PushConsumer extends GCP.Function<PushConsumer>()(
45 "PushConsumer",
46 {
47 main: import.meta.url,
48 handler: "PushConsumer",
49 location: "us-central1",
50 },
51 Effect.gen(function* () {
52 const topic = yield* PushOrders;
53 const bucket = yield* Markers;
54 const putObject = yield* GCP.Storage.PutObject(bucket);
55 yield* GCP.PubSub.consumeTopicMessages(topic, recordMessages(putObject));
56 return {
57 fetch: Effect.succeed(HttpServerResponse.text("ok")),
58 };
59 }).pipe(
60 Effect.provide(GCP.Run.TopicEventSource),
61 Effect.provide(GCP.Storage.PutObjectHttp),
62 ),
63) {}
64
65/**
66 * Effect-native Cloud Run worker pool consuming a topic with a pull loop.
67 * Deployed from {@link ../TopicEventSource.test.ts}.
68 */
69export class PullConsumer extends GCP.Run.WorkerPool<PullConsumer>()(
70 "PullConsumer",
71 {
72 main: import.meta.url,
73 handler: "PullConsumer",
74 location: "us-central1",
75 scaling: { manualInstanceCount: 1 },
76 },
77 Effect.gen(function* () {
78 const topic = yield* PullOrders;
79 const bucket = yield* Markers;
80 const putObject = yield* GCP.Storage.PutObject(bucket);
81 yield* GCP.PubSub.consumeTopicMessages(
82 topic,
83 { maxMessages: 5 },
84 recordMessages(putObject),
85 );
86 }).pipe(
87 Effect.provide(GCP.Run.TopicPullEventSource),
88 Effect.provide(GCP.Storage.PutObjectHttp),
89 ),
90) {}