From be602dfc6448e8df7e160d5b1e2fe7e46ce1426b Mon Sep 17 00:00:00 2001 From: Sivert Date: Mon, 5 Oct 2026 15:33:01 +0200 Subject: [PATCH] Every chat picture and video goes through the worker, and readers see a loader until it has (GRYT-1669) Co-Authored-By: Claude Opus 5.5 --- src/db/interfaces.ts | 4 +++ src/db/sqlite/messages.ts | 8 +++++ src/routes/uploadQuarantine.test.ts | 51 ++++++++++++++++++++++++++++ src/routes/uploadStorage.test.ts | 3 +- src/routes/uploads.ts | 2 +- src/services/chatAttachmentSettle.ts | 34 +++++++++++++++++++ src/services/storeUploadedFile.ts | 32 +++++++++++++---- src/services/workerCapabilities.ts | 9 +++++ src/socket/handlers/chat.ts | 11 +++++- src/utils/sniffMedia.test.ts | 28 +++++++++++++++ src/utils/sniffMedia.ts | 19 +++++++++++ 11 files changed, 191 insertions(+), 10 deletions(-) create mode 100644 src/services/chatAttachmentSettle.ts create mode 100644 src/utils/sniffMedia.test.ts create mode 100644 src/utils/sniffMedia.ts diff --git a/src/db/interfaces.ts b/src/db/interfaces.ts index af4be62b..3192506d 100644 --- a/src/db/interfaces.ts +++ b/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 { diff --git a/src/db/sqlite/messages.ts b/src/db/sqlite/messages.ts index 38b3541b..52485263 100644 --- a/src/db/sqlite/messages.ts +++ b/src/db/sqlite/messages.ts @@ -196,6 +196,14 @@ export async function getAllFileRecords(): Promise { })); } +/** The messages that carry a file, for telling their readers it has finished processing. */ +export async function listMessagesWithFile(fileId: string): Promise> { + 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> { const db = getSqliteDb(); const ids = new Set(); diff --git a/src/routes/uploadQuarantine.test.ts b/src/routes/uploadQuarantine.test.ts index fa6e3193..77d6c243 100644 --- a/src/routes/uploadQuarantine.test.ts +++ b/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; + 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; + }; + + 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]); + }); }); diff --git a/src/routes/uploadStorage.test.ts b/src/routes/uploadStorage.test.ts index b1930a54..18194618 100644 --- a/src/routes/uploadStorage.test.ts +++ b/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 \}\)/); }); }); diff --git a/src/routes/uploads.ts b/src/routes/uploads.ts index 98235cd8..4ebfb489 100644 --- a/src/routes/uploads.ts +++ b/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. diff --git a/src/services/chatAttachmentSettle.ts b/src/services/chatAttachmentSettle.ts new file mode 100644 index 00000000..cf01d23e --- /dev/null +++ b/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 { + 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); + } +} diff --git a/src/services/storeUploadedFile.ts b/src/services/storeUploadedFile.ts index f7f188f7..78033ee1 100644 --- a/src/services/storeUploadedFile.ts +++ b/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 { + 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 }; } diff --git a/src/services/workerCapabilities.ts b/src/services/workerCapabilities.ts index 31d33d00..96b646fd 100644 --- a/src/services/workerCapabilities.ts +++ b/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 { 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); } } diff --git a/src/socket/handlers/chat.ts b/src/socket/handlers/chat.ts index 16610518..b30e7bee 100644 --- a/src/socket/handlers/chat.ts +++ b/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 { +export async function enrichAttachments(messages: MessageRecord[]): Promise { const allFileIds = new Set(); 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(); + 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 { 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, diff --git a/src/utils/sniffMedia.test.ts b/src/utils/sniffMedia.test.ts new file mode 100644 index 00000000..07310895 --- /dev/null +++ b/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) => + 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(" [...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; +}