1import * as GCP from "@/GCP";
2import * as Effect from "effect/Effect";
3import * as Schedule from "effect/Schedule";
4import { serveProbes } from "../../bindingHost.ts";
5
6/** Topic the `Publish` binding writes to; `PublishTap` observes it. */
7export const PublishTopic = GCP.PubSub.Topic("PublishTopic", {});
8export const PublishTap = Effect.gen(function* () {
9 const topic = yield* PublishTopic;
10 return yield* GCP.PubSub.Subscription("PublishTap", { topic: topic.name });
11});
12
13/** Topic the `WriteTopic` client writes to; `WriteTap` observes it. */
14export const WriteTopicTarget = GCP.PubSub.Topic("WriteTopicTarget", {});
15export const WriteTap = Effect.gen(function* () {
16 const topic = yield* WriteTopicTarget;
17 return yield* GCP.PubSub.Subscription("WriteTap", { topic: topic.name });
18});
19
20/** Subscription the `Pull` / `Acknowledge` bindings consume. */
21export const PullTopic = GCP.PubSub.Topic("PullTopic", {});
22export const PullInbox = Effect.gen(function* () {
23 const topic = yield* PullTopic;
24 return yield* GCP.PubSub.Subscription("PullInbox", {
25 topic: topic.name,
26 ackDeadlineSeconds: 10,
27 });
28});
29
30/** Subscription the `ReadSubscription` client consumes. */
31export const ReadTopic = GCP.PubSub.Topic("ReadTopic", {});
32export const ReadInbox = Effect.gen(function* () {
33 const topic = yield* ReadTopic;
34 return yield* GCP.PubSub.Subscription("ReadInbox", {
35 topic: topic.name,
36 ackDeadlineSeconds: 10,
37 });
38});
39
40/** Avro schema for `GetSchema` / `ValidateMessage`. */
41export const EventSchema = GCP.PubSub.Schema("EventSchema", {
42 type: "AVRO",
43 definition: JSON.stringify({
44 type: "record",
45 name: "Event",
46 fields: [{ name: "id", type: "string" }],
47 }),
48});
49
50export const PUBLISH_PAYLOAD = "publish-binding";
51export const WRITE_PAYLOAD = "write-topic-binding";
52export const WRITE_BATCH_TEXT = "batch-text";
53/** Binary payload: must round-trip byte for byte. */
54export const WRITE_BATCH_BYTES = [0, 1, 255];
55
56const untilSome = <A>(page: ReadonlyArray<A> | undefined) =>
57 (page ?? []).length > 0;
58
59/**
60 * Effect-native Cloud Run service exercising every Pub/Sub binding as its
61 * own runtime service account. Deployed from {@link ../Bindings.test.ts}.
62 */
63export default class PubSubBindingsHost extends GCP.Function<PubSubBindingsHost>()(
64 "PubSubBindingsHost",
65 { main: import.meta.url, invokerIamDisabled: true },
66 Effect.gen(function* () {
67 const publish = yield* GCP.PubSub.Publish(PublishTopic);
68 const pull = yield* GCP.PubSub.Pull(PullInbox);
69 const acknowledge = yield* GCP.PubSub.Acknowledge(PullInbox);
70 const writeTopic = yield* GCP.PubSub.WriteTopic(WriteTopicTarget);
71 const readSubscription = yield* GCP.PubSub.ReadSubscription(ReadInbox);
72 const getSchema = yield* GCP.PubSub.GetSchema(EventSchema);
73 const validate = yield* GCP.PubSub.ValidateMessage(EventSchema);
74
75 return {
76 fetch: serveProbes({
77 publish: publish({
78 body: {
79 messages: [
80 {
81 data: Buffer.from(PUBLISH_PAYLOAD).toString("base64"),
82 attributes: { via: "Publish" },
83 },
84 ],
85 },
86 }),
87 pullAndAcknowledge: Effect.gen(function* () {
88 const page = yield* pull({
89 body: { maxMessages: 10, returnImmediately: false },
90 }).pipe(
91 Effect.repeat({
92 schedule: Schedule.spaced("1 second"),
93 until: (page) => untilSome(page.receivedMessages),
94 times: 20,
95 }),
96 );
97 const received = page.receivedMessages ?? [];
98 yield* acknowledge({
99 body: { ackIds: received.map((message) => message.ackId ?? "") },
100 });
101 return received.map((message) =>
102 Buffer.from(message.message?.data ?? "", "base64").toString("utf8"),
103 );
104 }),
105 writeTopic: Effect.gen(function* () {
106 const id = yield* writeTopic.publish({
107 data: WRITE_PAYLOAD,
108 attributes: { via: "WriteTopic" },
109 });
110 const batch = yield* writeTopic.publishBatch([
111 { data: WRITE_BATCH_TEXT },
112 { data: new Uint8Array(WRITE_BATCH_BYTES) },
113 ]);
114 const empty = yield* writeTopic.publishBatch([]);
115 return { id, batch, empty };
116 }),
117 readSubscription: Effect.gen(function* () {
118 const messages = yield* readSubscription
119 .pull({ maxMessages: 10 })
120 .pipe(
121 Effect.repeat({
122 schedule: Schedule.spaced("1 second"),
123 until: untilSome,
124 times: 20,
125 }),
126 );
127 // Extend, then acknowledge: exercises every client method.
128 const ackIds = messages.map((message) => message.ackId);
129 yield* readSubscription.modifyAckDeadline(ackIds, 30);
130 yield* readSubscription.acknowledge(ackIds);
131 return messages.map((message) => ({
132 text: message.text,
133 attributes: message.attributes,
134 }));
135 }),
136 getSchema: getSchema({ view: "FULL" }),
137 validateValid: validate({
138 encoding: "JSON",
139 message: Buffer.from(JSON.stringify({ id: "abc" })).toString(
140 "base64",
141 ),
142 }),
143 validateInvalid: validate({
144 encoding: "JSON",
145 message: Buffer.from(JSON.stringify({ wrong: 1 })).toString("base64"),
146 }),
147 }),
148 };
149 }).pipe(
150 Effect.provide(GCP.PubSub.PublishHttp),
151 Effect.provide(GCP.PubSub.PullHttp),
152 Effect.provide(GCP.PubSub.AcknowledgeHttp),
153 Effect.provide(GCP.PubSub.WriteTopicHttp),
154 Effect.provide(GCP.PubSub.ReadSubscriptionHttp),
155 Effect.provide(GCP.PubSub.GetSchemaHttp),
156 Effect.provide(GCP.PubSub.ValidateMessageHttp),
157 ),
158) {}