1import * as GCP from "@/GCP";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import {
6 docId,
7 Jobs,
8 SCHEDULE_BODY,
9 SCHEDULE_ID,
10 Store,
11 Uploads,
12 type JobMessage,
13} from "./serverless-resources.ts";
14
15/**
16 * Private worker Function (no public invoker) with three event sources:
17 *
18 * - Pub/Sub push from {@link Jobs} → `results/{id}`
19 * - Cloud Scheduler job {@link SCHEDULE_ID} → `schedules/{jobName}`
20 * - Eventarc Storage finalize on {@link Uploads} → `uploads/{object}`
21 *
22 * Every delivery is authenticated with an OIDC token for the worker's own
23 * service account.
24 */
25export default class SmokeWorker extends GCP.Function<SmokeWorker>()(
26 "SmokeWorker",
27 { main: import.meta.url, location: "us-central1" },
28 Effect.gen(function* () {
29 const uploads = yield* Uploads;
30 const store = yield* GCP.Firestore.WriteDatabase(Store);
31
32 yield* GCP.PubSub.consumeTopicMessages(Jobs, (messages) =>
33 messages.pipe(
34 Stream.runForEach(({ message }) =>
35 Effect.gen(function* () {
36 const job = yield* Effect.try(
37 () =>
38 JSON.parse(
39 Buffer.from(message.data ?? "", "base64").toString("utf8"),
40 ) as JobMessage,
41 );
42 yield* store.set(`results/${job.id}`, {
43 id: job.id,
44 payload: job.payload,
45 processed: job.payload.toUpperCase(),
46 messageId: message.messageId,
47 });
48 }),
49 ),
50 Effect.orDie,
51 ),
52 );
53
54 yield* GCP.CloudScheduler.consumeSchedule(
55 SCHEDULE_ID,
56 { schedule: "0 0 1 1 *", timeZone: "Etc/UTC", body: SCHEDULE_BODY },
57 (event) =>
58 store
59 .set(`schedules/${docId(event.jobName)}`, {
60 jobName: event.jobName,
61 body: event.body ?? null,
62 scheduleTime: event.scheduleTime ?? null,
63 })
64 .pipe(Effect.asVoid, Effect.orDie),
65 );
66
67 yield* GCP.Eventarc.consumeEvents(
68 "SmokeUploads",
69 {
70 eventFilters: [
71 {
72 attribute: "type",
73 value: "google.cloud.storage.object.v1.finalized",
74 },
75 { attribute: "bucket", value: uploads.bucketName },
76 ],
77 },
78 (event) =>
79 Effect.gen(function* () {
80 const data = event.data as { name?: string; bucket?: string };
81 yield* store.set(`uploads/${docId(data.name ?? "unknown")}`, {
82 id: event.id,
83 type: event.type,
84 bucket: data.bucket ?? null,
85 name: data.name ?? null,
86 });
87 }).pipe(Effect.orDie),
88 );
89
90 return { fetch: Effect.succeed(HttpServerResponse.text("ok")) };
91 }).pipe(
92 Effect.provide(GCP.Firestore.WriteDatabaseHttp),
93 Effect.provide(GCP.Run.TopicEventSource),
94 Effect.provide(GCP.Run.ScheduleEventSource),
95 Effect.provide(GCP.Run.EventarcEventSource),
96 ),
97) {}