Phase two: archive origin facet, redaction-aware reads, share revocation on redaction

- group-files: origin filter facet; GET /files/:id returns 410 for unavailable
  rows after the ACL recheck; fix latent group_members.status filter and unbound
  workspace parameter that would 500 listFiles
- archive-bot: markRedacted revokes live group_file_shares rows in-transaction
- new live-DB integration suite for the archive index surface
This commit is contained in:
glass beadian 2026-10-11 21:44:23 +00:00
parent 7cea73dc0d
commit 7ea0b1e137
6 changed files with 375 additions and 22 deletions

View file

@ -90,6 +90,8 @@ class FakeStore implements ArchiveStore {
roomsSaved = new Set<string>();
outbox = new Map<string, OutboxRow & { claimKey: string }>();
archive: (ArchiveRecord & { available: boolean })[] = [];
/** Live share rows per archive record — mirrors group_file_shares. */
shareLinks = new Map<string, { revokedAt: Date | null }>();
async loadCursor(_rooms: readonly RoomCursorTarget[]): Promise<string | null> {
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<OutboxRow> {
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<number> {
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<number> {
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);
});

View file

@ -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<OutboxRow>;
/** 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<number>;
countPending(workspaceIds: readonly string[]): Promise<number>;
}
@ -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'`,