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

import { once } from "node:events";
import { Readable, type Readable as NodeReadable } from "node:stream";

import type { ReadableStream as WebReadableStream } from "node:stream/web";
import * as zlib from "node:zlib";
import { extract, pack, type Pack } from "tar-stream";
import * as age from "age-encryption";
import { MANIFEST_PATH, PAYLOAD_DIR, type BundleManifest, type BundlePayload } from "./archive-index.js";

/** zstd level for bundle payloads; 3 ≈ gzip -6 speed with a better ratio. */
export const ZSTD_LEVEL = 3;

/** Native age recipient: lowercase bech32 `age1…` (X25519). */
const AGE_RECIPIENT = /^age1[02-9ac-hj-np-z]{58,}$/;

export class BundleError extends Error {}

/**
 * Parse and strictly validate `ARCHIVE_EXPORT_AGE_RECIPIENTS` (comma-separated
 * bech32 `age1…` public keys). Private keys never exist on this side: admins
 * generate keypairs and publish only the recipient in the room.
 */
export function parseAgeRecipients(raw: string): string[] {
  const recipients = raw.split(",").map((r) => r.trim()).filter((r) => r !== "");
  const seen = new Set<string>();
  recipients.forEach((recipient, index) => {
    if (!AGE_RECIPIENT.test(recipient)) {
      throw new BundleError(
        `invalid age recipient at index ${index} (want a bech32 age1… public key, got "${recipient.slice(0, 16)}…")`,
      );
    }
    if (seen.has(recipient)) throw new BundleError(`duplicate age recipient at index ${index}`);
    seen.add(recipient);
  });
  if (recipients.length === 0) throw new BundleError("at least one age recipient is required");
  return recipients;
}

function toBuffer(value: Uint8Array): Buffer {
  return Buffer.from(value.buffer, value.byteOffset, value.byteLength);
}

function errorOf(err: unknown): Error {
  return err instanceof Error ? err : new BundleError(String(err));
}

/**
 * tar(zstd) writer. Entry payloads stream into a `tar-stream` Pack piped
 * straight into zstd: bytes exist only in flight, never on local disk. Any
 * failure (payload fetch, tar, zstd) is rethrown from the next writer call and
 * destroys the zstd stream so a consumer sees the error instead of a truncated
 * "successful" bundle.
 */
export class TarZstdWriter {
  private readonly packer: Pack;
  private failure: unknown = null;

  private constructor(
    private readonly pack: Pack,
    readonly nodeStream: NodeReadable,
  ) {
    this.packer = pack;
    this.nodeStream.on("error", (err: unknown) => {
      this.failure ??= err;
    });
    this.pack.on("error", (err: unknown) => this.fail(err));
  }

  static create(level = ZSTD_LEVEL): TarZstdWriter {
    const packer = pack();
    const zstd = packer.pipe(zlib.createZstdCompress({ params: { [zlib.constants.ZSTD_c_compressionLevel]: level } }));
    return new TarZstdWriter(packer, zstd);
  }

  /** Copy one payload stream into a tar entry, honouring sink backpressure. */
  async writeEntry(name: string, body: ReadableStream<Uint8Array>, size: number): Promise<void> {
    this.assertNotFailed();
    const sink = this.pack.entry({ name, size });
    sink.on("error", (err: unknown) => this.fail(err));
    try {
      const reader = body.getReader();
      for (;;) {
        const { done, value } = await reader.read();
        if (done) break;
        if (!sink.write(toBuffer(value))) await once(sink, "drain");
      }
      sink.end(null);
      await once(sink, "finish");
    } catch (err) {
      this.fail(err);
      throw errorOf(err);
    }
    this.assertNotFailed();
  }

  async writeBytes(name: string, bytes: Uint8Array): Promise<void> {
    await this.writeEntry(name, new Blob([toBuffer(bytes)]).stream() as ReadableStream<Uint8Array>, bytes.byteLength);
  }

  /** Two zero blocks + zstd flush; resolves once the zstd stream has ended. */
  async finalize(): Promise<void> {
    this.assertNotFailed();
    this.pack.finalize();
    await once(this.nodeStream, "end");
    this.assertNotFailed();
  }

  /** Abort the pipeline after an unrecoverable error (e.g. a WebDAV abort). */
  abort(err: unknown): void {
    this.fail(err);
  }

  private fail(err: unknown): void {
    this.failure ??= err;
    this.nodeStream.destroy(errorOf(err));
  }

  private assertNotFailed(): void {
    if (this.failure !== null) throw errorOf(this.failure);
  }
}

interface TarEntry {
  name: string;
  stream: ReadableStream<Uint8Array>;
}

