glass beadian
2cb77ae6b4
- archive_sync_state keyed by room per migration 029 (drop synthetic id cursor) - putStaging buffers the stream and sends content-length (oCIS rejects chunked) - mimeClass passthrough from sniffed payload - edge-firewall: RETURN docker-bridge ingress before the CF-range catch-all so container egress (Matrix/R2/platform) survives the 80/443 source filter - files scaffold: keep the ocdav data-gateway hop on-loopback
91 lines
3.4 KiB
TypeScript
91 lines
3.4 KiB
TypeScript
import type { GroupFile } from "@tactical-media-broadcast-hub/core";
|
||
import { toGroupFileMapping, type RoomMapping } from "./config.js";
|
||
import type { AttachmentWork } from "./events.js";
|
||
import { inspectStream } from "./inspector.js";
|
||
import { replyTxnId } from "./keys.js";
|
||
import type { MatrixPort } from "./matrix.js";
|
||
import { safeFileName } from "./names.js";
|
||
import type { ArchiveStore, OutboxRow } from "./store.js";
|
||
import type { LibraryPort } from "./webdav.js";
|
||
|
||
export interface ArchivedAttachment {
|
||
file: GroupFile;
|
||
safeName: string;
|
||
}
|
||
|
||
/**
|
||
* Steps 2–3 of the processing contract for one claimed attachment:
|
||
* authenticated download → in-flight size/hash/MIME → staged WebDAV PUT →
|
||
* content-addressed MOVE → index row + outbox done in one transaction.
|
||
* The payload is never written to local disk.
|
||
*/
|
||
export class Archiver {
|
||
constructor(
|
||
private readonly matrix: MatrixPort,
|
||
private readonly library: LibraryPort,
|
||
private readonly store: ArchiveStore,
|
||
) {}
|
||
|
||
async archive(
|
||
mapping: RoomMapping,
|
||
work: AttachmentWork,
|
||
outbox: OutboxRow,
|
||
claimKey: string,
|
||
signal: AbortSignal,
|
||
): Promise<ArchivedAttachment> {
|
||
const { candidate, event } = work;
|
||
const download = await this.matrix.downloadMedia(candidate.mxcUri, signal);
|
||
const { stream, result } = inspectStream(download.body, download.contentLength);
|
||
const dir = ["Library", mapping.workspaceId, mapping.groupId];
|
||
// Deterministic per claim: a retried upload overwrites its own partial staging copy.
|
||
const stagingName = `${replyTxnId(claimKey)}.part`;
|
||
await this.library.putStaging(dir, stagingName, stream, signal);
|
||
const payload = await result;
|
||
|
||
const safeName = safeFileName(work.displayName, payload.sniffed.extension);
|
||
const resource = await this.library.commit(dir, stagingName, `${payload.sha256}-${safeName}`, signal);
|
||
const createdAt = new Date(event.origin_server_ts ?? Date.now());
|
||
const indexed = await this.store.commitArchived({
|
||
workspaceId: mapping.workspaceId,
|
||
groupId: mapping.groupId,
|
||
roomId: mapping.roomId,
|
||
eventId: event.event_id,
|
||
senderMxid: event.sender,
|
||
matrixEventJson: event,
|
||
originalName: work.displayName,
|
||
safeName,
|
||
mime: payload.sniffed.mime,
|
||
mimeClass: payload.sniffed.mimeClass,
|
||
byteSize: payload.byteSize,
|
||
sha256: payload.sha256,
|
||
ocisResourceId: resource.resourceId,
|
||
ocisResourceVersion: resource.version,
|
||
createdAt,
|
||
}, outbox.id);
|
||
|
||
const group = toGroupFileMapping(mapping);
|
||
return {
|
||
safeName,
|
||
file: {
|
||
id: indexed.id,
|
||
workspaceId: mapping.workspaceId,
|
||
groupId: mapping.groupId,
|
||
roomId: mapping.roomId,
|
||
senderId: event.sender,
|
||
eventId: event.event_id,
|
||
origin: { kind: "matrix", mxcUri: candidate.mxcUri },
|
||
name: safeName,
|
||
mimeType: payload.sniffed.mime,
|
||
mimeClass: payload.sniffed.mimeClass,
|
||
sizeBytes: payload.byteSize,
|
||
checksum: { algorithm: "sha256", value: payload.sha256 },
|
||
createdAt: createdAt.toISOString(),
|
||
indexedAt: indexed.indexedAt.toISOString(),
|
||
file: { spaceId: group.spaceId, resourceId: resource.resourceId, version: resource.version },
|
||
shareLink: { url: resource.url, access: "members", expiresAt: null },
|
||
retention: group.retention,
|
||
status: "available",
|
||
},
|
||
};
|
||
}
|
||
}
|