| 1 | import * as GCP from "alchemy/GCP"; |
| 2 | import * as Effect from "effect/Effect"; |
| 3 | import * as Result from "effect/Result"; |
| 4 | import * as Stream from "effect/Stream"; |
| 5 | import { createHash } from "node:crypto"; |
| 6 | import { |
| 7 | Jobs, |
| 8 | Results, |
| 9 | resultPath, |
| 10 | type JobMessage, |
| 11 | type JobResult, |
| 12 | } from "./resources.ts"; |
| 13 | |
| 14 | const decode = (data: string | undefined) => |
| 15 | Effect.try( |
| 16 | () => |
| 17 | JSON.parse( |
| 18 | Buffer.from(data ?? "", "base64").toString("utf8"), |
| 19 | ) as JobMessage, |
| 20 | ); |
| 21 | |
| 22 | /** The "work": hash and count the payload. */ |
| 23 | const runJob = (job: JobMessage) => |
| 24 | Effect.sync((): JobResult => ({ |
| 25 | status: "done", |
| 26 | sha256: createHash("sha256").update(job.payload, "utf8").digest("hex"), |
| 27 | words: job.payload.split(/\s+/).filter(Boolean).length, |
| 28 | chars: job.payload.length, |
| 29 | submittedAt: job.submittedAt, |
| 30 | completedAt: new Date().toISOString(), |
| 31 | })); |
| 32 | |
| 33 | /** |
| 34 | * The consumer. A Cloud Run worker pool has no inbound URL — it runs a |
| 35 | * long-lived container that pulls from Pub/Sub, which suits queue workers |
| 36 | * that should keep running regardless of HTTP traffic. |
| 37 | * |
| 38 | * `GCP.Run.TopicPullEventSource` creates the pull subscription, pulls |
| 39 | * batches in the background, and acks a batch only after the handler |
| 40 | * succeeds. Writing the result with `set` makes processing idempotent: a |
| 41 | * redelivered job overwrites its document with the same values. |
| 42 | */ |
| 43 | export default class Worker extends GCP.Run.WorkerPool<Worker>()( |
| 44 | "Worker", |
| 45 | { |
| 46 | main: import.meta.url, |
| 47 | scaling: { manualInstanceCount: 1 }, |
| 48 | }, |
| 49 | Effect.gen(function* () { |
| 50 | const results = yield* GCP.Firestore.WriteDatabase(Results); |
| 51 | |
| 52 | yield* GCP.PubSub.consumeTopicMessages( |
| 53 | Jobs, |
| 54 | { maxMessages: 10, ackDeadlineSeconds: 60 }, |
| 55 | (messages) => |
| 56 | messages.pipe( |
| 57 | Stream.runForEach(({ message }) => |
| 58 | Effect.gen(function* () { |
| 59 | const decoded = yield* Effect.result(decode(message.data)); |
| 60 | // A message that can never parse would be redelivered |
| 61 | // forever; drop it (the batch ack removes it) instead. |
| 62 | if (Result.isFailure(decoded)) { |
| 63 | yield* Effect.logWarning( |
| 64 | `dropping malformed message ${message.messageId}`, |
| 65 | ); |
| 66 | return; |
| 67 | } |
| 68 | const job = decoded.success; |
| 69 | const result = yield* runJob(job); |
| 70 | yield* results.set(resultPath(job.id), { ...result }); |
| 71 | yield* Effect.log(`job ${job.id}: ${result.words} word(s)`); |
| 72 | }), |
| 73 | ), |
| 74 | // A failure dies the batch, leaving it unacked so Pub/Sub |
| 75 | // redelivers it after the ack deadline. |
| 76 | Effect.orDie, |
| 77 | ), |
| 78 | ); |
| 79 | }).pipe( |
| 80 | Effect.provide([ |
| 81 | GCP.Run.TopicPullEventSource, |
| 82 | GCP.Firestore.WriteDatabaseHttp, |
| 83 | ]), |
| 84 | ), |
| 85 | ) {} |