Every chat picture and video goes through the worker, and readers see a loader until it has (GRYT-1669)

src/db/interfaces.ts

@@ -191,6 +191,10 @@ export interface EnrichedAttachment {
   width: number | null;
   height: number | null;
   has_thumbnail: boolean;
+  /** Still with the image worker: the client shows a loader instead of fetching it (GRYT-1669). */
+  processing?: boolean;
+  /** The worker could not write it out, so it is never served. */
+  refused?: boolean;
 }
 
 export interface FileRecord {

src/db/sqlite/messages.ts

@@ -196,6 +196,14 @@ export async function getAllFileRecords(): Promise<FileRecord[]> {
   }));
 }
 
+/** The messages that carry a file, for telling their readers it has finished processing. */
+export async function listMessagesWithFile(fileId: string): Promise<Array<{ conversation_id: string; message_id: string }>> {
+  const db = getSqliteDb();
+  return db
+    .prepare(`SELECT conversation_id, message_id FROM message_attachments WHERE file_id = ?`)
+    .all(fileId) as Array<{ conversation_id: string; message_id: string }>;
+}
+
 export async function getAllReferencedAttachmentIds(): Promise<Set<string>> {
   const db = getSqliteDb();
   const ids = new Set<string>();

src/routes/uploadQuarantine.test.ts

@@ -199,4 +199,55 @@ describe("uploads with a worker that clears quarantine", () => {
     assert.equal(row?.bannerVideo, true, "the member list says it plays");
     assert.equal(row?.avatarVideo, false);
   });
+
+  it("sends a chat picture labelled as anything else to the worker, by its bytes", async () => {
+    await workerSays(["quarantine-v1"]);
+    const gina = await member("gina");
+    const form = new FormData();
+    form.append("file", new Blob([new Uint8Array(await png())], { type: "application/octet-stream" }), "picture.bin");
+    const res = await fetch(`${base}/api/uploads`, { method: "POST", headers: auth(gina), body: form });
+    const body = (await res.json()) as Record<string, unknown>;
+    assert.equal(res.status, 201, JSON.stringify(body));
+    assert.equal(body.processing, true);
+    const file = await getFile(body.fileId as string);
+    assert.ok(file?.s3_key.startsWith("quarantine/uploads/"), file?.s3_key);
+    assert.equal(file?.mime, "image/png");
+  });
+
+  it("quarantines a chat video only where the worker transcodes chat videos", async () => {
+    const hal = await member("hal");
+    const mp4 = Buffer.concat([Buffer.from([0, 0, 0, 0x18]), Buffer.from("ftypisom"), Buffer.alloc(16)]);
+    const sendVideo = async () => {
+      const form = new FormData();
+      form.append("file", new Blob([new Uint8Array(mp4)], { type: "video/mp4" }), "clip.mp4");
+      const res = await fetch(`${base}/api/uploads`, { method: "POST", headers: auth(hal), body: form });
+      return (await res.json()) as Record<string, unknown>;
+    };
+
+    await workerSays(["quarantine-v1", "video-v1"]);
+    const raw = await sendVideo();
+    assert.equal(raw.processing, false);
+    assert.ok((await getFile(raw.fileId as string))?.s3_key.startsWith("uploads/"));
+
+    await workerSays(["quarantine-v1", "video-v1", "chatvideo-v1"]);
+    const held = await sendVideo();
+    assert.equal(held.processing, true);
+    assert.ok((await getFile(held.fileId as string))?.s3_key.startsWith("quarantine/uploads/"));
+  });
+
+  it("marks a quarantined attachment as processing, then refused once the worker gives up", async () => {
+    await workerSays(["quarantine-v1"]);
+    const ivy = await member("ivy");
+    const form = new FormData();
+    form.append("file", new Blob([new Uint8Array(await png())], { type: "image/png" }), "picture.png");
+    const fileId = ((await (await fetch(`${base}/api/uploads`, { method: "POST", headers: auth(ivy), body: form })).json()) as { fileId: string }).fileId;
+    // eslint-disable-next-line @typescript-eslint/no-require-imports
+    const { enrichAttachments } = require("../socket/handlers/chat") as typeof import("../socket/handlers/chat");
+    const message = { conversation_id: "general", message_id: "m1", sender_server_id: ivy.serverUserId, text: "look", created_at: new Date(), attachments: [fileId], reactions: null };
+    const first = (await enrichAttachments([message]))[0].enriched_attachments?.[0];
+    assert.deepEqual([first?.processing, first?.refused], [true, false]);
+    workerRefuses(fileId);
+    const after = (await enrichAttachments([message]))[0].enriched_attachments?.[0];
+    assert.deepEqual([after?.processing, after?.refused], [false, true]);
+  });
 });