/**
 * tar(zstd) reader for the restore path. Yields `manifest.json` then every
 * `files/…` payload as a byte stream; unknown entries are drained and skipped.
 *
 * The pipeline is pull-driven: tar entries are only emitted as the consumer
 * iterates, so the age-decrypted source is read no faster than files are
 * written out.
 */
export async function* readTarZstd(tarZstd: ReadableStream<Uint8Array> | NodeReadable):
AsyncGenerator<TarEntry> {
  const source = tarZstd instanceof Readable
    ? tarZstd
    : Readable.fromWeb(tarZstd as WebReadableStream<Uint8Array>);
  const untar = extract();
  const finished = new Promise<void>((resolve) => {
    untar.on("finish", () => resolve());
    untar.on("close", () => resolve());
  });
  // Never awaited directly: an error surfaces as an iteration throw or via `finished`.
  source.on("error", (err: unknown) => untar.destroy(errorOf(err)));
  const decompressed = source.pipe(zlib.createZstdDecompress());
  decompressed.on("error", (err: unknown) => {
    untar.destroy(errorOf(err));
  });
  decompressed.pipe(untar);
  try {
    for await (const entry of untar) {
      const name = entry.header.name;
      if (name === MANIFEST_PATH || name.startsWith(`${PAYLOAD_DIR}/`)) {
        yield { name, stream: Readable.toWeb(Readable.from(entry)) as ReadableStream<Uint8Array> };
      } else {
        entry.resume();
      }
    }
    await finished;
  } finally {
    if (!untar.destroyed) untar.destroy();
  }
}

/**
 * Build the encrypted bundle stream: `manifest.json` plus one entry per
 * payload, tar → zstd → age. The age file key is wrapped for every recipient,
 * so any admin identity can open the bundle; none of them exists server-side.
 *
 * Returns the ciphertext stream. Payload assembly runs in the background and
 * any failure (payload fetch, tar, zstd) errors the ciphertext stream rather
 * than ending it, so a partial upload can never look like a complete export.
 */
export async function encryptBundleStream(
  payload: BundlePayload,
  recipients: readonly string[],
  level = ZSTD_LEVEL,
): Promise<ReadableStream<Uint8Array>> {
  const writer = TarZstdWriter.create(level);
  const encrypter = new age.Encrypter();
  for (const recipient of recipients) encrypter.addRecipient(recipient);
  // age computes and writes its header before any payload byte is needed.
  const ciphertext = await encrypter.encrypt(Readable.toWeb(writer.nodeStream) as ReadableStream<Uint8Array>);
  void (async () => {
    try {
      const manifest: BundleManifest = { ...payload.manifest, recipients: [...recipients] };
      await writer.writeBytes(MANIFEST_PATH, Buffer.from(JSON.stringify(manifest, null, 2), "utf8"));
      for (const [path, bytes] of payload.payloads) {
        await writer.writeEntry(path, new Blob([toBuffer(bytes)]).stream() as ReadableStream<Uint8Array>, bytes.byteLength);
      }
      await writer.finalize();
    } catch (err) {
      writer.abort(err);
      await ciphertext.cancel().catch(() => undefined);
    }
  })();
  return ciphertext;
}

/**
 * Encrypt one arbitrary byte stream with age. Used by the restorer's self-test
 * and any future single-object encryption need.
 */
export async function ageEncryptStream(
  source: ReadableStream<Uint8Array>,
  recipients: readonly string[],
): Promise<ReadableStream<Uint8Array>> {
  const encrypter = new age.Encrypter();
  for (const recipient of recipients) encrypter.addRecipient(recipient);
  return encrypter.encrypt(source);
}

/** Decrypt with `AGE-SECRET-KEY-1…` identities; fails closed on a bad header MAC. */
export async function ageDecryptStream(
  ciphertext: ReadableStream<Uint8Array>,
  identities: readonly string[],
): Promise<ReadableStream<Uint8Array>> {
  if (identities.length === 0) throw new BundleError("no age identities supplied");
  const decrypter = new age.Decrypter();
  for (const identity of identities) decrypter.addIdentity(identity);
  return decrypter.decrypt(ciphertext);
}

/** Fully buffer one stream (bounded use: manifests, test payloads). */
export async function drainToBuffer(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
  const chunks: Uint8Array[] = [];
  let total = 0;
  const reader = stream.getReader();
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    chunks.push(value);
    total += value.byteLength;
  }
  const out = new Uint8Array(total);
  let offset = 0;
  for (const chunk of chunks) {
    out.set(chunk, offset);
    offset += chunk.byteLength;
  }
  return out;
}