1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import * as Namespace from "../../Namespace.ts";
6import { Topic } from "../PubSub/Topic.ts";
7import { TopicEventSource } from "../PubSub/TopicEventSource.ts";
8import type { Bucket } from "./Bucket.ts";
9import {
10 BucketEventSource,
11 toBucketEvent,
12 type BucketEvent,
13 type BucketEventSourceProps,
14 type BucketEventSourceService,
15} from "./BucketEventSource.ts";
16import { Notification } from "./Notification.ts";
17
18/**
19 * Implementation of `GCP.Storage.BucketEventSource` over Pub/Sub.
20 *
21 * Deploy-time: creates a Pub/Sub topic owned by the host
22 * (`{bucket}-BucketEvents`) and a `JSON_API_V1` bucket notification that
23 * publishes the requested event types (and object prefix) to it; the
24 * notification grants the Cloud Storage service agent
25 * `roles/pubsub.publisher` on the topic. Delivery is delegated to the
26 * provided `GCP.PubSub.TopicEventSource` implementation, so provide one
27 * alongside this layer: `GCP.Run.TopicEventSource` (push, HTTP hosts) or
28 * `GCP.Run.TopicPullEventSource` (pull, Jobs and WorkerPools). Runtime:
29 * each message is parsed into a `BucketEvent`.
30 *
31 * ### Choosing the delivery
32 * **Example:** Push to a Cloud Run service
33 * ```typescript
34 * Effect.gen(function* () {
35 * yield* GCP.Storage.consumeBucketEvents(bucket, (events) =>
36 * events.pipe(Stream.runForEach((event) => Effect.log(event.object))),
37 * );
38 * }).pipe(
39 * Effect.provide(GCP.Storage.BucketEventSourceLive),
40 * Effect.provide(GCP.Run.TopicEventSource),
41 * );
42 * ```
43 *
44 * **Example:** Pull on a worker pool
45 * ```typescript
46 * Effect.gen(function* () {
47 * yield* GCP.Storage.consumeBucketEvents(bucket, (events) =>
48 * events.pipe(Stream.runForEach((event) => Effect.log(event.object))),
49 * );
50 * return { run: Effect.never };
51 * }).pipe(
52 * Effect.provide(GCP.Storage.BucketEventSourceLive),
53 * Effect.provide(GCP.Run.TopicPullEventSource),
54 * );
55 * ```
56 *
57 * @layer
58 * @provides GCP.Storage.BucketEventSource
59 * @category Storage
60 */
61export const BucketEventSourceLive = Layer.effect(
62 BucketEventSource,
63 Effect.gen(function* () {
64 const topics = yield* Topic;
65 const notifications = yield* Notification;
66 const source = yield* TopicEventSource;
67
68 return Effect.fn(function* <Req = never>(
69 bucket: Bucket,
70 props: BucketEventSourceProps,
71 process: (
72 events: Stream.Stream<BucketEvent>,
73 ) => Effect.Effect<void, never, Req>,
74 ) {
75 const host = yield* Binding.Host;
76 const hostId = host?.LogicalId ?? "Host";
77 // Declared in both phases: the delivery implementation declares its
78 // subscription against this topic at runtime too.
79 const topic = yield* Namespace.push(
80 hostId,
81 topics(`${bucket.LogicalId}-BucketEvents`, {}),
82 );
83
84 if (!globalThis.__ALCHEMY_RUNTIME__) {
85 yield* Namespace.push(
86 hostId,
87 notifications(`${bucket.LogicalId}-Notification`, {
88 bucketName: bucket.bucketName,
89 topic: topic.name,
90 payloadFormat: "JSON_API_V1",
91 eventTypes: props.eventTypes ?? ["OBJECT_FINALIZE"],
92 objectNamePrefix: props.prefix,
93 }),
94 );
95 }
96
97 const { eventTypes: _eventTypes, prefix: _prefix, ...delivery } = props;
98 yield* source(topic, delivery, (messages) =>
99 process(messages.pipe(Stream.mapEffect(toBucketEvent))),
100 );
101 }) as BucketEventSourceService;
102 }),
103);