296 lines
11 KiB
TypeScript
296 lines
11 KiB
TypeScript
import { Pool, type PoolClient } from "pg";
|
|
import type { ArchiveIndexPort, IndexedFile } from "./archive-index.js";
|
|
|
|
export type OutboxStatus = "pending" | "done" | "failed";
|
|
|
|
export interface OutboxRow {
|
|
id: string;
|
|
status: OutboxStatus;
|
|
attempts: number;
|
|
lastError: string | null;
|
|
updatedAt: Date;
|
|
}
|
|
|
|
export interface ArchiveRecord {
|
|
workspaceId: string;
|
|
groupId: string;
|
|
roomId: string;
|
|
eventId: string;
|
|
senderMxid: string;
|
|
matrixEventJson: unknown;
|
|
originalName: string;
|
|
safeName: string;
|
|
mime: string;
|
|
mimeClass: string;
|
|
byteSize: number;
|
|
sha256: string;
|
|
ocisResourceId: string;
|
|
ocisResourceVersion: string;
|
|
createdAt: Date;
|
|
}
|
|
|
|
/**
|
|
* Cursor rows are keyed by room (029: archive_sync_state(room_id PRIMARY KEY,
|
|
* workspace_id NOT NULL, ...)), so the store needs the room → tenant pin the
|
|
* config mappings already carry. RoomMapping satisfies this structurally.
|
|
*/
|
|
export interface RoomCursorTarget {
|
|
roomId: string;
|
|
workspaceId: string;
|
|
}
|
|
|
|
/** Durable state the processor depends on; faked in tests. */
|
|
export interface ArchiveStore {
|
|
loadCursor(rooms: readonly RoomCursorTarget[]): Promise<string | null>;
|
|
saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void>;
|
|
claim(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise<OutboxRow>;
|
|
/** 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>;
|
|
/**
|
|
* 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>;
|
|
/**
|
|
* Phase-B export manifest source (ArchiveIndexPort): every archive row for
|
|
* one room, oldest first. Read-only; no payload bytes leave this interface.
|
|
*/
|
|
listRoom(workspaceId: string, roomId: string): Promise<IndexedFile[]>;
|
|
}
|
|
|
|
interface OutboxDbRow {
|
|
id: string;
|
|
status: OutboxStatus;
|
|
attempts: number;
|
|
last_error: string | null;
|
|
updated_at: Date;
|
|
}
|
|
|
|
function toOutbox(row: OutboxDbRow): OutboxRow {
|
|
return { id: row.id, status: row.status, attempts: row.attempts, lastError: row.last_error, updatedAt: row.updated_at };
|
|
}
|
|
|
|
const OUTBOX_COLUMNS = "id, status, attempts, last_error, updated_at";
|
|
|
|
interface RoomArchiveDbRow {
|
|
id: string;
|
|
event_id: string;
|
|
sender_mxid: string;
|
|
original_name: string;
|
|
safe_name: string;
|
|
mime: string;
|
|
mime_class: string;
|
|
byte_size: string | number;
|
|
sha256: string;
|
|
ocis_resource_id: string;
|
|
created_at: Date;
|
|
available: boolean;
|
|
/** Extracted from the stored event JSON; NULL for pre-archive rows without one. */
|
|
mxc_uri: string | null;
|
|
}
|
|
|
|
function toIndexedFile(row: RoomArchiveDbRow): IndexedFile {
|
|
return {
|
|
id: row.id,
|
|
eventId: row.event_id,
|
|
mxcUri: row.mxc_uri ?? "",
|
|
originalName: row.original_name,
|
|
safeName: row.safe_name,
|
|
mime: row.mime,
|
|
mimeClass: row.mime_class as IndexedFile["mimeClass"],
|
|
byteSize: typeof row.byte_size === "string" ? Number(row.byte_size) : row.byte_size,
|
|
sha256: row.sha256,
|
|
sender: row.sender_mxid,
|
|
createdAt: row.created_at.toISOString(),
|
|
ocisResourceId: row.ocis_resource_id,
|
|
available: row.available,
|
|
};
|
|
}
|
|
const LAST_ERROR_MAX = 1000;
|
|
|
|
export class PgArchiveStore implements ArchiveStore, ArchiveIndexPort {
|
|
constructor(private readonly pool: Pool) {}
|
|
|
|
/** Single service connection; tenant context per transaction, like apps/api withTenant. */
|
|
static create(databaseUrl: string): PgArchiveStore {
|
|
return new PgArchiveStore(new Pool({ connectionString: databaseUrl, max: 1 }));
|
|
}
|
|
|
|
async close(): Promise<void> {
|
|
await this.pool.end();
|
|
}
|
|
|
|
/**
|
|
* The sync stream is global but cursor rows are per room, so each mapped room
|
|
* holds a copy. A partial write leaves copies disagreeing: resume from the
|
|
* least recently written (oldest) one — replay is idempotent.
|
|
*/
|
|
async loadCursor(rooms: readonly RoomCursorTarget[]): Promise<string | null> {
|
|
let oldest: { cursor: string; at: Date } | null = null;
|
|
for (const room of rooms) {
|
|
const row = await this.withTenant(room.workspaceId, async (c) => {
|
|
const res = await c.query<{ sync_cursor: string | null; updated_at: Date }>(
|
|
"SELECT sync_cursor, updated_at FROM archive_sync_state WHERE room_id = $1 AND workspace_id = $2",
|
|
[room.roomId, room.workspaceId],
|
|
);
|
|
return res.rows[0];
|
|
});
|
|
if (row?.sync_cursor == null) continue;
|
|
if (oldest === null || row.updated_at < oldest.at) oldest = { cursor: row.sync_cursor, at: row.updated_at };
|
|
}
|
|
return oldest?.cursor ?? null;
|
|
}
|
|
|
|
async saveCursor(rooms: readonly RoomCursorTarget[], cursor: string): Promise<void> {
|
|
for (const room of rooms) {
|
|
await this.withTenant(room.workspaceId, (c) => c.query(
|
|
`INSERT INTO archive_sync_state (room_id, workspace_id, sync_cursor, updated_at)
|
|
VALUES ($1, $2, $3, now())
|
|
ON CONFLICT (room_id) DO UPDATE SET sync_cursor = EXCLUDED.sync_cursor, updated_at = now()`,
|
|
[room.roomId, room.workspaceId, cursor],
|
|
));
|
|
}
|
|
}
|
|
|
|
async claim(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise<OutboxRow> {
|
|
return this.withTenant(workspaceId, async (c) => {
|
|
await c.query(
|
|
`INSERT INTO archive_sync_outbox (workspace_id, room_id, event_id, claim_key)
|
|
SELECT $1, $2, $3, $4
|
|
WHERE NOT EXISTS (SELECT 1 FROM archive_sync_outbox WHERE workspace_id = $1 AND claim_key = $4)
|
|
ON CONFLICT DO NOTHING`,
|
|
[workspaceId, roomId, eventId, claimKey],
|
|
);
|
|
const res = await c.query<OutboxDbRow>(
|
|
`SELECT ${OUTBOX_COLUMNS} FROM archive_sync_outbox
|
|
WHERE workspace_id = $1 AND claim_key = $2 ORDER BY created_at LIMIT 1`,
|
|
[workspaceId, claimKey],
|
|
);
|
|
const row = res.rows[0];
|
|
if (row === undefined) throw new Error("outbox claim not visible after insert");
|
|
return toOutbox(row);
|
|
});
|
|
}
|
|
|
|
async commitArchived(record: ArchiveRecord, outboxId: string): Promise<{ id: string; indexedAt: Date }> {
|
|
return this.withTenant(record.workspaceId, async (c) => {
|
|
await c.query(
|
|
`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
|
|
) VALUES ($1, $2, $3, $4, 'matrix', $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
|
|
ON CONFLICT DO NOTHING`,
|
|
[
|
|
record.workspaceId, record.groupId, record.roomId, record.eventId, record.senderMxid,
|
|
JSON.stringify(record.matrixEventJson), record.originalName, record.safeName, record.mime,
|
|
record.mimeClass, record.byteSize, record.sha256, record.ocisResourceId,
|
|
record.ocisResourceVersion, record.createdAt,
|
|
],
|
|
);
|
|
const indexed = await c.query<{ id: string; indexed_at: Date }>(
|
|
`SELECT id, indexed_at FROM chat_media_archive
|
|
WHERE workspace_id = $1 AND room_id = $2 AND event_id = $3 LIMIT 1`,
|
|
[record.workspaceId, record.roomId, record.eventId],
|
|
);
|
|
const row = indexed.rows[0];
|
|
if (row === undefined) throw new Error("archive row not visible after insert");
|
|
await c.query(
|
|
`UPDATE archive_sync_outbox SET status = 'done', last_error = NULL, updated_at = now()
|
|
WHERE id = $1 AND workspace_id = $2`,
|
|
[outboxId, record.workspaceId],
|
|
);
|
|
return { id: row.id, indexedAt: row.indexed_at };
|
|
});
|
|
}
|
|
|
|
async recordFailure(workspaceId: string, outboxId: string, error: string, terminal: boolean): Promise<OutboxRow> {
|
|
return this.withTenant(workspaceId, async (c) => {
|
|
const res = await c.query<OutboxDbRow>(
|
|
`UPDATE archive_sync_outbox
|
|
SET attempts = attempts + 1, last_error = $3, updated_at = now(),
|
|
status = CASE WHEN $4 THEN 'failed' ELSE 'pending' END
|
|
WHERE id = $1 AND workspace_id = $2 AND status = 'pending'
|
|
RETURNING ${OUTBOX_COLUMNS}`,
|
|
[outboxId, workspaceId, error.slice(0, LAST_ERROR_MAX), terminal],
|
|
);
|
|
const row = res.rows[0];
|
|
if (row === undefined) throw new Error("outbox row not pending");
|
|
return toOutbox(row);
|
|
});
|
|
}
|
|
|
|
async markRedacted(workspaceId: string, roomId: string, eventId: string, claimKey: string): Promise<number> {
|
|
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
|
|
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'`,
|
|
[workspaceId, claimKey],
|
|
);
|
|
return res.rowCount ?? 0;
|
|
});
|
|
}
|
|
|
|
async countPending(workspaceIds: readonly string[]): Promise<number> {
|
|
let total = 0;
|
|
for (const workspaceId of workspaceIds) {
|
|
total += await this.withTenant(workspaceId, async (c) => {
|
|
const res = await c.query<{ n: string }>(
|
|
"SELECT count(*) AS n FROM archive_sync_outbox WHERE workspace_id = $1 AND status = 'pending'",
|
|
[workspaceId],
|
|
);
|
|
return Number(res.rows[0]?.n ?? 0);
|
|
});
|
|
}
|
|
return total;
|
|
}
|
|
|
|
async listRoom(workspaceId: string, roomId: string): Promise<IndexedFile[]> {
|
|
return this.withTenant(workspaceId, async (c) => {
|
|
const res = await c.query<RoomArchiveDbRow>(
|
|
`SELECT id, event_id, sender_mxid, original_name, safe_name, mime, mime_class,
|
|
byte_size, sha256, ocis_resource_id, created_at, available,
|
|
matrix_event_json->'content'->>'url' AS mxc_uri
|
|
FROM chat_media_archive
|
|
WHERE workspace_id = $1 AND room_id = $2
|
|
ORDER BY created_at, id`,
|
|
[workspaceId, roomId],
|
|
);
|
|
return res.rows.map(toIndexedFile);
|
|
});
|
|
}
|
|
|
|
private async withTenant<T>(workspaceId: string, fn: (client: PoolClient) => Promise<T>): Promise<T> {
|
|
const client = await this.pool.connect();
|
|
try {
|
|
await client.query("BEGIN");
|
|
await client.query("SELECT set_config('app.workspace_id', $1, true)", [workspaceId]);
|
|
const result = await fn(client);
|
|
await client.query("COMMIT");
|
|
return result;
|
|
} catch (err) {
|
|
await client.query("ROLLBACK").catch(() => undefined);
|
|
throw err;
|
|
} finally {
|
|
client.release();
|
|
}
|
|
}
|
|
}
|