T-HUB/apps/archive-bot/src/exporter.ts

import type { Logger } from "pino";
import type { RoomMapping } from "./config.js";
import {
  newBundleId,
  payloadPath,
  type BundleManifest,
  type BundlePayload,
  type IndexedFile,
  type ManifestEntry,
} from "./archive-index.js";
import { BundleError, encryptBundleStream, parseAgeRecipients } from "./bundle.js";
import type { LibraryPort } from "./webdav.js";
import type { ArchiveIndexPort } from "./archive-index.js";
import { exportObjectKey, R2ExportStore } from "./r2-export.js";

/** Per-attachment budget for one bundle payload in RAM (bounded, like staging). */
export const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024;
/** Hard ceiling for one bundle's total payload bytes (100 × max attachment). */
export const MAX_BUNDLE_BYTES = 100 * MAX_ATTACHMENT_BYTES;

export class ExportError extends Error {}

export interface ExportResult {
  objectKey: string;
  fileCount: number;
  totalBytes: number;
  byteSize: number;
}

/** Admin gate: power level 50+ in the room (spec default moderator). */
export const ADMIN_POWER_LEVEL = 50;

export function exportNotice(objectKey: string, bucket: string, recipients: readonly string[]): Record<string, unknown> {
  return {
    msgtype: "m.notice",
    body: `Encrypted archive bundle uploaded: ${objectKey} (${bucket}) — ${recipients.length} age recipient(s). `
      + "Ciphertext only; open it with the admin age identity from this room.",
  };
}

/** `!archive export` from a room admin. */
export function isExportCommand(content: Record<string, unknown>): boolean {
  const body = typeof content.body === "string" ? content.body.trim() : "";
  return body === "!archive export" || body.startsWith("!archive export ");
}

/**
 * Phase-B exporter: builds an encrypted bundle of a room's archived media and
 * uploads the ciphertext to R2. Payloads stream oCIS → tar → zstd → age; only
 * the ciphertext spill file (see R2ExportStore) ever touches disk.
 */
export class BundleExporter {
  private readonly recipients: string[];
  private readonly r2: R2ExportStore;
  /** room → attachments archived since the last successful export */
  private readonly counters = new Map<string, number>();
  private readonly libraryDir: (mapping: RoomMapping) => string[];

  constructor(
    private readonly index: ArchiveIndexPort,
    private readonly library: LibraryPort,
    private readonly upload: R2ExportStore,
    recipients: readonly string[],
    private readonly bucket: string,
    private readonly logger: Logger,
    private readonly now: () => Date = () => new Date(),
  ) {
    this.recipients = parseAgeRecipients(recipients.join(","));
    this.r2 = upload;
    this.bucket = bucket;
    this.libraryDir = (mapping) => ["Library", mapping.workspaceId, mapping.groupId];
  }

  /** Count one archived attachment toward the room's rolling auto-export. */
  countAttachment(roomId: string): void {
    this.counters.set(roomId, (this.counters.get(roomId) ?? 0) + 1);
  }

  /** True (and resets) when the room crosses ARCHIVE_EXPORT_EVERY. */
  shouldAutoExport(roomId: string, every: number): boolean {
    if (every <= 0) return false;
    const count = this.counters.get(roomId) ?? 0;
    if (count < every) return false;
    this.counters.set(roomId, 0);
    return true;
  }

