workers/media-events.ts

Functions

Imports

It imports app/db/index.ts, app/lib/error-message.mjs, app/lib/media/backup.server.ts, app/lib/media/classify.mjs, app/lib/media/core.server.ts. Imported by workers/app.ts.

Source

1import { deleteMediaRecord, upsertDerivedMedia } from "~/db";
2import { bucketFor, classify, isRaster, roleOf, storageOf } from "~/lib/media/classify.mjs";
3import { mirrorObject } from "~/lib/media/backup.server";
4import { measureDimensions, placeholderFor } from "~/lib/media/core.server";
5import { errorMessage } from "~/lib/error-message.mjs";
6
7/**
8 * The authoritative media row writer, fed by R2 event notifications rather than a dual write. Must be
9 * idempotent: Queues delivers at least once and out of order, so the object is re-read, never trusted
10 * from the event, and R2 wins any disagreement.
11 */
12
13function objectKeyOf(body: unknown): string | null {
14  if (typeof body !== "object" || body === null) return null;
15  const object = (body as { object?: unknown }).object;
16  if (typeof object !== "object" || object === null) return null;
17  const key = (object as { key?: unknown }).key;
18  return typeof key === "string" && key.length > 0 ? key : null;
19}
20
21// A missing object deletes the row, so reading the wrong bucket for a key deletes it too. Use the
22// shared `bucketFor`; never write a second copy.
23
24export async function handleMediaEvents(batch: MessageBatch<unknown>, env: Env) {
25  for (const message of batch.messages) {
26    const key = objectKeyOf(message.body);
27    if (!key) {
28      console.error(JSON.stringify({ alert: "media-event-no-object-key", body: message.body }));
29      message.ack();
30      continue;
31    }
32
33    try {
34      await indexOne(env, key);
35      message.ack();
36    } catch (error) {
37      // Never ack a transient failure, or the row silently never appears.
38      console.error(JSON.stringify({ alert: "media-event-failed", key, detail: errorMessage(error) }));
39      message.retry();
40    }
41  }
42}
43
44/** Ignores the event's action on purpose: asking R2 what is true now makes reordering safe. */
45async function indexOne(env: Env, key: string) {
46  const object = await bucketFor(env, key).get(key);
47
48  if (!object) {
49    // Logged, because a genuine delete and a key looked up in the wrong bucket both land here.
50    console.log(
51      JSON.stringify({
52        event: "media-event-no-object",
53        key,
54        bucket: storageOf(key) === "r2-derived" ? "OG" : "MEDIA",
55        action: "deleting any row",
56      }),
57    );
58    await deleteMediaRecord(env, key);
59    return;
60  }
61
62  const { kind, mime } = classify(key);
63
64  let width: number | null = null;
65  let height: number | null = null;
66  let placeholder: string | null = null;
67  // An unreadable image still gets a row, or the index would disagree with the bucket; the stored
68  // sizes are left alone and the failure is thrown after the write, so the message retries and a
69  // permanent failure lands in the dead-letter queue.
70  let measureFailure: unknown = null;
71  if (isRaster(key)) {
72    try {
73      const dimensions = await measureDimensions(env, object.body);
74      width = dimensions?.width ?? null;
75      height = dimensions?.height ?? null;
76    } catch (error) {
77      measureFailure = error;
78    }
79
80    // Derived here, or the upsert erases the bulk pass's placeholder. A second GET, because `.info()`
81    // consumed the first body stream.
82    const reread = await bucketFor(env, key).get(key);
83    if (reread) placeholder = await placeholderFor(env, reread.body);
84  }
85
86  await upsertDerivedMedia(env, {
87    key,
88    storage: storageOf(key),
89    kind,
90    role: roleOf(key),
91    mime,
92    bytes: object.size,
93    width,
94    height,
95    // From the object: a content-addressed key cannot yield a filename. Null means "do not touch".
96    originalName: object.customMetadata?.originalName ?? null,
97    // Null means "do not touch", not "erase".
98    placeholder,
99    uploadedAt:
100      object.uploaded instanceof Date ? object.uploaded.toISOString() : String(object.uploaded ?? ""),
101  });
102
103  // Nothing here deletes a backup twin: the backup is the copy that survives a delete. OG cards are
104  // regenerable, so not mirrored. A failed mirror is swallowed; the health poll sees the drift.
105  if (storageOf(key) === "r2") {
106    try {
107      await mirrorObject(env, key);
108    } catch (error) {
109      // The row stands; media-backup-drift will report it and backup_media repairs it.
110      console.log(JSON.stringify({ event: "media-mirror-failed", key, detail: errorMessage(error) }));
111    }
112  }
113
114  // Last, so the row and the mirror both happened before the retry is asked for.
115  if (measureFailure) throw measureFailure;
116}