416 lines
18 KiB
TypeScript
416 lines
18 KiB
TypeScript
import pino from "pino";
|
|
import { beforeEach, describe, expect, it } from "vitest";
|
|
import { Archiver } from "./archiver.js";
|
|
import { BACKOFF_BASE_MS } from "./backoff.js";
|
|
import type { RoomMapping } from "./config.js";
|
|
import type { MatrixEvent, RoomEncryption } from "./events.js";
|
|
import { claimKey, replyTxnId } from "./keys.js";
|
|
import type { MatrixPort, MediaDownload, SyncResult } from "./matrix.js";
|
|
import { ArchiveProcessor } from "./processor.js";
|
|
import type { ArchiveRecord, ArchiveStore, OutboxRow, RoomCursorTarget } from "./store.js";
|
|
import type { IndexedFile } from "./archive-index.js";
|
|
import type { LibraryPort, LibraryResource } from "./webdav.js";
|
|
|
|
const mapping: RoomMapping = {
|
|
roomId: "!room:example.org",
|
|
workspaceId: "11111111-1111-4111-8111-111111111111",
|
|
groupId: "22222222-2222-4222-8222-222222222222",
|
|
};
|
|
const BOT = "@archive:example.org";
|
|
const PNG = Uint8Array.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1, 2, 3]);
|
|
|
|
function imageEvent(eventId: string, extra: Partial<MatrixEvent> = {}): MatrixEvent {
|
|
return {
|
|
type: "m.room.message",
|
|
event_id: eventId,
|
|
sender: "@alice:example.org",
|
|
origin_server_ts: 1_700_000_000_000,
|
|
content: { msgtype: "m.image", body: "../cat", url: `mxc://example.org/${eventId.slice(1)}` },
|
|
...extra,
|
|
};
|
|
}
|
|
|
|
class FakeMatrix implements MatrixPort {
|
|
readonly userId = BOT;
|
|
batches: SyncResult[] = [];
|
|
syncedWith: (string | null)[] = [];
|
|
encryption: RoomEncryption = "unencrypted";
|
|
notices: { roomId: string; txnId: string; content: Record<string, unknown> }[] = [];
|
|
failNotices = 0;
|
|
joined: string[] = [];
|
|
|
|
async sync(since: string | null): Promise<SyncResult> {
|
|
this.syncedWith.push(since);
|
|
const next = this.batches.shift();
|
|
if (next === undefined) throw new Error("no batch queued");
|
|
return next;
|
|
}
|
|
async backfill(): Promise<MatrixEvent[]> {
|
|
return [];
|
|
}
|
|
async roomEncryption(): Promise<RoomEncryption> {
|
|
return this.encryption;
|
|
}
|
|
async joinRoom(roomId: string): Promise<void> {
|
|
this.joined.push(roomId);
|
|
}
|
|
async downloadMedia(): Promise<MediaDownload> {
|
|
return {
|
|
body: new ReadableStream({ start: (c) => { c.enqueue(PNG); c.close(); } }),
|
|
contentLength: PNG.byteLength,
|
|
};
|
|
}
|
|
async sendNotice(roomId: string, txnId: string, content: Record<string, unknown>): Promise<void> {
|
|
if (this.failNotices > 0) {
|
|
this.failNotices -= 1;
|
|
throw new Error("notice failed");
|
|
}
|
|
this.notices.push({ roomId, txnId, content });
|
|
}
|
|
powerLevels: Record<string, unknown> = { users_default: 0, users: {} };
|
|
roomEncryptionByRoom: Map<string, RoomEncryption> = new Map();
|
|
joinedRoomList: string[] = [mapping.roomId];
|
|
async roomPowerLevels(roomId: string): Promise<Record<string, unknown>> {
|
|
void roomId;
|
|
return this.powerLevels;
|
|
}
|
|
async joinedRooms(): Promise<string[]> {
|
|
return this.joinedRoomList;
|
|
}
|
|
}
|
|
|
|
class FakeLibrary implements LibraryPort {
|
|
failPuts = 0;
|
|
committed: string[] = [];
|
|
async putStaging(_dir: string[], _name: string, body: ReadableStream<Uint8Array>): Promise<void> {
|
|
for await (const chunk of body) void chunk;
|
|
if (this.failPuts > 0) {
|
|
this.failPuts -= 1;
|
|
throw new Error("WebDAV PUT failed: 507");
|
|
}
|
|
}
|
|
async commit(dir: string[], _staging: string, finalName: string): Promise<LibraryResource> {
|
|
const path = [...dir, finalName].join("/");
|
|
this.committed.push(path);
|
|
return { path, url: `https://files/${path}`, resourceId: "rid", version: "v1" };
|
|
}
|
|
/** files keyed by `${dir}/${name}` — the export read side. */
|
|
libraryFiles = new Map<string, Uint8Array>();
|
|
async getResource(dir: string[], finalName: string): Promise<ReadableStream<Uint8Array>> {
|
|
const key = [...dir, finalName].join("/");
|
|
const bytes = this.libraryFiles.get(key);
|
|
if (bytes === undefined) throw new Error(`no such resource ${key}`);
|
|
return new Blob([bytes]).stream() as ReadableStream<Uint8Array>;
|
|
}
|
|
}
|
|
|
|
class FakeStore implements ArchiveStore {
|
|
cursor: string | null = null;
|
|
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;
|
|
}
|
|
async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void> {
|
|
for (const room of rooms) this.roomsSaved.add(room.roomId);
|
|
this.cursor = cursor;
|
|
}
|
|
async claim(_ws: string, _room: string, _event: string, key: string): Promise<OutboxRow> {
|
|
const existing = this.outbox.get(key);
|
|
if (existing !== undefined) return { ...existing };
|
|
const row = { id: `ob${this.outbox.size + 1}`, status: "pending" as const, attempts: 0, lastError: null, updatedAt: new Date(0), claimKey: key };
|
|
this.outbox.set(key, row);
|
|
return { ...row };
|
|
}
|
|
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, indexedAt: new Date(0) };
|
|
}
|
|
async recordFailure(_ws: string, outboxId: string, error: string, terminal: boolean): Promise<OutboxRow> {
|
|
const row = this.byId(outboxId);
|
|
row.attempts += 1;
|
|
row.lastError = error;
|
|
row.status = terminal ? "failed" : "pending";
|
|
row.updatedAt = clock.now;
|
|
return { ...row };
|
|
}
|
|
async markRedacted(_ws: string, _room: string, eventId: string, key: string): Promise<number> {
|
|
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 affected.length;
|
|
}
|
|
async countPending(): Promise<number> {
|
|
return [...this.outbox.values()].filter((r) => r.status === "pending").length;
|
|
}
|
|
async listRoom(_ws: string, roomId: string): Promise<IndexedFile[]> {
|
|
return this.archive.filter((r) => r.roomId === roomId).map((r, i) => ({
|
|
id: `file-${i + 1}`,
|
|
eventId: r.eventId,
|
|
mxcUri: "",
|
|
originalName: r.originalName,
|
|
safeName: r.safeName,
|
|
mime: r.mime,
|
|
mimeClass: r.mimeClass as IndexedFile["mimeClass"],
|
|
byteSize: r.byteSize,
|
|
sha256: r.sha256,
|
|
sender: r.senderMxid,
|
|
createdAt: r.createdAt.toISOString(),
|
|
ocisResourceId: "rid",
|
|
available: r.available,
|
|
}));
|
|
}
|
|
private byId(id: string): OutboxRow {
|
|
const row = [...this.outbox.values()].find((r) => r.id === id);
|
|
if (row === undefined) throw new Error("no row");
|
|
return row;
|
|
}
|
|
}
|
|
|
|
const clock = { now: new Date(1_000_000) };
|
|
|
|
function batch(nextBatch: string, events: MatrixEvent[], extra: { stateEvents?: MatrixEvent[]; invited?: string[] } = {}): SyncResult {
|
|
return {
|
|
nextBatch,
|
|
invited: extra.invited ?? [],
|
|
joined: new Map([[mapping.roomId, { timeline: { events, limited: false, prevBatch: null }, stateEvents: extra.stateEvents ?? [] }]]),
|
|
};
|
|
}
|
|
|
|
describe("ArchiveProcessor", () => {
|
|
let matrix: FakeMatrix;
|
|
let library: FakeLibrary;
|
|
let store: FakeStore;
|
|
let processor: ArchiveProcessor;
|
|
const signal = new AbortController().signal;
|
|
|
|
beforeEach(async () => {
|
|
clock.now = new Date(1_000_000);
|
|
matrix = new FakeMatrix();
|
|
library = new FakeLibrary();
|
|
store = new FakeStore();
|
|
processor = new ArchiveProcessor(
|
|
matrix, store, new Archiver(matrix, library, store), [mapping], pino({ level: "silent" }), () => clock.now,
|
|
);
|
|
await processor.init();
|
|
});
|
|
|
|
it("archives, indexes, replies once with the stable txn id, then advances the cursor", async () => {
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]));
|
|
expect(await processor.runBatch(signal)).toEqual({ advanced: true, retryAt: null });
|
|
|
|
expect(store.cursor).toBe("s1");
|
|
expect([...store.roomsSaved]).toEqual([mapping.roomId]);
|
|
expect(store.archive).toHaveLength(1);
|
|
const row = store.archive[0];
|
|
expect(row).toMatchObject({ mime: "image/png", mimeClass: "image", safeName: "cat.png", originalName: "../cat", byteSize: PNG.byteLength });
|
|
expect(library.committed[0]).toBe(`Library/${mapping.workspaceId}/${mapping.groupId}/${row?.sha256}-cat.png`);
|
|
const key = claimKey(mapping.workspaceId, mapping.roomId, "$a");
|
|
expect(matrix.notices).toEqual([{
|
|
roomId: mapping.roomId,
|
|
txnId: replyTxnId(key),
|
|
content: expect.objectContaining({ body: "Archived to the group library: cat.png" }) as unknown,
|
|
}]);
|
|
|
|
// Replaying the same event (e.g. after a crash before cursor save) neither re-archives nor re-replies.
|
|
matrix.batches.push(batch("s2", [imageEvent("$a")]));
|
|
await processor.runBatch(signal);
|
|
expect(store.archive).toHaveLength(1);
|
|
expect(matrix.notices).toHaveLength(1);
|
|
});
|
|
|
|
it("holds the cursor on a WebDAV failure, records attempts, and retries after backoff", async () => {
|
|
library.failPuts = 1;
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]));
|
|
const first = await processor.runBatch(signal);
|
|
expect(first.advanced).toBe(false);
|
|
expect(first.retryAt?.getTime()).toBe(clock.now.getTime() + BACKOFF_BASE_MS);
|
|
expect(store.cursor).toBeNull();
|
|
const row = [...store.outbox.values()][0];
|
|
expect(row).toMatchObject({ status: "pending", attempts: 1, lastError: "WebDAV PUT failed: 507" });
|
|
|
|
// Re-sync before the row is due: still blocked, nothing uploaded.
|
|
matrix.batches.push(batch("s1b", [imageEvent("$a")]));
|
|
expect((await processor.runBatch(signal)).advanced).toBe(false);
|
|
expect(store.archive).toHaveLength(0);
|
|
expect(matrix.syncedWith).toEqual([null, null]);
|
|
|
|
clock.now = new Date(clock.now.getTime() + BACKOFF_BASE_MS);
|
|
matrix.batches.push(batch("s1c", [imageEvent("$a")]));
|
|
expect((await processor.runBatch(signal)).advanced).toBe(true);
|
|
expect(store.cursor).toBe("s1c");
|
|
expect(store.archive).toHaveLength(1);
|
|
});
|
|
|
|
it("retries a failed reply without re-archiving", async () => {
|
|
matrix.failNotices = 1;
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]), batch("s2", [imageEvent("$a")]));
|
|
expect((await processor.runBatch(signal)).advanced).toBe(false);
|
|
expect(store.archive).toHaveLength(1);
|
|
expect((await processor.runBatch(signal)).advanced).toBe(true);
|
|
expect(matrix.notices).toHaveLength(1);
|
|
expect(store.archive).toHaveLength(1);
|
|
});
|
|
|
|
it("marks permanent payload errors failed and lets the cursor advance", async () => {
|
|
matrix.downloadMedia = async () => ({
|
|
body: new ReadableStream({ start: (c) => { c.enqueue(PNG); c.close(); } }),
|
|
contentLength: PNG.byteLength + 10,
|
|
});
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]));
|
|
expect((await processor.runBatch(signal)).advanced).toBe(true);
|
|
expect([...store.outbox.values()][0]).toMatchObject({ status: "failed", attempts: 1 });
|
|
expect(matrix.notices).toHaveLength(0);
|
|
});
|
|
|
|
it("never mirrors when the room is encrypted or its state is unknown", async () => {
|
|
matrix.encryption = "unknown";
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]));
|
|
await processor.runBatch(signal);
|
|
matrix.encryption = "unencrypted";
|
|
const encryptionEvent: MatrixEvent = {
|
|
type: "m.room.encryption", event_id: "$enc", sender: "@admin:x", state_key: "", content: { algorithm: "m.megolm.v1.aes-sha2" },
|
|
};
|
|
matrix.batches.push(batch("s2", [imageEvent("$b"), encryptionEvent]));
|
|
await processor.runBatch(signal);
|
|
expect(store.outbox.size).toBe(0);
|
|
expect(store.archive).toHaveLength(0);
|
|
expect(store.cursor).toBe("s2");
|
|
});
|
|
|
|
it("ignores its own events", async () => {
|
|
matrix.batches.push(batch("s1", [imageEvent("$a", { sender: BOT })]));
|
|
await processor.runBatch(signal);
|
|
expect(store.outbox.size).toBe(0);
|
|
});
|
|
|
|
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);
|
|
});
|
|
|
|
it("runs the encryption lint over joined rooms and reports violations in health", async () => {
|
|
matrix.joinedRoomList = [mapping.roomId, "!stray:x"];
|
|
matrix.roomEncryptionByRoom = new Map<string, RoomEncryption>([["!stray:x", "unencrypted"]]);
|
|
const report = await processor.lintEncryption(signal);
|
|
expect(report.issues.map((i) => i.kind)).toEqual(["unmapped-unencrypted"]);
|
|
const health = await processor.health();
|
|
expect(health.encryption).toMatchObject({ checked: 2, violations: ["!stray:x"], mappedUnencrypted: [mapping.roomId] });
|
|
});
|
|
|
|
it("handles !archive export from a room admin: uploads ciphertext and posts the object key", async () => {
|
|
const { createServer } = await import("node:http");
|
|
const { mkdtemp, rm } = await import("node:fs/promises");
|
|
const { tmpdir } = await import("node:os");
|
|
const { join } = await import("node:path");
|
|
const age = await import("age-encryption");
|
|
const { BundleExporter } = await import("./exporter.js");
|
|
const { R2ExportStore } = await import("./r2-export.js");
|
|
|
|
const identity = await age.generateIdentity();
|
|
const recipient = await age.identityToRecipient(identity);
|
|
const dir = await mkdtemp(join(tmpdir(), "proc-export-"));
|
|
const puts: { key: string; body: Buffer }[] = [];
|
|
const server = createServer((req, res) => {
|
|
const chunks: Buffer[] = [];
|
|
req.on("data", (c: Buffer) => chunks.push(c));
|
|
req.on("end", () => { puts.push({ key: req.url ?? "", body: Buffer.concat(chunks) }); res.writeHead(200).end("{}"); });
|
|
});
|
|
await new Promise<void>((r) => server.listen(0, "127.0.0.1", r));
|
|
try {
|
|
matrix.powerLevels = { users: { "@alice:example.org": 100 }, users_default: 0 };
|
|
const upload = new R2ExportStore({
|
|
endpoint: `http://127.0.0.1:${(server.address() as { port: number }).port}`,
|
|
region: "auto", accessKeyId: "k", secretAccessKey: "s", spillDir: dir, allowInsecureEndpoint: true,
|
|
});
|
|
const exportConfig = {
|
|
enabled: true, every: 0, ageRecipients: [recipient],
|
|
r2: { endpoint: "http://stub", region: "auto", accessKeyId: "k", secretAccessKey: "s", bucket: "hub-matrix-media", tmpDir: dir },
|
|
};
|
|
const exporter = new BundleExporter(store, library, upload, [recipient], "hub-matrix-media", pino({ level: "silent" }));
|
|
const proc = new ArchiveProcessor(
|
|
matrix, store, new Archiver(matrix, library, store), [mapping], pino({ level: "silent" }), () => clock.now,
|
|
exporter, exportConfig,
|
|
);
|
|
await proc.init();
|
|
|
|
// archive one attachment so the exporter has a payload to bundle
|
|
matrix.batches.push(batch("s1", [imageEvent("$a")]));
|
|
await proc.runBatch(signal);
|
|
expect(store.archive).toHaveLength(1);
|
|
const archived = store.archive[0];
|
|
if (archived === undefined) throw new Error("no archive row");
|
|
library.libraryFiles.set(`Library/${mapping.workspaceId}/${mapping.groupId}/${archived.sha256}-${archived.safeName}`, PNG);
|
|
|
|
// admin command
|
|
matrix.batches.push(batch("s2", [{
|
|
type: "m.room.message",
|
|
event_id: "$cmd",
|
|
sender: "@alice:example.org",
|
|
origin_server_ts: 1,
|
|
content: { msgtype: "m.text", body: "!archive export" },
|
|
}]));
|
|
await proc.runBatch(signal);
|
|
|
|
expect(puts).toHaveLength(1);
|
|
expect(puts[0]?.key).toContain("/hub-matrix-media/exports/");
|
|
const notice = matrix.notices.at(-1);
|
|
expect(String(notice?.content.body)).toContain("exports/");
|
|
// the uploaded bytes decrypt with the admin identity and contain the payload
|
|
const dec = new age.Decrypter();
|
|
dec.addIdentity(identity);
|
|
const plain = await dec.decrypt(new Blob([puts[0]?.body ?? Buffer.alloc(0)]).stream() as ReadableStream<Uint8Array>);
|
|
const { readTarZstd, drainToBuffer } = await import("./bundle.js");
|
|
const names: string[] = [];
|
|
for await (const entry of readTarZstd(plain)) { names.push(entry.name); await drainToBuffer(entry.stream); }
|
|
expect(names).toContain("manifest.json");
|
|
expect(names).toContain(`files/$a/${archived.safeName}`);
|
|
const health = await proc.health();
|
|
expect(health.export).toMatchObject({ enabled: true, lastObjectKey: expect.stringContaining("exports/") });
|
|
} finally {
|
|
await rm(dir, { recursive: true, force: true });
|
|
await new Promise<void>((r) => server.close(() => r()));
|
|
}
|
|
});
|
|
|
|
it("ignores !archive export when exports are not configured or the sender is not an admin", async () => {
|
|
// not configured: no exporter attached
|
|
matrix.batches.push(batch("s1", [{
|
|
type: "m.room.message",
|
|
event_id: "$cmd",
|
|
sender: "@alice:example.org",
|
|
origin_server_ts: 1,
|
|
content: { msgtype: "m.text", body: "!archive export" },
|
|
}]));
|
|
await processor.runBatch(signal);
|
|
expect(matrix.notices).toHaveLength(0);
|
|
});
|
|
|
|
it("joins invites only for mapped rooms and reports health", async () => {
|
|
matrix.batches.push({ nextBatch: "s1", joined: new Map(), invited: [mapping.roomId, "!other:x"] });
|
|
await processor.runBatch(signal);
|
|
expect(matrix.joined).toEqual([mapping.roomId]);
|
|
expect(await processor.health()).toMatchObject({ rooms: 1, cursor: "s1", pending: 0, lastProcessedAt: null });
|
|
});
|
|
});
|