1import type * as storage from "@distilled.cloud/gcp/storage_v1";
2import * as Effect from "effect/Effect";
3import type * as Stream from "effect/Stream";
4import * as Binding from "../../Binding.ts";
5import type {
6 TopicEventSourceProps,
7 TopicMessage,
8} from "../PubSub/TopicEventSource.ts";
9import type { Bucket } from "./Bucket.ts";
10
11/** Cloud Storage object event types delivered through Pub/Sub. */
12export type BucketEventType =
13 | "OBJECT_FINALIZE"
14 | "OBJECT_DELETE"
15 | "OBJECT_ARCHIVE"
16 | "OBJECT_METADATA_UPDATE";
17
18/** One Cloud Storage object change. */
19export interface BucketEvent {
20 /** What happened to the object. */
21 eventType: BucketEventType;
22 /** Bucket the object lives in. */
23 bucket: string;
24 /** Object name (key). */
25 object: string;
26 /** Object generation the event refers to. */
27 generation: string;
28 /** RFC3339 time the event occurred. */
29 eventTime: string;
30 /**
31 * Object metadata at the time of the event (the `JSON_API_V1` payload):
32 * `size`, `contentType`, `md5Hash`, `metadata`, …
33 */
34 metadata: storage.Storage_Object;
35}
36
37export interface BucketEventSourceProps extends Omit<
38 TopicEventSourceProps,
39 "filter"
40> {
41 /**
42 * Event types to deliver. Filtered by the bucket notification, so other
43 * events never reach the host.
44 * @default ["OBJECT_FINALIZE"]
45 */
46 eventTypes?: BucketEventType[];
47 /**
48 * Only deliver events for objects whose names start with this prefix.
49 * Filtered by the bucket notification.
50 */
51 prefix?: string;
52}
53
54export type BucketEventsHandler<Req> = (
55 events: Stream.Stream<BucketEvent>,
56) => Effect.Effect<void, never, Req>;
57
58export type BucketEventSourceService = <Req = never>(
59 bucket: Bucket,
60 props: BucketEventSourceProps,
61 process: BucketEventsHandler<Req>,
62) => Effect.Effect<void, never, never>;
63
64/**
65 * Event source streaming a Cloud Storage {@link Bucket}'s object changes
66 * (created, deleted, archived, metadata updated) into the hosting compute.
67 *
68 * `GCP.Storage.BucketEventSourceLive` provisions a Pub/Sub topic owned by
69 * the host plus a `JSON_API_V1` bucket notification on it (filtered by
70 * event type and object prefix), then delegates delivery to whichever
71 * `GCP.PubSub.TopicEventSource` implementation is provided alongside it:
72 * `GCP.Run.TopicEventSource` (push, for `GCP.Function` /
73 * `GCP.CloudFunctions.Function`) or `GCP.Run.TopicPullEventSource` (pull,
74 * for `GCP.Run.Job` / `GCP.Run.WorkerPool`).
75 *
76 * Consume it through {@link consumeBucketEvents}.
77 *
78 * ### Consuming Bucket Events
79 * **Example:** Process uploads on a Cloud Run service
80 * ```typescript
81 * export class Thumbnails extends GCP.Function<Thumbnails>()(
82 * "Thumbnails",
83 * { main: import.meta.url },
84 * Effect.gen(function* () {
85 * const uploads = yield* GCP.Storage.Bucket("Uploads", {});
86 * yield* GCP.Storage.consumeBucketEvents(
87 * uploads,
88 * { prefix: "incoming/" },
89 * (events) =>
90 * events.pipe(
91 * Stream.runForEach((event) =>
92 * Effect.log(`${event.object} (${event.metadata.size} bytes)`),
93 * ),
94 * ),
95 * );
96 * }).pipe(
97 * Effect.provide(GCP.Storage.BucketEventSourceLive),
98 * Effect.provide(GCP.Run.TopicEventSource),
99 * ),
100 * ) {}
101 * ```
102 *
103 * **Example:** Deletions on a worker pool
104 * ```typescript
105 * yield* GCP.Storage.consumeBucketEvents(
106 * uploads,
107 * { eventTypes: ["OBJECT_DELETE", "OBJECT_ARCHIVE"] },
108 * (events) => events.pipe(Stream.runForEach(handle)),
109 * );
110 * // …provided with GCP.Storage.BucketEventSourceLive
111 * // and GCP.Run.TopicPullEventSource
112 * ```
113 *
114 * @binding
115 * @category Storage
116 */
117export interface BucketEventSource extends Binding.Service<
118 BucketEventSource,
119 "GCP.Storage.BucketEventSource",
120 BucketEventSourceService
121> {}
122
123export const BucketEventSource = Binding.Service<BucketEventSource>(
124 "GCP.Storage.BucketEventSource",
125);
126
127/**
128 * Subscribe an Effect handler to object changes in a Cloud Storage
129 * {@link Bucket}. See {@link BucketEventSource} for the implementations.
130 */
131export function consumeBucketEvents<Req = never, ResourceReq = never>(
132 bucket: Bucket | Effect.Effect<Bucket, never, ResourceReq>,
133 process: BucketEventsHandler<Req>,
134): Effect.Effect<void, never, BucketEventSource | ResourceReq>;
135export function consumeBucketEvents<Req = never, ResourceReq = never>(
136 bucket: Bucket | Effect.Effect<Bucket, never, ResourceReq>,
137 props: BucketEventSourceProps,
138 process: BucketEventsHandler<Req>,
139): Effect.Effect<void, never, BucketEventSource | ResourceReq>;
140export function consumeBucketEvents<Req = never, ResourceReq = never>(
141 bucket: Bucket | Effect.Effect<Bucket, never, ResourceReq>,
142 propsOrProcess: BucketEventSourceProps | BucketEventsHandler<Req>,
143 maybeProcess?: BucketEventsHandler<Req>,
144): Effect.Effect<void, never, BucketEventSource | ResourceReq> {
145 const [props, process] =
146 typeof propsOrProcess === "function"
147 ? [{} as BucketEventSourceProps, propsOrProcess]
148 : [propsOrProcess, maybeProcess!];
149 // Accept the resource or the Effect that declares it, like bindings do.
150 const resolved = Effect.isEffect(bucket) ? bucket : Effect.succeed(bucket);
151 return resolved.pipe(
152 Effect.flatMap((value) =>
153 BucketEventSource.use((source) => source(value, props, process)),
154 ),
155 );
156}
157
158const decodeBase64Utf8 = (data: string) => {
159 const binary = atob(data);
160 const bytes = new Uint8Array(binary.length);
161 for (let index = 0; index < binary.length; index++) {
162 bytes[index] = binary.charCodeAt(index);
163 }
164 return new TextDecoder().decode(bytes);
165};
166
167/**
168 * Parse a Cloud Storage Pub/Sub notification into a {@link BucketEvent}.
169 * The routing fields come from the message attributes; `metadata` is the
170 * `JSON_API_V1` object resource in `data` (empty when the notification
171 * uses `NONE` or the payload is unreadable).
172 */
173export const toBucketEvent = ({ message }: TopicMessage) =>
174 Effect.sync((): BucketEvent => {
175 const attributes = message.attributes ?? {};
176 let metadata: storage.Storage_Object = {};
177 try {
178 if (message.data) {
179 metadata = JSON.parse(
180 decodeBase64Utf8(message.data),
181 ) as storage.Storage_Object;
182 }
183 } catch {
184 metadata = {};
185 }
186 return {
187 eventType: (attributes.eventType ?? "OBJECT_FINALIZE") as BucketEventType,
188 bucket: attributes.bucketId ?? metadata.bucket ?? "",
189 object: attributes.objectId ?? metadata.name ?? "",
190 generation:
191 attributes.objectGeneration ?? String(metadata.generation ?? ""),
192 eventTime: attributes.eventTime ?? message.publishTime ?? "",
193 metadata,
194 };
195 });