1import * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import * as Effect from "effect/Effect";
3import { bindGcpHost } from "../Host.ts";
4import { grantFor, type BindingIam } from "../HttpBinding.ts";
5import type {
6 PulledMessage,
7 ReadSubscriptionClient,
8} from "./ReadSubscription.ts";
9import type { Subscription } from "./Subscription.ts";
10import type { Topic } from "./Topic.ts";
11import type { PublishMessage, WriteTopicClient } from "./WriteTopic.ts";
12
13/**
14 * Shared HTTP scaffolding for the Pub/Sub `WriteTopic` / `ReadSubscription`
15 * bindings. NOT exported from `index.ts`.
16 */
17
18export const writeTopicGrant: BindingIam = {
19 role: "roles/pubsub.publisher",
20 on: "pubsub.topic",
21};
22
23export const readSubscriptionGrant: BindingIam = {
24 role: "roles/pubsub.subscriber",
25 on: "pubsub.subscription",
26};
27
28const encode = (data: string | Uint8Array) =>
29 typeof data === "string"
30 ? Buffer.from(data, "utf8").toString("base64")
31 : Buffer.from(data.buffer, data.byteOffset, data.byteLength).toString(
32 "base64",
33 );
34
35const toPubsubMessage = (message: PublishMessage): pubsub.PubsubMessage => ({
36 data: encode(message.data),
37 attributes: message.attributes,
38 orderingKey: message.orderingKey,
39});
40
41const decode = (received: pubsub.ReceivedMessage): PulledMessage => {
42 const buffer = Buffer.from(received.message?.data ?? "", "base64");
43 return {
44 ...received,
45 ackId: received.ackId ?? "",
46 messageId: received.message?.messageId ?? "",
47 data: new Uint8Array(buffer),
48 text: buffer.toString("utf8"),
49 attributes: Object.fromEntries(
50 Object.entries(received.message?.attributes ?? {}).filter(
51 (entry): entry is [string, string] => entry[1] !== undefined,
52 ),
53 ),
54 };
55};
56
57export const makeWriteTopicBinding = Effect.gen(function* () {
58 const publish = yield* pubsub.publishProjectsTopics;
59 return Effect.fn(function* (topic: Topic) {
60 yield* bindGcpHost({
61 tag: "GCP.PubSub.WriteTopic",
62 resource: topic,
63 iam: [grantFor(writeTopicGrant, topic.name)],
64 });
65 const topicName = yield* topic.name;
66 const publishBatch = (messages: ReadonlyArray<PublishMessage>) =>
67 messages.length === 0
68 ? Effect.succeed<string[]>([])
69 : Effect.gen(function* () {
70 const body = yield* Effect.sync(() => ({
71 messages: messages.map(toPubsubMessage),
72 }));
73 const response = yield* publish({
74 topic: yield* topicName,
75 body,
76 });
77 return response.messageIds ?? [];
78 });
79 return {
80 publish: (message) =>
81 publishBatch([message]).pipe(Effect.map((ids) => ids[0] ?? "")),
82 publishBatch,
83 } satisfies WriteTopicClient;
84 });
85});
86
87export const makeReadSubscriptionBinding = Effect.gen(function* () {
88 const pull = yield* pubsub.pullProjectsSubscriptions;
89 const acknowledge = yield* pubsub.acknowledgeProjectsSubscriptions;
90 const modifyAckDeadline =
91 yield* pubsub.modifyAckDeadlineProjectsSubscriptions;
92 return Effect.fn(function* (subscription: Subscription) {
93 yield* bindGcpHost({
94 tag: "GCP.PubSub.ReadSubscription",
95 resource: subscription,
96 iam: [grantFor(readSubscriptionGrant, subscription.name)],
97 });
98 const subscriptionName = yield* subscription.name;
99 return {
100 pull: (options) =>
101 Effect.gen(function* () {
102 const response = yield* pull({
103 subscription: yield* subscriptionName,
104 body: {
105 maxMessages: options?.maxMessages ?? 10,
106 returnImmediately: options?.returnImmediately ?? false,
107 },
108 });
109 return yield* Effect.sync(() =>
110 (response.receivedMessages ?? []).map(decode),
111 );
112 }),
113 acknowledge: (ackIds) =>
114 ackIds.length === 0
115 ? Effect.void
116 : Effect.gen(function* () {
117 yield* acknowledge({
118 subscription: yield* subscriptionName,
119 body: { ackIds: [...ackIds] },
120 });
121 }),
122 modifyAckDeadline: (ackIds, seconds) =>
123 ackIds.length === 0
124 ? Effect.void
125 : Effect.gen(function* () {
126 yield* modifyAckDeadline({
127 subscription: yield* subscriptionName,
128 body: { ackIds: [...ackIds], ackDeadlineSeconds: seconds },
129 });
130 }),
131 } satisfies ReadSubscriptionClient;
132 });
133});