1import * as GCP from "@/GCP";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5
6export const INCOMING_PREFIX = "incoming/";
7
8/** Marker object the consumer writes for one delivered event. */
9export const markerFor = (object: string) =>
10 `markers/${object.replaceAll("/", "_")}.json`;
11
12/** Bucket whose `incoming/` uploads trigger the consumer. */
13export const Uploads = GCP.Storage.Bucket("Uploads", {
14 location: "US-CENTRAL1",
15 forceDestroy: true,
16});
17
18/**
19 * Effect-native Cloud Run service that consumes `incoming/` finalize events
20 * over Pub/Sub push and records each one as a marker object in the same
21 * bucket (outside the notification prefix, so markers don't re-trigger).
22 * Deployed from {@link ../BucketEventSource.test.ts}.
23 */
24export default class BucketEventsService extends GCP.Function<BucketEventsService>()(
25 "BucketEventsService",
26 {
27 main: import.meta.url,
28 location: "us-central1",
29 },
30 Effect.gen(function* () {
31 const bucket = yield* Uploads;
32 const putObject = yield* GCP.Storage.PutObject(bucket);
33
34 yield* GCP.Storage.consumeBucketEvents(
35 bucket,
36 { prefix: INCOMING_PREFIX },
37 (events) =>
38 events.pipe(
39 Stream.runForEach((event) =>
40 putObject({
41 name: markerFor(event.object),
42 body: JSON.stringify({
43 eventType: event.eventType,
44 bucket: event.bucket,
45 object: event.object,
46 generation: event.generation,
47 eventTime: event.eventTime,
48 size: event.metadata.size,
49 contentType: event.metadata.contentType,
50 }),
51 }).pipe(Effect.orDie),
52 ),
53 ),
54 );
55
56 return {
57 fetch: Effect.succeed(HttpServerResponse.text("ok")),
58 };
59 }).pipe(
60 Effect.provide(GCP.Storage.BucketEventSourceLive),
61 Effect.provide(GCP.Run.TopicEventSource),
62 Effect.provide(GCP.Storage.PutObjectHttp),
63 ),
64) {}