  /**
   * Build and upload one bundle. Reads every archive row for the room (payload
   * streams come from oCIS WebDAV), wraps the file key for all recipients and
   * PUTs the ciphertext under `exports/<room-id>/<timestamp>.tar.zst.age`.
   */
  async exportRoom(
    mapping: RoomMapping,
    signal: AbortSignal,
    reason: "manual" | "auto",
  ): Promise<ExportResult> {
    const started = this.now();
    const rows = await this.index.listRoom(mapping.workspaceId, mapping.roomId);
    const manifest = buildManifest(mapping, rows, started, this.recipients);
    const payload = await this.collectPayloads(mapping, rows, manifest);

    const objectKey = exportObjectKey(mapping.roomId, started);
    const ciphertext = await encryptBundleStream(payload, this.recipients);
    await this.r2.putBundle(objectKey, ciphertext, signal);
    this.logger.info(
      { roomId: mapping.roomId, objectKey, files: manifest.fileCount, totalBytes: manifest.totalBytes, reason },
      "archive-export: bundle uploaded",
    );
    return { objectKey, fileCount: manifest.fileCount, totalBytes: manifest.totalBytes, byteSize: manifest.totalBytes };
  }

  /** Stream every available payload out of oCIS; unavailable rows contribute metadata only. */
  private async collectPayloads(mapping: RoomMapping, rows: IndexedFile[], manifest: BundleManifest): Promise<BundlePayload> {
    const payloads = new Map<string, Uint8Array>();
    let total = 0;
    const dir = this.libraryDir(mapping);
    for (const row of rows) {
      if (!row.available) continue;
      if (row.byteSize > MAX_ATTACHMENT_BYTES) {
        throw new ExportError(`attachment ${row.eventId} exceeds the ${MAX_ATTACHMENT_BYTES} byte bundle budget`);
      }
      if (total + row.byteSize > MAX_BUNDLE_BYTES) {
        throw new ExportError(`bundle for ${mapping.roomId} exceeds ${MAX_BUNDLE_BYTES} bytes; raise ARCHIVE_EXPORT_EVERY`);
      }
      const stream = await this.library.getResource(dir, `${row.sha256}-${row.safeName}`, AbortSignal.timeout(120_000));
      const bytes = await readAll(stream, row.byteSize);
      if (bytes.byteLength !== row.byteSize) {
        throw new ExportError(`payload ${row.eventId} size mismatch (want ${row.byteSize}, got ${bytes.byteLength})`);
      }
      payloads.set(payloadPath(row), bytes);
      total += row.byteSize;
    }
    return { manifest, payloads };
  }
}

/** Manifest header; `files` follows from the index rows. */
export function buildManifest(
  mapping: RoomMapping,
  rows: IndexedFile[],
  exportedAt: Date,
  recipients: readonly string[],
): BundleManifest {
  const files: ManifestEntry[] = rows.map((row) => ({
    eventId: row.eventId,
    mxcUri: row.mxcUri,
    originalName: row.originalName,
    safeName: row.safeName,
    mime: row.mime,
    mimeClass: row.mimeClass,
    byteSize: row.byteSize,
    sha256: row.sha256,
    sender: row.sender,
    createdAt: row.createdAt,
    ocisResourceId: row.ocisResourceId,
    available: row.available,
  }));
  const available = files.filter((f) => f.available);
  return {
    format: "chat-media-archive-bundle",
    version: 1,
    roomId: mapping.roomId,
    workspaceId: mapping.workspaceId,
    groupId: mapping.groupId,
    exportedAt: exportedAt.toISOString(),
    objectKey: null,
    recipients: [...recipients],
    fileCount: available.length,
    totalBytes: available.reduce((n, f) => n + f.byteSize, 0),
    files,
  };
}

/** Read one payload stream fully, enforcing the declared size (bounded RAM). */
async function readAll(stream: ReadableStream<Uint8Array>, expected: number): Promise<Uint8Array> {
  const chunks: Uint8Array[] = [];
  let total = 0;
  const reader = stream.getReader();
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    total += value.byteLength;
    if (total > expected) {
      await reader.cancel().catch(() => undefined);
      throw new ExportError(`payload longer than the indexed size (${expected})`);
    }
    chunks.push(value);
  }
  const out = new Uint8Array(total);
  let offset = 0;
  for (const chunk of chunks) {
    out.set(chunk, offset);
    offset += chunk.byteLength;
  }
  return out;
}

export { BundleError, newBundleId };