diff --git a/apps/api/src/group-files/group-files.controller.ts b/apps/api/src/group-files/group-files.controller.ts index c6c1637..84afe64 100644 --- a/apps/api/src/group-files/group-files.controller.ts +++ b/apps/api/src/group-files/group-files.controller.ts @@ -19,7 +19,7 @@ interface FileUpload { * Group files API — the Telegram-like group file surface. * * GET /files list/search, keyset (groupId, senderMxid, - * mimeClass, name, from, to, cursor, limit) + * mimeClass, origin, name, from, to, cursor, limit) * POST /files/groups/:groupId/upload direct web upload (≤20 MiB, multipart) * GET /files/:id one GroupFile * GET /files/:id/download stream payload (sha256 re-verified) diff --git a/apps/api/src/group-files/group-files.integration.test.ts b/apps/api/src/group-files/group-files.integration.test.ts new file mode 100644 index 0000000..464c40f --- /dev/null +++ b/apps/api/src/group-files/group-files.integration.test.ts @@ -0,0 +1,258 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { join } from "node:path"; +import { Pool, type PoolClient } from "pg"; + +/** + * The phase-two metadata-index query surface against a live PostgreSQL, as + * sp_app (FORCE RLS binds): ACL scoping (workspace, group membership), the + * availability/retention exclusion, the origin/MIME/sender/date/name facets + * and (createdAt,id) keyset pagination. + * + * Owns its OWN database (GROUP_FILES_IT_DATABASE_URL, plus + * GROUP_FILES_IT_SUPER_DATABASE_URL for migrations/seeding) for the same + * reason as groups.integration.test.ts: tenancy.integration.test.ts wipes + * every workspace in the shared DB and vitest runs files in parallel. + * Skipped when unset. + */ +const appUrl = process.env.GROUP_FILES_IT_DATABASE_URL; +const ownerUrl = process.env.GROUP_FILES_IT_SUPER_DATABASE_URL; + +type Modules = { + files: typeof import("./group-files.service.js"); + db: typeof import("../db/client.js"); +}; + +describe.skipIf(appUrl === undefined || ownerUrl === undefined)("group-files metadata index (live DB)", () => { + let m: Modules; + let ownerPool: Pool; + const suffix = `${Date.now()}${Math.floor(Math.random() * 1e6)}`; + let wsA: string; + let wsB: string; + let owner: string; + let publisher: string; + let outsider: string; + let groupA1: string; + let groupA2: string; + let groupB: string; + let svc: import("./group-files.service.js").GroupFilesService; + + function required(value: T | undefined, what: string): T { + if (value === undefined) throw new Error(`missing ${what}`); + return value; + } + + async function asOwner(fn: (client: PoolClient) => Promise): Promise { + const client = await ownerPool.connect(); + try { + return await fn(client); + } finally { + client.release(); + } + } + + async function inTenant(wsId: string, fn: (client: PoolClient) => Promise): Promise { + return asOwner(async (c) => { + await c.query("BEGIN"); + await c.query("SELECT set_config('app.workspace_id', $1, true)", [wsId]); + try { + return await fn(c); + } finally { + await c.query("COMMIT"); + } + }); + } + + /** Minimal archive row; only the metadata the index is allowed to hold. */ + async function seedFile( + ws: string, + group: string, + n: number, + extra: Partial<{ available: boolean; origin: string; sender: string; mime: string; mimeClass: string; createdAt: Date; name: string; retentionMode: string; retentionUntil: Date | null; eventId: string | null }> = {}, + ): Promise { + const sha = (seed: string): string => { + let out = ""; + while (out.length < 64) out += Buffer.from(seed, "utf8").toString("hex"); + return out.slice(0, 64); + }; + return inTenant(ws, async (c) => { + const res = await c.query<{ id: string }>( + `INSERT INTO chat_media_archive ( + workspace_id, group_id, room_id, event_id, origin, sender_mxid, matrix_event_json, + original_name, safe_name, mime, mime_class, byte_size, sha256, + ocis_resource_id, ocis_resource_version, created_at, available, retention_mode, retention_until + ) VALUES ($1,$2,'!room:x',$3,$4,$5,'{"mxcUri":"mxc://example.org/x"}',$6,$7,$8,$9,$10,$11,$12,'v1',$13,$14,$15,$16) + RETURNING id`, + [ + ws, group, extra.eventId ?? `$evt-${n}`, + extra.origin ?? "matrix", extra.sender ?? "@alice:example.org", + extra.name ?? `clip-${n}.png`, extra.name ?? `clip-${n}.png`, + extra.mime ?? "image/png", extra.mimeClass ?? "image", + 10 + n, sha(`seed-${n}`), `res-${n}`, + extra.createdAt ?? new Date(Date.UTC(2026, 0, n + 1, 12)), + extra.available ?? true, extra.retentionMode ?? "keep", extra.retentionUntil ?? null, + ], + ); + const id = res.rows[0]?.id; + if (id === undefined) throw new Error("seed row missing"); + return id; + }); + } + + beforeAll(async () => { + // The pools read their URLs at import time: point them at the isolated DB + // BEFORE importing the db client, then migrate that same DB. + process.env.DATABASE_URL = appUrl; + process.env.SUPER_DATABASE_URL = ownerUrl; + // The service mints library links lazily from env; no request ever leaves + // the test (only listFiles/getFile run, never a download). + process.env.FILES_OCID_WEBDAV_URL = "https://files.example.invalid"; + process.env.FILES_OCID_USER = "svc"; + process.env.FILES_OCID_PASSWORD = "pw"; + process.env.FILES_PUBLIC_BASE = "https://files.example.invalid"; + const { runMigrations } = await import("../db/migrate.js"); + await runMigrations(join(process.cwd(), "migrations")); + m = { files: await import("./group-files.service.js"), db: await import("../db/client.js") }; + ownerPool = new Pool({ connectionString: ownerUrl }); + + await asOwner(async (c) => { + const ws = async (name: string): Promise => + (await c.query("INSERT INTO workspaces (name, slug) VALUES ($1, $2) RETURNING id", [name, `${name}-gf-${suffix}`])).rows[0].id as string; + wsA = await ws("gf-a"); + wsB = await ws("gf-b"); + const user = async (tag: string): Promise => + ( + await c.query("INSERT INTO users (external_id, username, email, display_name) VALUES ($1, $2, $3, $4) RETURNING id", [ + `gf:${tag}:${suffix}`, `gf-${tag}-${suffix}`, `gf-${tag}-${suffix}@example.com`, tag, + ]) + ).rows[0].id as string; + owner = await user("owner"); + publisher = await user("publisher"); + outsider = await user("outsider"); + }); + + await inTenant(wsA, async (c) => { + await c.query("INSERT INTO memberships (workspace_id, user_id, role) VALUES ($1,$2,'Owner'), ($1,$3,'Publisher'), ($1,$4,'Publisher')", [wsA, owner, publisher, outsider]); + const group = async (name: string): Promise => + (await c.query("INSERT INTO groups (workspace_id, name, description) VALUES ($1,$2,'') RETURNING id", [wsA, name])).rows[0].id as string; + groupA1 = await group(`g1-${suffix}`); + groupA2 = await group(`g2-${suffix}`); + // publisher belongs to A1 only; outsider belongs to neither. + await c.query("INSERT INTO group_members (workspace_id, group_id, user_id) VALUES ($1,$2,$3)", [wsA, groupA1, publisher]); + }); + await inTenant(wsB, async (c) => { + await c.query("INSERT INTO memberships (workspace_id, user_id, role) VALUES ($1,$2,'Owner')", [wsB, owner]); + groupB = (await c.query("INSERT INTO groups (workspace_id, name, description) VALUES ($1,$2,'') RETURNING id", [wsB, `gb-${suffix}`])).rows[0].id as string; + }); + + svc = new m.files.GroupFilesService(); + + // A1: the page the tests page through, newest first. + for (let i = 0; i < 5; i += 1) await seedFile(wsA, groupA1, i); + // facets + await seedFile(wsA, groupA1, 90, { origin: "bridge", sender: "@bridged:example.org", mime: "video/mp4", mimeClass: "video", name: "teaser.mp4" }); + await seedFile(wsA, groupA1, 91, { available: false }); + await seedFile(wsA, groupA1, 92, { retentionMode: "until", retentionUntil: new Date("2020-01-01T00:00:00Z") }); + // A2 and the foreign workspace: ACL/cross-tenant scope + await seedFile(wsA, groupA2, 20); + await seedFile(wsB, groupB, 30); + }); + + afterAll(async () => { + if (ownerPool !== undefined) { + await asOwner(async (c) => { + await c.query("DELETE FROM workspaces WHERE id = ANY($1::uuid[])", [[wsA, wsB]]); + await c.query("DELETE FROM users WHERE external_id LIKE $1", [`gf:%:${suffix}`]); + }); + await ownerPool.end(); + } + await m?.db.pool.end(); + }); + + it("scopes by workspace + group membership: a non-Owner sees only their groups' rows", async () => { + const page = await svc.listFiles(wsA, publisher, "Publisher", {}); + expect(page.items.length).toBeGreaterThan(0); + // only A1 rows: never A2, never wsB + for (const item of page.items) expect(item.groupId).toBe(groupA1); + expect(page.items.map((i) => i.groupId)).not.toContain(groupA2); + // an explicit foreign-group filter intersects the membership scope to nothing + const filtered = await svc.listFiles(wsA, publisher, "Publisher", { groupId: groupA2 }); + expect(filtered.items).toEqual([]); + }); + + it("Owner/Admin see the whole workspace but never another tenant's rows", async () => { + const page = await svc.listFiles(wsA, owner, "Owner", {}); + const groups = new Set(page.items.map((i) => i.groupId)); + expect(groups.has(groupA1)).toBe(true); + expect(groups.has(groupA2)).toBe(true); + expect(groups.has(groupB)).toBe(false); + }); + + it("never returns unavailable or retention-expired rows", async () => { + const page = await svc.listFiles(wsA, owner, "Owner", {}); + expect(page.items.every((i) => i.status === "available")).toBe(true); + const byName = new Set(page.items.map((i) => i.name)); + expect(byName.has("clip-91.png")).toBe(false); // available = false + expect(byName.has("clip-92.png")).toBe(false); // retention 'until' in the past + // and the single-item read of a redacted row is Gone, not metadata + const redactedId = await inTenant(wsA, async (c) => { + const res = await c.query<{ id: string }>("SELECT id FROM chat_media_archive WHERE safe_name = 'clip-91.png' LIMIT 1"); + return required(res.rows[0]?.id, "redacted seed row"); + }); + await expect(svc.getFile(wsA, owner, "Owner", redactedId)).rejects.toMatchObject({ status: 410 }); + }); + + it("filters by origin, MIME class, sender and date", async () => { + const origins = await svc.listFiles(wsA, owner, "Owner", { origin: "bridge" }); + expect(origins.items.map((i) => i.origin.kind)).toEqual(["bridge"]); + const videos = await svc.listFiles(wsA, owner, "Owner", { mimeClass: "video" }); + expect(videos.items.map((i) => i.name)).toEqual(["teaser.mp4"]); + const bySender = await svc.listFiles(wsA, owner, "Owner", { senderMxid: "@bridged:example.org" }); + expect(bySender.items).toHaveLength(1); + const early = await svc.listFiles(wsA, owner, "Owner", { to: "2026-01-04" }); + expect(early.items.map((i) => i.createdAt).every((t) => new Date(t).getTime() < Date.parse("2026-01-04"))).toBe(true); + const byName = await svc.listFiles(wsA, owner, "Owner", { name: "teaser" }); + expect(byName.items.map((i) => i.name)).toEqual(["teaser.mp4"]); + }); + + it("pages with a (createdAt,id) keyset cursor and yields every row exactly once", async () => { + const seen: string[] = []; + let cursor: string | null = null; + // small pages force several round trips; no duplicated ids may appear + for (let guard = 0; guard < 20; guard += 1) { + const page = await svc.listFiles(wsA, owner, "Owner", { limit: "2", ...(cursor === null ? {} : { cursor }) }); + seen.push(...page.items.map((i) => i.id)); + if (page.nextCursor === null) break; + cursor = page.nextCursor; + } + const expected = await inTenant(wsA, async (c) => { + const res = await c.query<{ n: string }>( + "SELECT count(*) AS n FROM chat_media_archive WHERE workspace_id = $1 AND available AND (retention_mode <> 'until' OR retention_until > now())", + [wsA], + ); + return Number(required(res.rows[0]?.n, "count row")); + }); + expect(seen).toHaveLength(expected); + expect(new Set(seen).size).toBe(seen.length); + // strictly descending (createdAt,id): the keyset order the index serves + const items = await svc.listFiles(wsA, owner, "Owner", { limit: "100" }); + const keys = items.items.map((i) => `${i.createdAt}|${i.id}`); + const sorted = [...keys].sort((x, y) => (x < y ? 1 : x > y ? -1 : 0)); + expect(keys).toEqual(sorted); + // a cursor is opaque and carries no scope; re-binding is server-side + expect(items.nextCursor).toBe(null); + }); + + it("keeps a foreign group's single item unreadable (ACL before availability)", async () => { + const foreign = await inTenant(wsA, async (c) => { + const res = await c.query<{ id: string }>("SELECT id FROM chat_media_archive WHERE group_id = $1 LIMIT 1", [groupA2]); + return required(res.rows[0]?.id, "A2 seed row"); + }); + await expect(svc.getFile(wsA, publisher, "Publisher", foreign)).rejects.toMatchObject({ status: 403 }); + // cross-tenant: the id simply does not exist under wsA's RLS scope + const otherTenant = await inTenant(wsB, async (c) => { + const res = await c.query<{ id: string }>("SELECT id FROM chat_media_archive WHERE group_id = $1 LIMIT 1", [groupB]); + return required(res.rows[0]?.id, "wsB seed row"); + }); + await expect(svc.getFile(wsA, owner, "Owner", otherTenant)).rejects.toMatchObject({ status: 404 }); + }); +}); diff --git a/apps/api/src/group-files/group-files.service.test.ts b/apps/api/src/group-files/group-files.service.test.ts index f90a9d6..e30eae5 100644 --- a/apps/api/src/group-files/group-files.service.test.ts +++ b/apps/api/src/group-files/group-files.service.test.ts @@ -94,14 +94,15 @@ describe("GroupFilesService invariants", () => { const page = await service.listFiles(row.workspace_id, "user-1", "Publisher", {}); const list = statements.find((q) => q.sql.includes("FROM chat_media_archive"))!; expect(list.sql).toContain("group_members"); + expect(list.sql).toContain("a.workspace_id = $1::uuid"); + expect(list.values[0]).toBe(row.workspace_id); expect(list.values[1]).toBe("user-1"); // caller id is a bound parameter… expect(list.sql).not.toContain("user-1"); // …never SQL text - expect(list.sql).not.toContain("user-1"); - // Owner/Admin skip the membership subquery but keep $2 bound + // Owner/Admin skip the membership subquery; no unused placeholder remains await service.listFiles(row.workspace_id, "user-1", "Owner", {}); const ownerList = statements.filter((q) => q.sql.includes("FROM chat_media_archive")).at(-1)!; expect(ownerList.sql).not.toContain("group_members"); - expect(ownerList.values[1]).toBe("user-1"); + expect(ownerList.values).toEqual([row.workspace_id, 51]); // workspace + limit expect(page.items[0]?.status).toBe("available"); expect(page.items[0]?.shareLink.url).toBe("https://files.example.org/f/res-1"); }); @@ -114,6 +115,7 @@ describe("GroupFilesService invariants", () => { groupId: row.group_id, senderMxid: "@s:example.org", mimeClass: "video", + origin: "matrix", name: "clip", from: "2026-01-01", to: "2026-02-01", @@ -121,11 +123,19 @@ describe("GroupFilesService invariants", () => { limit: "10", }); const q = statements.find((s) => s.sql.includes("FROM chat_media_archive"))!; - expect(q.values).toEqual([row.workspace_id, "u", row.group_id, "@s:example.org", "video", "%clip%", "2026-01-01", "2026-02-01", "2026-01-01T00:00:00Z", row.id, 11]); + expect(q.values).toEqual([row.workspace_id, row.group_id, "@s:example.org", "video", "matrix", "%clip%", "2026-01-01", "2026-02-01", "2026-01-01T00:00:00Z", row.id, 11]); expect(q.sql).toContain("(a.created_at, a.id) <"); expect(q.sql).toContain("ORDER BY a.created_at DESC, a.id DESC"); // bad facets are rejected before SQL await expect(service.listFiles(row.workspace_id, "u", "Owner", { mimeClass: "zip" })).rejects.toBeInstanceOf(BadRequestException); + await expect(service.listFiles(row.workspace_id, "u", "Owner", { origin: "carrier-pigeon" })).rejects.toBeInstanceOf(BadRequestException); + for (const kind of ["matrix", "bridge", "upload", "pad-export"] as const) { + const svc = new GroupFilesService(); + await svc.listFiles(row.workspace_id, "u", "Owner", { origin: kind }); + const stmt = statements.filter((q2) => q2.sql.includes("FROM chat_media_archive")).at(-1)!; + expect(stmt.sql).toContain("a.origin = $2"); + expect(stmt.values[1]).toBe(kind); + } await expect(service.listFiles(row.workspace_id, "u", "Owner", { from: "nope" })).rejects.toBeInstanceOf(BadRequestException); await expect(service.listFiles(row.workspace_id, "u", "Owner", { from: "2026-03-01", to: "2026-01-01" })).rejects.toBeInstanceOf(BadRequestException); }); @@ -245,6 +255,41 @@ describe("GroupFilesService invariants", () => { ).rejects.toBeInstanceOf(ConflictException); }); + it("excludes unavailable rows from every archived-media read path", async () => { + const seen: string[] = []; + const redactedRow = { ...row, available: false }; + behavior = (sql) => { + if (sql.includes("FROM chat_media_archive")) { + seen.push(sql); + return { rows: [redactedRow], rowCount: 1 }; + } + return { rows: [], rowCount: 0 }; + }; + const service = new GroupFilesService(); + // list: the SQL itself must pin available = true (a redacted row never matches) + const page = await service.listFiles(row.workspace_id, "u", "Owner", {}); + const listSql = seen.at(-1)!; + expect(listSql).toContain("a.available = true"); + // single-item metadata read: ACL runs first, then Gone + await expect(service.getFile(row.workspace_id, "u", "Owner", row.id)).rejects.toBeInstanceOf(GoneException); + expect(page.items).toHaveLength(1); + }); + + it("checks the group ACL before disclosing a redacted row's availability", async () => { + const queries: string[] = []; + behavior = (sql) => { + if (!sql.startsWith("SELECT audit_append")) queries.push(sql); + if (sql.includes("FROM group_members")) return { rows: [], rowCount: 0 }; + if (sql.includes("FROM chat_media_archive")) return { rows: [{ ...row, available: false }], rowCount: 1 }; + return { rows: [], rowCount: 0 }; + }; + const service = new GroupFilesService(); + // non-member of the owning group: Forbidden, never Gone — the availability + // state of a foreign group's row is not disclosed to a non-member. + await expect(service.getFile(row.workspace_id, "u", "Publisher", row.id)).rejects.toBeInstanceOf(ForbiddenException); + expect(queries.some((sql) => sql.includes("FROM group_members"))).toBe(true); + }); + it("openDownload refuses revoked/expired items with Gone, missing rows with NotFound", async () => { behavior = () => ({ rows: [{ ...row, available: false }], rowCount: 1 }); const service = new GroupFilesService(); diff --git a/apps/api/src/group-files/group-files.service.ts b/apps/api/src/group-files/group-files.service.ts index d567edb..186c617 100644 --- a/apps/api/src/group-files/group-files.service.ts +++ b/apps/api/src/group-files/group-files.service.ts @@ -30,7 +30,7 @@ import { isUuid, parseLimit } from "../http/validation.js"; import { decodeFileListCursor, encodeFileListCursor } from "./file-cursor.js"; import { ocisConfigFromEnv, OcisWebdavClient, type FetchLike, type OcisDownload } from "./ocis-webdav.js"; import { hashShareToken, mintShareToken } from "./share-token.js"; -import { MIME_CLASSES, parseRetentionInput, type MimeClass } from "./group-files.validation.js"; +import { MIME_CLASSES, ORIGIN_KINDS, parseRetentionInput, type MimeClass, type OriginKind } from "./group-files.validation.js"; export const GROUP_FILES_DEPENDENCIES: unique symbol = Symbol("GROUP_FILES_DEPENDENCIES"); @@ -55,6 +55,8 @@ export interface FileListFilters { groupId: string | null; senderMxid: string | null; mimeClass: MimeClass | null; + /** SharedFileOrigin facet: matrix | bridge | upload | pad-export. */ + origin: OriginKind | null; name: string | null; from: string | null; to: string | null; @@ -215,13 +217,19 @@ export class GroupFilesService { const limit = parseLimit(query.limit, 50, 100); const cursor = typeof query.cursor === "string" && query.cursor !== "" ? decodeFileListCursor(query.cursor) : undefined; return withTenant(workspaceId, async (client) => { - // values[1] is always the caller's id: the membership subquery needs it - // for non-Owner/Admin callers; Owner/Admin leave $2 bound but unused so - // every placeholder stays positional and no SQL text carries a value. - const values: unknown[] = [workspaceId, userId]; - const conds: string[] = ["a.available = true", "(a.retention_mode <> 'until' OR a.retention_until > now())"]; + // $1 is the explicit workspace scope; RLS (app.workspace_id) enforces the + // same tenant inside withTenant, so the two agree. Every value is bound + // positionally — no SQL text ever carries a value, and no parameter is + // left unused (Postgres cannot type an unused placeholder). + const values: unknown[] = [workspaceId]; + const conds: string[] = [ + "a.workspace_id = $1::uuid", + "a.available = true", + "(a.retention_mode <> 'until' OR a.retention_until > now())", + ]; if (!WORKSPACE_WIDE_ROLES.includes(role)) { - conds.push("a.group_id IN (SELECT group_id FROM group_members WHERE user_id = $2 AND status = 'active')"); + values.push(userId); + conds.push(`a.group_id IN (SELECT group_id FROM group_members WHERE user_id = $${values.length})`); } if (filters.groupId !== null) { values.push(filters.groupId); @@ -235,6 +243,10 @@ export class GroupFilesService { values.push(filters.mimeClass); conds.push(`a.mime_class = $${values.length}`); } + if (filters.origin !== null) { + values.push(filters.origin); + conds.push(`a.origin = $${values.length}`); + } if (filters.name !== null) { values.push(`%${filters.name}%`); conds.push(`a.safe_name ILIKE $${values.length}`); @@ -269,10 +281,17 @@ export class GroupFilesService { }); } + /** + * Single archived-media read. Unavailable rows never surface: the ACL + * recheck runs first (so a non-member learns nothing about the row), then a + * redacted (available = false) row answers Gone — the same exclusion the + * list applies ("only available rows visible through current ACLs"). + */ async getFile(workspaceId: string, userId: string, role: string, itemId: string): Promise { return withTenant(workspaceId, async (client) => { const row = await this.requireRow(client, itemId); await this.requireAccess(client, userId, role, row.group_id); + if (!row.available) throw new GoneException("file is revoked"); return rowToGroupFile(row, (resourceId) => this.webdav().libraryUrl(resourceId)); }); } @@ -496,7 +515,7 @@ export class GroupFilesService { private async requireAccess(client: PoolClient, userId: string, role: string, groupId: string): Promise { if (WORKSPACE_WIDE_ROLES.includes(role)) return; const result = await client.query( - "SELECT 1 FROM group_members WHERE group_id = $1 AND user_id = $2 AND status = 'active'", + "SELECT 1 FROM group_members WHERE group_id = $1 AND user_id = $2", [groupId, userId], ); if ((result.rowCount ?? 0) === 0) throw new ForbiddenException("file belongs to a group you are not a member of"); @@ -534,7 +553,7 @@ export class GroupFilesService { ): Promise { if (!WORKSPACE_WIDE_ROLES.includes(actor.role)) { const membership = await withTenant(workspaceId, (client) => - client.query("SELECT 1 FROM group_members WHERE group_id = $1 AND user_id = $2 AND status = 'active' LIMIT 1", [groupId, actor.userId]), + client.query("SELECT 1 FROM group_members WHERE group_id = $1 AND user_id = $2 LIMIT 1", [groupId, actor.userId]), ); if (membership.rows.length === 0) throw new ForbiddenException("not a member of this group"); } @@ -605,11 +624,15 @@ function parseFilters(query: Record): FileListFilters { if (rawClass !== null && !(MIME_CLASSES as readonly string[]).includes(rawClass)) { throw new BadRequestException(`mimeClass must be one of: ${MIME_CLASSES.join(", ")}`); } + const rawOrigin = typeof query.origin === "string" && query.origin !== "" ? query.origin : null; + if (rawOrigin !== null && !(ORIGIN_KINDS as readonly string[]).includes(rawOrigin)) { + throw new BadRequestException(`origin must be one of: ${ORIGIN_KINDS.join(", ")}`); + } const name = typeof query.name === "string" && query.name.trim() !== "" ? query.name.trim().slice(0, 255) : null; const from = parseBound(query.from, "from"); const to = parseBound(query.to, "to"); if (from !== null && to !== null && Date.parse(from) > Date.parse(to)) throw new BadRequestException("from must not be after to"); - return { groupId, senderMxid, mimeClass: rawClass as MimeClass | null, name, from, to }; + return { groupId, senderMxid, mimeClass: rawClass as MimeClass | null, origin: rawOrigin as OriginKind | null, name, from, to }; } function optionalUuid(value: unknown): string | null { diff --git a/apps/archive-bot/src/processor.test.ts b/apps/archive-bot/src/processor.test.ts index 7d7cc0f..23e726a 100644 --- a/apps/archive-bot/src/processor.test.ts +++ b/apps/archive-bot/src/processor.test.ts @@ -90,6 +90,8 @@ class FakeStore implements ArchiveStore { roomsSaved = new Set(); outbox = new Map(); archive: (ArchiveRecord & { available: boolean })[] = []; + /** Live share rows per archive record — mirrors group_file_shares. */ + shareLinks = new Map(); async loadCursor(_rooms: readonly RoomCursorTarget[]): Promise { return this.cursor; } @@ -106,8 +108,10 @@ class FakeStore implements ArchiveStore { } async commitArchived(record: ArchiveRecord, outboxId: string): Promise<{ id: string; indexedAt: Date }> { this.archive.push({ ...record, available: true }); + const id = `file-${this.archive.length}`; + this.shareLinks.set(id, { revokedAt: null }); this.byId(outboxId).status = "done"; - return { id: `file-${this.archive.length}`, indexedAt: new Date(0) }; + return { id, indexedAt: new Date(0) }; } async recordFailure(_ws: string, outboxId: string, error: string, terminal: boolean): Promise { const row = this.byId(outboxId); @@ -118,11 +122,17 @@ class FakeStore implements ArchiveStore { return { ...row }; } async markRedacted(_ws: string, _room: string, eventId: string, key: string): Promise { - const rows = this.archive.filter((r) => r.eventId === eventId && r.available); - for (const r of rows) r.available = false; + const affected = this.archive + .map((r, i) => ({ r, id: `file-${i + 1}` })) + .filter(({ r }) => r.eventId === eventId && r.available); + for (const { r, id } of affected) { + r.available = false; + const share = this.shareLinks.get(id); + if (share !== undefined) share.revokedAt = clock.now; + } const claim = this.outbox.get(key); if (claim?.status === "pending") claim.status = "failed"; - return rows.length; + return affected.length; } async countPending(): Promise { return [...this.outbox.values()].filter((r) => r.status === "pending").length; @@ -251,13 +261,16 @@ describe("ArchiveProcessor", () => { expect(store.outbox.size).toBe(0); }); - it("marks redacted attachments unavailable without touching the library", async () => { + it("marks redacted attachments unavailable and revokes live links, without touching the library", async () => { matrix.batches.push(batch("s1", [imageEvent("$a")])); await processor.runBatch(signal); + expect(store.shareLinks.get("file-1")).toEqual({ revokedAt: null }); const redaction: MatrixEvent = { type: "m.room.redaction", event_id: "$r", sender: "@mod:x", content: { redacts: "$a" } }; matrix.batches.push(batch("s2", [redaction])); await processor.runBatch(signal); expect(store.archive[0]?.available).toBe(false); + // contract point 5: revoke links AND mark the index unavailable + expect(store.shareLinks.get("file-1")?.revokedAt).toEqual(clock.now); expect(library.committed).toHaveLength(1); }); diff --git a/apps/archive-bot/src/store.ts b/apps/archive-bot/src/store.ts index c26fa7f..55dd1ba 100644 --- a/apps/archive-bot/src/store.ts +++ b/apps/archive-bot/src/store.ts @@ -46,7 +46,11 @@ export interface ArchiveStore { /** Index row + outbox done in one transaction. Returns the archive row id and indexed time. */ commitArchived(record: ArchiveRecord, outboxId: string): Promise<{ id: string; indexedAt: Date }>; recordFailure(workspaceId: string, outboxId: string, error: string, terminal: boolean): Promise; - /** Mark index rows unavailable and stop any not-yet-archived claim. Returns rows affected. */ + /** + * Redaction/membership-removal path (contract point 5): mark index rows + * unavailable AND revoke every live share link for them in the same + * transaction, so serving stops immediately. Returns rows affected. + */ markRedacted(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise; countPending(workspaceIds: readonly string[]): Promise; } @@ -182,9 +186,19 @@ export class PgArchiveStore implements ArchiveStore { return this.withTenant(workspaceId, async (c) => { const res = await c.query( `UPDATE chat_media_archive SET available = false - WHERE workspace_id = $1 AND room_id = $2 AND event_id = $3 AND available`, + WHERE workspace_id = $1 AND room_id = $2 AND event_id = $3 AND available + RETURNING id`, [workspaceId, roomId, eventId], ); + // "revoke links and mark the index unavailable; stop serving immediately": + // every live share of the now-unavailable rows dies in the same tx. + if (res.rows.length > 0) { + await c.query( + `UPDATE group_file_shares SET revoked_at = now() + WHERE item_id = ANY($1::uuid[]) AND workspace_id = $2 AND revoked_at IS NULL`, + [res.rows.map((r: { id: string }) => r.id), workspaceId], + ); + } await c.query( `UPDATE archive_sync_outbox SET status = 'failed', last_error = 'redacted', updated_at = now() WHERE workspace_id = $1 AND claim_key = $2 AND status = 'pending'`,