src/routes/uploadStorage.test.ts

@@ -189,6 +189,7 @@ describe("the route uses it", () => {
   });
 
   it("answers the uploader the same whether or not a poster ever comes", () => {
-    assert.match(route, /res\.status\(201\)\.json\(\{ fileId: stored\.fileId, key: stored\.key, thumbnailKey: null \}\)/);
+    // `processing` says it is in quarantine, which is known at upload; nothing waits on a poster.
+    assert.match(route, /res\.status\(201\)\.json\(\{ fileId: stored\.fileId, key: stored\.key, thumbnailKey: null, processing: !!stored\.processing \}\)/);
   });
 });

src/routes/uploads.ts

@@ -179,7 +179,7 @@ uploadsRouter.post(
           res.status(stored.status).json({ error: stored.error, message: stored.message });
           return;
         }
-        res.status(201).json({ fileId: stored.fileId, key: stored.key, thumbnailKey: null });
+        res.status(201).json({ fileId: stored.fileId, key: stored.key, thumbnailKey: null, processing: !!stored.processing });
       })
       // Every exit path, including the early returns and anything that threw:
       // multer's temp file is ours and nothing else removes it.

src/services/chatAttachmentSettle.ts

@@ -0,0 +1,34 @@
+import consola from "consola";
+
+import { getMessageById, getUserByServerId, listMessagesWithFile } from "../db";
+import { pluginRefs } from "../plugins/refs";
+import { enrichAttachments } from "../socket/handlers/chat";
+import { recipientClientIds } from "../socket/utils/recipients";
+import { resolveConversationAccess } from "../socket/utils/conversationAccess";
+import { settleQuarantine } from "./quarantineUpload";
+
+/* A chat attachment shows a loader until the worker has written it out. Once it has, or has
+   refused it, each message carrying it is sent again to the people who can read it (GRYT-1669). */
+export async function settleChatAttachment(fileId: string): Promise<void> {
+  try {
+    if ((await settleQuarantine(fileId)) === "timeout") return;
+    // The socket layer's refs, which carry the SFU client that working out a channel's readers needs.
+    const refs = pluginRefs();
+    if (!refs) return;
+    for (const { conversation_id, message_id } of await listMessagesWithFile(fileId)) {
+      const message = await getMessageById(conversation_id, message_id);
+      if (!message) continue;
+      const access = await resolveConversationAccess(conversation_id, message.sender_server_id);
+      if (!access.allowed) continue;
+      const sender = await getUserByServerId(message.sender_server_id);
+      const [enriched] = await enrichAttachments([{ ...message, sender_nickname: sender?.nickname ?? message.sender_nickname }]);
+      if (!enriched?.enriched_attachments) continue;
+      const payload = { conversation_id, message_id, enriched_attachments: enriched.enriched_attachments };
+      for (const cid of await recipientClientIds(conversation_id, access, refs.clientsInfo, refs.sfuClient)) {
+        refs.io.sockets.sockets.get(cid)?.emit("chat:attachments", payload);
+      }
+    }
+  } catch (err) {
+    consola.warn(`[uploads] Could not settle attachment ${fileId}`, err);
+  }
+}

src/services/storeUploadedFile.ts

@@ -1,12 +1,14 @@
 import consola from "consola";
-import { readFile } from "fs/promises";
+import { open, readFile } from "fs/promises";
 import { v4 as uuidv4 } from "uuid";
 
 import { insertFile, insertImageJob } from "../db";
 import { storageForUpload } from "../routes/uploadStorage";
 import { putObject } from "../storage";
 import { QUARANTINE_PREFIX } from "./quarantineUpload";
-import { workerClearsQuarantine } from "./workerCapabilities";
+import { workerClearsQuarantine, workerTranscodesChatVideo } from "./workerCapabilities";
+import { settleChatAttachment } from "./chatAttachmentSettle";
+import { sniffMedia } from "../utils/sniffMedia";
 import { validateImage } from "../utils/imageValidation";
 import { sanitizeSvg } from "../utils/svgSanitize";
 import { PARSE_LIMITS, readVideoDimensionsFromFile } from "../utils/videoDimensions";
@@ -16,9 +18,20 @@ import { PARSE_LIMITS, readVideoDimensionsFromFile } from "../utils/videoDimensi
 export const IMAGE_VALIDATION_MAX_BYTES = 64 * 1024 * 1024;
 
 export type StoreUploadedFileResult =
-  | { ok: true; fileId: string; key: string }
+  | { ok: true; fileId: string; key: string; processing?: boolean }
   | { ok: false; status: 400 | 413; error: "invalid_file" | "file_too_large"; message: string };
 
