116 lines
4.8 KiB
TypeScript
116 lines
4.8 KiB
TypeScript
import { createWriteStream } from "node:fs";
|
|
import { open, stat, unlink } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { randomBytes } from "node:crypto";
|
|
import { once } from "node:events";
|
|
import { createReadStream } from "node:fs";
|
|
import { PutObjectCommand, S3Client, S3ServiceException } from "@aws-sdk/client-s3";
|
|
|
|
export const EXPORT_BUCKET = "hub-matrix-media";
|
|
export const EXPORT_PREFIX = "exports";
|
|
/** Default bounded spill directory (tmpfs-style; see README "Spill file"). */
|
|
export const DEFAULT_SPILL_DIR = join(tmpdir(), "archive-bot-export");
|
|
|
|
export class R2ExportError extends Error {}
|
|
|
|
export interface R2ExportOptions {
|
|
/** R2 S3 endpoint, e.g. `https://<account>.r2.cloudflarestorage.com`. */
|
|
endpoint: string;
|
|
region: string;
|
|
accessKeyId: string;
|
|
secretAccessKey: string;
|
|
bucket?: string;
|
|
/** Bounded spill dir for the ciphertext before the PUT (see README). */
|
|
spillDir?: string;
|
|
/** Test seam ONLY: allow plain-http local endpoints. Never set in production. */
|
|
allowInsecureEndpoint?: boolean;
|
|
}
|
|
|
|
/** `exports/<room-id>/<timestamp>.tar.zst.age` — the documented key contract. */
|
|
export function exportObjectKey(roomId: string, timestamp: Date): string {
|
|
const iso = timestamp.toISOString().replace(/[-:]/g, "").replace(/\.\d+Z$/, "Z");
|
|
return `${EXPORT_PREFIX}/${roomId}/${iso}.tar.zst.age`;
|
|
}
|
|
|
|
/**
|
|
* Minimal S3/R2 sink for export bundles. PUT-only by design: the credentials
|
|
* are the tuwunel media-provider key, so this class never lists, reads or
|
|
* deletes — it only ever creates ciphertext objects.
|
|
*/
|
|
export class R2ExportStore {
|
|
private readonly client: S3Client;
|
|
private readonly bucket: string;
|
|
private readonly spillDir: string;
|
|
|
|
constructor(opts: R2ExportOptions) {
|
|
let endpoint: URL;
|
|
try {
|
|
endpoint = new URL(opts.endpoint);
|
|
} catch {
|
|
throw new R2ExportError(`R2 endpoint must be an absolute URL, got "${opts.endpoint}"`);
|
|
}
|
|
if (endpoint.protocol !== "https:" && opts.allowInsecureEndpoint !== true) {
|
|
throw new R2ExportError("R2 endpoint must be https");
|
|
}
|
|
this.client = new S3Client({
|
|
endpoint: opts.endpoint,
|
|
region: opts.region,
|
|
credentials: { accessKeyId: opts.accessKeyId, secretAccessKey: opts.secretAccessKey },
|
|
forcePathStyle: true,
|
|
// R2 keys: no flexible checksums; keeps the PUT a single signed request.
|
|
requestChecksumCalculation: "WHEN_REQUIRED",
|
|
responseChecksumValidation: "WHEN_REQUIRED",
|
|
});
|
|
this.bucket = opts.bucket ?? EXPORT_BUCKET;
|
|
this.spillDir = opts.spillDir ?? DEFAULT_SPILL_DIR;
|
|
}
|
|
|
|
/**
|
|
* Upload one ciphertext bundle under `key`.
|
|
*
|
|
* The final object length only exists once zstd+age finish, so the stream is
|
|
* spilled to ONE bounded temp file (mode 0600, O_EXCL) and PUT with an
|
|
* explicit Content-Length, then unlinked in `finally`. This is the single
|
|
* documented disk touch in the export path — it is ciphertext only, never a
|
|
* payload cache; mount `ARCHIVE_EXPORT_TMP_DIR` on tmpfs if disk writes are
|
|
* unacceptable (see README).
|
|
*/
|
|
async putBundle(key: string, ciphertext: ReadableStream<Uint8Array>, signal: AbortSignal): Promise<void> {
|
|
const spill = join(this.spillDir, `${randomBytes(16).toString("hex")}.tar.zst.age`);
|
|
const marker = await open(spill, "wx", 0o600);
|
|
await marker.close();
|
|
const out = createWriteStream(spill, { flags: "w", mode: 0o600 });
|
|
const reader = ciphertext.getReader();
|
|
try {
|
|
for (;;) {
|
|
if (signal.aborted) throw new R2ExportError("export aborted");
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
if (!out.write(Buffer.from(value.buffer, value.byteOffset, value.byteLength))) await once(out, "drain");
|
|
}
|
|
await new Promise<void>((resolve, reject) => out.end((err?: Error | null) => (err === null || err === undefined ? resolve() : reject(err))));
|
|
const size = (await stat(spill)).size;
|
|
await this.client.send(new PutObjectCommand({
|
|
Bucket: this.bucket,
|
|
Key: key,
|
|
Body: createReadStream(spill),
|
|
ContentLength: size,
|
|
ContentType: "application/octet-stream",
|
|
}), { abortSignal: signal });
|
|
} catch (err) {
|
|
throw err instanceof R2ExportError ? err : toR2Error(err);
|
|
} finally {
|
|
await reader.cancel().catch(() => undefined);
|
|
await new Promise<void>((resolve) => out.close(() => resolve())).catch(() => undefined);
|
|
await unlink(spill).catch(() => undefined);
|
|
}
|
|
}
|
|
}
|
|
|
|
function toR2Error(err: unknown): R2ExportError {
|
|
if (err instanceof S3ServiceException) {
|
|
return new R2ExportError(`R2 upload failed: ${err.$metadata.httpStatusCode ?? "HTTP"} ${err.name}`);
|
|
}
|
|
return err instanceof Error ? new R2ExportError(err.message) : new R2ExportError(String(err));
|
|
}
|