1import * as GCP from "alchemy/GCP";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import {
6 documentFor,
7 Files,
8 UPLOADS_PREFIX,
9 Uploads,
10 type FileRecord,
11} from "./resources.ts";
12
13const sha256Hex = (bytes: Uint8Array) =>
14 Effect.promise(() =>
15 crypto.subtle.digest("SHA-256", bytes as Uint8Array<ArrayBuffer>),
16 ).pipe(
17 Effect.map((digest) =>
18 Array.from(new Uint8Array(digest), (byte) =>
19 byte.toString(16).padStart(2, "0"),
20 ).join(""),
21 ),
22 );
23
24/**
25 * The background half of the pipeline: index every finished upload.
26 *
27 * Cloud Storage publishes an `OBJECT_FINALIZE` notification for each
28 * object written under `uploads/`; Pub/Sub pushes it to this service,
29 * which reads the object back, hashes it, and records the result in
30 * Firestore. Nothing here runs on the upload's request path, so a slow
31 * hash of a large file never delays the client.
32 *
33 * The service stays private: Pub/Sub authenticates each push with an
34 * OIDC token for the service's own identity.
35 */
36export default class Indexer extends GCP.Function<Indexer>()(
37 "Indexer",
38 {
39 main: import.meta.url,
40 },
41 Effect.gen(function* () {
42 const bucket = yield* Uploads;
43 const database = yield* Files;
44
45 const uploads = yield* GCP.Storage.ReadBucket(bucket);
46 const files = yield* GCP.Firestore.WriteDatabase(database);
47
48 yield* GCP.Storage.consumeBucketEvents(
49 bucket,
50 { eventTypes: ["OBJECT_FINALIZE"], prefix: UPLOADS_PREFIX },
51 (events) =>
52 events.pipe(
53 Stream.runForEach((event) =>
54 Effect.gen(function* () {
55 // Read the exact generation the event describes. If it has
56 // since been overwritten or deleted, a later event (or the
57 // delete) owns the record, so there is nothing to do.
58 const object = yield* uploads.get(event.object, {
59 generation: event.generation,
60 });
61 if (object === undefined) {
62 yield* Effect.log(`indexer: ${event.object} is gone, skipping`);
63 return;
64 }
65
66 const name = event.object.slice(UPLOADS_PREFIX.length);
67 const record: FileRecord = {
68 name,
69 object: event.object,
70 generation: event.generation,
71 size: object.body.byteLength,
72 contentType:
73 object.contentType ??
74 event.metadata.contentType ??
75 "application/octet-stream",
76 sha256: yield* sha256Hex(object.body),
77 indexedAt: new Date(),
78 };
79
80 yield* files.set(documentFor(name), { ...record });
81 yield* Effect.log(`indexer: ${name} (${record.size} bytes)`);
82 }).pipe(
83 // A failure fails the push, so Pub/Sub redelivers the event.
84 Effect.orDie,
85 ),
86 ),
87 ),
88 );
89
90 return {
91 fetch: Effect.succeed(HttpServerResponse.text("ok")),
92 };
93 }).pipe(
94 Effect.provide([
95 GCP.Storage.BucketEventSourceLive,
96 GCP.Storage.ReadBucketHttp,
97 GCP.Firestore.WriteDatabaseHttp,
98 ]),
99 Effect.provide(GCP.Run.TopicEventSource),
100 ),
101) {}