1// Imports: the reverse of dumps. PUT /import takes a file of events, a2// JSONL (strfry's export is one) or a JSON array, keeps it in R2 under3// <name>/imports/, and hands it to a job that reads it back in bounded4// byte ranges, one round at a time, so a big file never has to fit in a5// request. Signatures are checked, bans and kind rules apply, the write6// rule does not: the owner asked for these. The file counts as media while7// it exists and is deleted when the job is done.8import { validate, now, type Event } from "./event.ts";9import { verifyNIP98 } from "./auth.ts";10import { can } from "./roles.ts";11import { ERR_DUPLICATE } from "./store.ts";12import { newJobID, type Job } from "./jobs.ts";13import type { Relay } from "./relay.ts";1415// The door reads the body whole to check the NIP-98 payload hash, so the16// cap is what an object can hold comfortably next to its own work.17export const IMPORT_MAX_BYTES = 64 * 1024 * 1024;18const CHUNK = 1024 * 1024; // bytes read from R2 per round19const MAX_LINE = 512 * 1024; // a line longer than this is not an event2021export function importKey(slug: string, id: string): string {22 return `${slug}/imports/${id}.jsonl`;23}2425export function importBytes(sql: SqlStorage): number {26 return sql.exec<{ n: number | null }>(`SELECT sum(bytes) AS n FROM imports`).one().n ?? 0;27}2829// importUpload serves PUT /import for a NIP-98 signer with the storage30// action. The body is JSONL or a JSON array; arrays are rewritten to JSONL31// so the job has one shape to read.32export async function importUpload(relay: Relay, req: Request): Promise<Response> {33 const json = (b: unknown, status: number) => new Response(JSON.stringify(b), { status, headers: { "content-type": "application/json", "access-control-allow-origin": "*" } });34 const declared = Number(req.headers.get("content-length"));35 if (declared > IMPORT_MAX_BYTES) return json({ error: `invalid: at most ${IMPORT_MAX_BYTES / 1024 / 1024} MB per import` }, 413);36 const body = await req.text();37 if (body.length > IMPORT_MAX_BYTES) return json({ error: `invalid: at most ${IMPORT_MAX_BYTES / 1024 / 1024} MB per import` }, 413);38 const auth = verifyNIP98(req.headers.get("authorization") ?? "", req.url, req.method, body);39 if (typeof auth === "string") return json({ error: auth }, 401);40 if (!can(relay.settings.roleOf(auth.pubkey), "storage")) return json({ error: "restricted: not the relay owner" }, 403);41 if (relay.settings.policy.owner === "") return json({ error: "restricted: this relay is unclaimed" }, 403);42 if (relay.fuelStatus().outOfFuel) return json({ error: "restricted: this relay is out of fuel" }, 403);43 let text = body.trim();44 if (text === "") return json({ error: "invalid: the file is empty" }, 400);45 if (text.startsWith("[")) {46 let arr: unknown;47 try {48 arr = JSON.parse(text);49 } catch {50 return json({ error: "invalid: not a JSON array" }, 400);51 }52 if (!Array.isArray(arr)) return json({ error: "invalid: not a JSON array" }, 400);53 text = arr.map((e) => JSON.stringify(e)).join("\n");54 }55 const bytes = new TextEncoder().encode(text + "\n");56 const id = newJobID();57 const name = (new URL(req.url).searchParams.get("name") ?? "import.jsonl").replace(/[^\w.-]/g, "_").slice(0, 80) || "import.jsonl";58 await relay.media.put(importKey(relay.slug, id), bytes, { httpMetadata: { contentType: "application/x-ndjson" } });59 relay.sql.exec(`INSERT INTO imports(id,name,bytes,at) VALUES(?,?,?,?)`, id, name, bytes.length, now());60 relay.meterBytes(bytes.length, 0);61 const job = await relay.addChecked({ kind: "import", label: "import", relays: [name], filter: {}, every: 0, running: false, startedAt: 0, rounds: 0, failures: 0, relayIndex: 0, cursor: 0, stored: 0, skipped: 0, blobs: 0, sent: 0, refused: 0, duplicates: 0, last: null, object: id, size: bytes.length, carry: "" });62 if (typeof job === "string") {63 await forgetImport(relay, id);64 return json({ error: job }, 409);65 }66 return json({ job: job.id, bytes: bytes.length, name }, 202);67}6869// forgetImport drops the object and its row, whatever state the job is in.70export async function forgetImport(relay: Relay, id: string) {71 await relay.media.delete(importKey(relay.slug, id)).catch(() => {});72 relay.sql.exec(`DELETE FROM imports WHERE id=?`, id);73}7475const b64 = {76 enc: (b: Uint8Array) => btoa(String.fromCharCode(...b)),77 dec: (s: string) => Uint8Array.from(atob(s), (c) => c.charCodeAt(0)),78};7980// runImportRound reads the next byte range, finishes the line carried over81// from the last round, and stores what checks out. more: call again.82export async function runImportRound(relay: Relay, job: Job): Promise<{ more: boolean; error: string }> {83 job.rounds++;84 const id = job.object ?? "";85 const size = job.size ?? 0;86 if (!id || job.cursor >= size) {87 await forgetImport(relay, id);88 return { more: false, error: "" };89 }90 const obj = await relay.media.get(importKey(relay.slug, id), { range: { offset: job.cursor, length: Math.min(CHUNK, size - job.cursor) } });91 if (!obj) return { more: false, error: "the import file is gone" };92 const fresh = new Uint8Array(await obj.arrayBuffer());93 const carried = job.carry ? b64.dec(job.carry) : new Uint8Array(0);94 const buf = new Uint8Array(carried.length + fresh.length);95 buf.set(carried);96 buf.set(fresh, carried.length);97 job.cursor += fresh.length;98 const atEnd = job.cursor >= size;99 // Split on newlines; the tail after the last one waits for the next round.100 let start = 0;101 const dec = new TextDecoder();102 const t = now();103 for (let i = 0; i < buf.length; i++) {104 if (buf[i] !== 10) continue;105 take(relay, dec.decode(buf.subarray(start, i)), job, t);106 start = i + 1;107 }108 const tail = buf.subarray(start);109 if (atEnd) {110 if (tail.length) take(relay, dec.decode(tail), job, t);111 job.carry = "";112 await forgetImport(relay, id);113 return { more: false, error: "" };114 }115 if (tail.length > MAX_LINE) {116 job.skipped++;117 job.carry = "";118 } else job.carry = b64.enc(tail);119 return { more: true, error: "" };120}121122// take stores one line, counting what became of it.123function take(relay: Relay, line: string, job: Job, t: number) {124 const s = line.trim();125 if (s === "" || s === "[" || s === "]") return;126 let e: Event;127 try {128 e = JSON.parse(s.replace(/,$/, "")) as Event;129 } catch {130 job.skipped++;131 return;132 }133 if (validate(e) || !relay.settings.kindAllowed(e.kind)) {134 job.skipped++;135 return;136 }137 const r = relay.accept(e, null);138 if (r.stored) {139 job.stored++;140 relay.broadcast(e, false);141 } else if (r.msg === ERR_DUPLICATE) job.duplicates = (job.duplicates ?? 0) + 1;142 else job.skipped++;143 void t;144}