+async function readHead(path: string): Promise<Uint8Array> {
+  const handle = await open(path, "r");
+  try {
+    const head = Buffer.alloc(16);
+    const { bytesRead } = await handle.read(head, 0, 16, 0);
+    return head.subarray(0, bytesRead);
+  } finally {
+    await handle.close();
+  }
+}
+
 function positiveInt(n: number | null | undefined): number | null {
   return typeof n === "number" && Number.isFinite(n) && n > 0 ? Math.round(n) : null;
 }
@@ -45,12 +58,15 @@ export async function storeUploadedFile(input: {
   const fileId = input.fileId ?? uuidv4();
   const { bucket, path, size } = input;
 
+  // The bytes over the label: a picture or video sent as anything else still goes to the worker.
+  const sniffed = input.sealed ? null : sniffMedia(await readHead(path));
+
   // In its own file because nothing in here loads in a test, and because
   // sealing is the part with a security answer. See `uploadStorage.ts`.
   const storage = storageForUpload({
     sealed: input.sealed,
     fileId,
-    mimetype: input.mimetype,
+    mimetype: sniffed ?? input.mimetype,
     originalName: input.originalName,
   });
 
@@ -66,8 +82,9 @@ export async function storeUploadedFile(input: {
   }
 
   const { storedMime } = storage;
-  // A plain image goes to quarantine when the worker clears it; videos keep today's path for now.
-  const quarantined = storage.queueImageJob && storage.validateAsImage && workerClearsQuarantine();
+  // Pictures and videos both go to quarantine when the worker can write them out again (GRYT-1669).
+  const quarantined = storage.queueImageJob && workerClearsQuarantine()
+    && (storage.validateAsImage || (storage.measureAsVideo && workerTranscodesChatVideo()));
   const key = quarantined ? `${QUARANTINE_PREFIX}${storage.key}` : storage.key;
   let width: number | null = null;
   let height: number | null = null;
@@ -167,5 +184,6 @@ export async function storeUploadedFile(input: {
     }).catch((e: unknown) => consola.warn("Failed to queue image job", e));
   }
 
-  return { ok: true, fileId, key };
+  if (quarantined) void settleChatAttachment(fileId);
+  return { ok: true, fileId, key, processing: quarantined };
 }

src/services/workerCapabilities.ts

@@ -6,6 +6,7 @@ const REFRESH_MS = 60_000;
 
 let clears = false;
 let video = false;
+let chatVideo = false;
 let timer: NodeJS.Timeout | null = null;
 
 export function workerClearsQuarantine(): boolean {
@@ -17,9 +18,15 @@ export function workerTranscodesVideo(): boolean {
   return video;
 }
 
+/** The worker transcodes chat videos too, keeping their sound (GRYT-1669). */
+export function workerTranscodesChatVideo(): boolean {
+  return chatVideo;
+}
+
 export async function refreshWorkerCapabilities(url = process.env.IMAGE_WORKER_URL, fetchImpl: typeof fetch = fetch): Promise<boolean> {
   if (!url) {
     video = false;
+    chatVideo = false;
     return (clears = false);
   }
   try {
@@ -28,10 +35,12 @@ export async function refreshWorkerCapabilities(url = process.env.IMAGE_WORKER_U
     const caps: unknown[] = Array.isArray(body?.capabilities) ? body.capabilities : [];
     const next = caps.includes("quarantine-v1");
     video = next && caps.includes("video-v1");
+    chatVideo = next && caps.includes("chatvideo-v1");
     if (next !== clears) consola.info(`[uploads] Quarantine through the image worker ${next ? "on" : "off"}`);
     return (clears = next);
   } catch {
     video = false;
+    chatVideo = false;
     return (clears = false);
   }
 }

src/socket/handlers/chat.ts

@@ -21,6 +21,7 @@ import {
   getMessageById,
   updateMessageText,
   getFilesByIds,
+  getImageJobStatusForFile,
   getServerConfig,
   getWebhooksByIds,
   getConversation,
@@ -45,6 +46,7 @@ import {
   setThreadTags,
   type ThreadRecord,
 } from "../../db";
+import { isQuarantined } from "../../services/quarantineUpload";
 import { processProfanity, type CensorStyle, type ProfanityMode } from "../../utils/profanityFilter";
 import { checkRateLimit, RateLimitRule } from "../../utils/rateLimiter";
 import { textMuteError, textMuteFor } from "../../moderation/textMute";
@@ -205,13 +207,18 @@ async function enrichMessages(messages: MessageRecord[]): Promise<MessageRecord[
   });
 }
 
-async function enrichAttachments(messages: MessageRecord[]): Promise<MessageRecord[]> {
+export async function enrichAttachments(messages: MessageRecord[]): Promise<MessageRecord[]> {
   const allFileIds = new Set<string>();
   for (const m of messages) {
     if (m.attachments) m.attachments.forEach(id => allFileIds.add(id));
   }
   if (allFileIds.size === 0) return messages;
   const fileMap = await getFilesByIds([...allFileIds]);
+  // Only files still in quarantine need their job's status, which is usually none of them.
+  const refusedIds = new Set<string>();
+  for (const f of fileMap.values()) {
+    if (isQuarantined(f.s3_key) && (await getImageJobStatusForFile(f.file_id)) === "error") refusedIds.add(f.file_id);
+  }
 
   const result: MessageRecord[] = [];
   for (const m of messages) {
@@ -220,7 +227,9 @@ async function enrichAttachments(messages: MessageRecord[]): Promise<MessageReco
     const enriched = m.attachments.map(id => {
       const f = fileMap.get(id);
       if (!f) return { file_id: id, mime: null, size: null, original_name: null, width: null, height: null, has_thumbnail: false };
+      const inQuarantine = isQuarantined(f.s3_key);
       return {
+        ...(inQuarantine ? { processing: !refusedIds.has(f.file_id), refused: refusedIds.has(f.file_id) } : {}),
         file_id: f.file_id,
         mime: f.mime,
         size: f.size,

src/utils/sniffMedia.test.ts

@@ -0,0 +1,28 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+
+import { sniffMedia } from "./sniffMedia";
+
+const b = (...parts: Array<string | number[]>) =>
+  new Uint8Array(parts.flatMap((p) => (typeof p === "string" ? [...p].map((c) => c.charCodeAt(0)) : p)));
+
+describe("sniffMedia", () => {
+  it("knows each format the worker reads by its first bytes", () => {
+    assert.equal(sniffMedia(b([0xff, 0xd8, 0xff, 0xe0])), "image/jpeg");
+    assert.equal(sniffMedia(b([0x89], "PNG\r\n\x1a\n")), "image/png");
+    assert.equal(sniffMedia(b("GIF89a")), "image/gif");
+    assert.equal(sniffMedia(b("RIFF", [0, 0, 0, 0], "WEBP")), "image/webp");
+    assert.equal(sniffMedia(b([0, 0, 0, 0x1c], "ftypavif")), "image/avif");
+    assert.equal(sniffMedia(b([0, 0, 0, 0x18], "ftypisom")), "video/mp4");
+    assert.equal(sniffMedia(b([0, 0, 0, 0x14], "ftypqt  ")), "video/quicktime");
+    assert.equal(sniffMedia(b([0x1a, 0x45, 0xdf, 0xa3])), "video/webm");
+  });
+
+  it("says nothing about everything else, so it stays a download", () => {
+    assert.equal(sniffMedia(b("PK", [3, 4])), null);
+    assert.equal(sniffMedia(b("%PDF-1.7")), null);
+    assert.equal(sniffMedia(b("<svg")), null);
+    assert.equal(sniffMedia(b("#EXTM3U\n")), null, "an HLS playlist named .mp4 is not a video");
+    assert.equal(sniffMedia(new Uint8Array()), null);
+  });
+});

src/utils/sniffMedia.ts

@@ -0,0 +1,19 @@
+/* What the first bytes of an upload say it is, for the formats the worker reads. The label a
+   client sends is a claim; a picture sent as application/octet-stream still goes to the worker. */
+
+const ascii = (b: Uint8Array, at: number, s: string) => [...s].every((c, i) => b[at + i] === c.charCodeAt(0));
+
+export function sniffMedia(b: Uint8Array): string | null {
+  if (b.length >= 3 && b[0] === 0xff && b[1] === 0xd8 && b[2] === 0xff) return "image/jpeg";
+  if (b.length >= 8 && b[0] === 0x89 && ascii(b, 1, "PNG\r\n\x1a\n")) return "image/png";
+  if (ascii(b, 0, "GIF87a") || ascii(b, 0, "GIF89a")) return "image/gif";
+  if (b.length >= 12 && ascii(b, 0, "RIFF") && ascii(b, 8, "WEBP")) return "image/webp";
+  if (b.length >= 12 && ascii(b, 4, "ftyp")) {
+    const brand = String.fromCharCode(...b.subarray(8, 12));
+    if (brand === "avif" || brand === "avis") return "image/avif";
+    if (brand === "heic" || brand === "heix" || brand === "mif1") return null;
+    return brand === "qt  " ? "video/quicktime" : "video/mp4";
+  }
+  if (b.length >= 4 && b[0] === 0x1a && b[1] === 0x45 && b[2] === 0xdf && b[3] === 0xa3) return "video/webm";
+  return null;
+}