1import type * as pubsub from "@distilled.cloud/gcp/pubsub_v1";
2import type * as Effect from "effect/Effect";
3import * as Binding from "../../Binding.ts";
4import type { RuntimeContext } from "../../RuntimeContext.ts";
5import type { Subscription } from "./Subscription.ts";
6
7export interface PullSubscriptionOptions {
8 /**
9 * Maximum messages to return.
10 * @default 10
11 */
12 maxMessages?: number;
13 /**
14 * Return immediately when no messages are available instead of waiting
15 * briefly for some to arrive.
16 * @default false
17 */
18 returnImmediately?: boolean;
19}
20
21/** A pulled message with its payload decoded. */
22export interface PulledMessage extends pubsub.ReceivedMessage {
23 /** Ack id for `acknowledge` / `modifyAckDeadline`. */
24 ackId: string;
25 /** Server-assigned message id. */
26 messageId: string;
27 /** Decoded message body. */
28 data: Uint8Array;
29 /** `data` decoded as UTF-8. */
30 text: string;
31 /** Message attributes (empty when none were set). */
32 attributes: Record<string, string>;
33}
34
35/** Consumer client for one Pub/Sub subscription. */
36export interface ReadSubscriptionClient {
37 /** Pull up to `maxMessages` messages; an empty array when none arrived. */
38 pull(
39 options?: PullSubscriptionOptions,
40 ): Effect.Effect<
41 PulledMessage[],
42 pubsub.PullProjectsSubscriptionsError,
43 RuntimeContext
44 >;
45 /** Acknowledge messages by ack id. An empty list is a no-op. */
46 acknowledge(
47 ackIds: ReadonlyArray<string>,
48 ): Effect.Effect<
49 void,
50 pubsub.AcknowledgeProjectsSubscriptionsError,
51 RuntimeContext
52 >;
53 /**
54 * Extend (or, with `0`, release for redelivery) the ack deadline of
55 * messages. An empty list is a no-op.
56 */
57 modifyAckDeadline(
58 ackIds: ReadonlyArray<string>,
59 seconds: number,
60 ): Effect.Effect<
61 void,
62 pubsub.ModifyAckDeadlineProjectsSubscriptionsError,
63 RuntimeContext
64 >;
65}
66
67/**
68 * Consume access to a Pub/Sub {@link Subscription}: `pull`, `acknowledge`,
69 * `modifyAckDeadline`. Grants `roles/pubsub.subscriber` on the subscription
70 * only.
71 *
72 * ### Consuming messages
73 * **Example:** Pull and acknowledge
74 * ```typescript
75 * const inbox = yield* GCP.PubSub.ReadSubscription(subscription);
76 * const messages = yield* inbox.pull({ maxMessages: 10 });
77 * for (const message of messages) {
78 * yield* Effect.log(message.text, message.attributes);
79 * }
80 * yield* inbox.acknowledge(messages.map((m) => m.ackId));
81 * // …provided with Effect.provide(GCP.PubSub.ReadSubscriptionHttp)
82 * ```
83 *
84 * **Example:** Release a message for redelivery
85 * ```typescript
86 * yield* inbox.modifyAckDeadline([message.ackId], 0);
87 * ```
88 *
89 * @binding
90 * @category PubSub
91 */
92export interface ReadSubscription extends Binding.Service<
93 ReadSubscription,
94 "GCP.PubSub.ReadSubscription",
95 (subscription: Subscription) => Effect.Effect<ReadSubscriptionClient>
96> {}
97
98export const ReadSubscription = Binding.Service<ReadSubscription>(
99 "GCP.PubSub.ReadSubscription",
100);