12import { dirname, join } from "node:path";3import { blake3 } from "@noble/hashes/blake3.js";4import { bytesToHex } from "@noble/hashes/utils.js";5import { SQL } from "bun";6import { openBucket } from "../src/bucket/reader.ts";7import { checkRelease } from "../src/bucket/repo.ts";8import { compareVersions } from "../src/releases/releases.ts";9import { openKeprDb } from "../src/server/db.ts";1011type Row = Record<string, string | number | null>;12const env = (name: string) => process.env[name] ?? "";13const dry = process.argv.includes("--dry-run");14const pg = new SQL(env("KEPR2_DATABASE_URL"));15const kepr = openKeprDb(env("KEPR_DATA"));16const folder = env("KOH_FOLDER");17const VARS = "ENDPOINT BUCKET REGION ACCESS_KEY_ID SECRET_ACCESS_KEY".split(" ");18const [endpoint, name, region, accessKeyId, secretAccessKey] = VARS.map((v) => env(`KOH_S3_${v}`));19const s3 = { endpoint, region, accessKeyId, secretAccessKey };20const bucket = openBucket(folder ? { kind: "folder", path: folder } : { kind: "s3", name, ...s3 });2122async function put(key: string, body: string | Uint8Array): Promise<void> {23 const bytes = typeof body === "string" ? new TextEncoder().encode(body) : body;24 const had = (await bucket.get(key)) ?? (dry ? null : await write(key, bytes));25 console.log(had ? "have" : dry ? "would put" : "put", key);26 if (had && !Bun.deepEquals(had, bytes)) throw new Error(`${key} holds other bytes`);27}2829async function write(key: string, bytes: Uint8Array): Promise<Uint8Array | null> {30 if (folder) return Bun.write(join(folder, key), bytes).then(() => null);31 const url = new Bun.S3Client({ ...s3, bucket: name }).presign(key, { method: "PUT" });32 const res = await fetch(url, { method: "PUT", body: bytes, headers: { "If-None-Match": "*" } });33 if (res.ok || res.status === 412) return res.ok ? null : bucket.get(key);34 throw new Error(`put ${key}: ${res.status}`);35}36function insert(table: string, row: Row, tail: string): number | null {37 const cols = Object.keys(row);38 const sql = `INSERT INTO ${table} (${cols}) VALUES (${cols.map((c) => `$${c}`)}) ${tail} RETURNING rowid AS id`;39 const id = kepr.query<{ id: number }, [Row]>(sql).get(row)?.id ?? null;40 console.log(table, JSON.stringify(row).slice(0, 80), id ? "set" : "have");41 return id;42}43const errors: string[] = [],44 writes: [string, string | Uint8Array][] = [];45kepr.run("BEGIN");46const upsert = "ON CONFLICT DO UPDATE SET public = excluded.public, category = excluded.category";47for (const repo of await pg<Row[]>`48 SELECT name, (private = 0)::int AS public, trim(split_part(COALESCE(categories, ''), ',', 1)) AS category FROM repos`)49 insert("repos", repo, upsert);505152for (const { replies, ...i } of await pg<(Row & { replies: Row[] })[]>`53 SELECT i.repo, i.number, i.title, i.description AS body, CASE WHEN i.type IN ('bug', 'question') THEN i.type ELSE 'idea' END AS type,54 COALESCE(i.filed_ver, '') AS version, COALESCE(NULLIF(i.reporter_name, ''), '@' || a.username, '') AS reporter_name,55 COALESCE(NULLIF(i.contact_email, ''), i.contact, '') AS contact_email, i.filed_at,56 CASE WHEN i.status IN ('resolved', 'wontfix', 'closed') THEN 'closed' ELSE 'open' END AS status, CASE WHEN i.closed_save ~ '^[0-9a-f]{12}$' THEN i.closed_save ELSE '' END AS fixed_in,57 CASE WHEN i.status IN ('resolved', 'wontfix', 'closed') THEN COALESCE(i.closed_at, i.filed_at) END AS closed_at,58 (SELECT COALESCE(json_agg(json_build_object('author', CASE WHEN e.account_id IS NULL OR e.account_id = 'operator' OR o.tier = 'operator' THEN 'operator' WHEN e.account_id = i.account_id THEN 'reporter' END,59 'body', e.body, 'created_at', e.created_at) ORDER BY e.created_at, e.id), '[]') FROM issue_entries e LEFT JOIN accounts o ON o.id = e.account_id WHERE e.issue_id = i.id AND e.entry_type = 'reply' AND e.body IS NOT NULL) AS replies60 FROM issues i LEFT JOIN accounts a ON a.id = i.account_id ORDER BY i.repo, i.number`) {61 const id = insert("issues", i, "ON CONFLICT DO NOTHING");62 if (replies.some((r) => !r.author)) errors.push(`${i.repo}#${i.number}: third-party reply`);63 else for (const r of id ? replies : []) insert("issue_entries", { issue_id: id, ...r }, "");64}6566for (const count of await pg<Row[]>`67 SELECT repo_id AS repo, COUNT(*) AS count FROM engagement_events WHERE event_type = 'nix_install' AND repo_id IS NOT NULL GROUP BY repo_id`)68 insert("installs", count, "ON CONFLICT DO NOTHING");697071type Release = { repo: string; version: string; hidden: boolean; record: Row; stored: string[] };72const newest = new Map<string, string>();73for (const { repo, version, hidden, record, stored } of await pg<Release[]>`74 SELECT a.repo_id AS repo, a.version, bool_or(r.builds_enabled = 0) AS hidden, json_agg(concat_ws('/', 'releases', a.repo_id, a.version, a.platform, a.filename) ORDER BY a.platform) AS stored,75 json_build_object('format', 1, 'repo', a.repo_id, 'version', a.version, 'save', (array_agg(b.save_id ORDER BY b.started_at DESC))[1],76 'published_at', max(COALESCE(b.finished_at, a.built_at)), 'machine', 'kepr2', 'notes', COALESCE((array_agg(b.release_notes ORDER BY b.started_at DESC))[1], ''),77 'files', json_agg(json_build_object('platform', a.platform, 'name', CASE WHEN a.filename = a.repo_id THEN a.repo_id || '-' || a.platform ELSE a.filename END,78 'size', a.size_bytes, 'blake3', a.blake3, 'sha256', a.sha256) ORDER BY a.platform)) AS record79 FROM release_artifacts a JOIN build_runs b ON b.id = a.build_run JOIN repos r ON r.name = a.repo_id WHERE b.status IN ('success', 'partial') GROUP BY a.repo_id, a.version ORDER BY a.repo_id, a.version`) {80 const prefix = `${repo}/releases/${version}`;81 try {82 for (const [n, f] of checkRelease(repo, version, record).files.entries()) {83 const path = join(env("KEPR2_DATA"), stored[n] ?? "");84 const bytes = await Bun.file((await Bun.file(path).exists()) ? path : dirname(path)).bytes();85 if (bytesToHex(blake3(bytes)) !== f.blake3) throw new Error(`${f.platform}: hash mismatch`);86 writes.push([`${prefix}/${f.platform}/${f.name}`, bytes]);87 }88 writes.push([`${prefix}/release.json`, JSON.stringify(record)]);89 if (hidden) insert("release_settings", { repo, version, hidden: 1 }, "ON CONFLICT DO NOTHING");90 if (compareVersions(version, newest.get(repo) ?? "") > 0) newest.set(repo, version);91 } catch (error) {92 errors.push(`${prefix}: ${error instanceof Error ? error.message : error}`);93 }94}9596if (errors.length > 0) throw new Error(`not imported, nothing written:\n${errors.join("\n")}`);97for (const [repo, version] of newest) writes.push([`${repo}/releases/latest`, version]);98for (const [key, body] of writes) await put(key, body);99kepr.run(dry ? "ROLLBACK" : "COMMIT");100await pg.close();