123import { dirname, join } from "node:path";4import { blake3 } from "@noble/hashes/blake3.js";5import { bytesToHex } from "@noble/hashes/utils.js";6import { SQL } from "bun";7import { openBucket } from "../src/bucket/reader.ts";8import { checkRelease } from "../src/bucket/repo.ts";9import { compareVersions } from "../src/releases/releases.ts";10import { openKeprDb } from "../src/server/db.ts";1112type Row = Record<string, string | number | null>;13const env = (name: string) => process.env[name] ?? "";14const dry = process.argv.includes("--dry-run");15const pg = new SQL(env("KEPR2_DATABASE_URL"));16const kepr = openKeprDb(env("KEPR_DATA"));17const folder = env("KOH_FOLDER");18const VARS = "ENDPOINT BUCKET REGION ACCESS_KEY_ID SECRET_ACCESS_KEY".split(" ");19const [endpoint, name, region, accessKeyId, secretAccessKey] = VARS.map((v) => env(`KOH_S3_${v}`));20const s3 = { endpoint, region, accessKeyId, secretAccessKey };21const bucket = openBucket(folder ? { kind: "folder", path: folder } : { kind: "s3", name, ...s3 });2223async function put(key: string, body: string | Uint8Array): Promise<void> {24 const bytes = typeof body === "string" ? new TextEncoder().encode(body) : body;25 const had = (await bucket.get(key)) ?? (dry ? null : await write(key, bytes));26 console.log(had ? "have" : dry ? "would put" : "put", key);27 if (had && !Bun.deepEquals(had, bytes)) throw new Error(`${key} holds other bytes`);28}293031async function write(key: string, bytes: Uint8Array): Promise<Uint8Array | null> {32 if (folder) return Bun.write(join(folder, key), bytes).then(() => null);33 const url = new Bun.S3Client({ ...s3, bucket: name }).presign(key, { method: "PUT" });34 const res = await fetch(url, { method: "PUT", body: bytes, headers: { "If-None-Match": "*" } });35 if (res.ok || res.status === 412) return res.ok ? null : bucket.get(key);36 throw new Error(`put ${key}: ${res.status}`);37}38function insert(table: string, row: Row, tail: string): number | null {39 const cols = Object.keys(row);40 const sql = `INSERT INTO ${table} (${cols})41 VALUES (${cols.map((c) => `$${c}`)})42 ${tail}43 RETURNING rowid AS id`;44 const id = kepr.query<{ id: number }, [Row]>(sql).get(row)?.id ?? null;45 console.log(table, JSON.stringify(row).slice(0, 80), id ? "set" : "have");46 return id;47}48const errors: string[] = [],49 writes: [string, string | Uint8Array][] = [];50kepr.run("BEGIN");51const upsert = "ON CONFLICT DO UPDATE SET public = excluded.public, category = excluded.category";52for (const repo of await pg<Row[]>`53 SELECT54 name,55 (private = 0)::int AS public,56 trim(split_part(COALESCE(categories, ''), ',', 1)) AS category57 FROM repos`) {58 insert("repos", repo, upsert);59}60616263for (const { replies, ...i } of await pg<(Row & { replies: Row[] })[]>`64 SELECT65 i.repo,66 i.number,67 i.title,68 i.description AS body,69 CASE WHEN i.type IN ('bug', 'question') THEN i.type ELSE 'idea' END AS type,70 COALESCE(i.filed_ver, '') AS version,71 COALESCE(NULLIF(i.reporter_name, ''), '@' || a.username, '') AS reporter_name,72 COALESCE(NULLIF(i.contact_email, ''), i.contact, '') AS contact_email,73 i.filed_at,74 CASE75 WHEN i.status IN ('resolved', 'wontfix', 'closed') THEN 'closed'76 ELSE 'open'77 END AS status,78 CASE79 WHEN i.closed_save ~ '^[0-9a-f]{12}$' THEN i.closed_save80 ELSE ''81 END AS fixed_in,82 CASE83 WHEN i.status IN ('resolved', 'wontfix', 'closed') THEN COALESCE(i.closed_at, i.filed_at)84 END AS closed_at,85 (86 SELECT COALESCE(87 json_agg(88 json_build_object(89 'author',90 CASE91 WHEN e.account_id IS NULL92 OR e.account_id = 'operator'93 OR o.tier = 'operator'94 THEN 'operator'95 WHEN e.account_id = i.account_id THEN 'reporter'96 END,97 'body', e.body,98 'created_at', e.created_at99 )100 ORDER BY e.created_at, e.id101 ),102 '[]'103 )104 FROM issue_entries e105 LEFT JOIN accounts o ON o.id = e.account_id106 WHERE e.issue_id = i.id107 AND e.entry_type = 'reply'108 AND e.body IS NOT NULL109 ) AS replies110 FROM issues i111 LEFT JOIN accounts a ON a.id = i.account_id112 ORDER BY i.repo, i.number`) {113 const id = insert("issues", i, "ON CONFLICT DO NOTHING");114 if (replies.some((r) => !r.author)) errors.push(`${i.repo}#${i.number}: third-party reply`);115 else {116 for (const r of id ? replies : []) {117 const name = r.author === "reporter" ? i.reporter_name : "";118 insert("issue_entries", { issue_id: id, ...r, name }, "");119 }120 }121}122123124for (const count of await pg<Row[]>`125 SELECT126 repo_id AS repo,127 COUNT(*) AS count128 FROM engagement_events129 WHERE event_type = 'nix_install'130 AND repo_id IS NOT NULL131 GROUP BY repo_id`) {132 insert("installs", count, "ON CONFLICT DO NOTHING");133}134135136137type Release = { repo: string; version: string; hidden: boolean; record: Row; stored: string[] };138const newest = new Map<string, string>();139for (const { repo, version, hidden, record, stored } of await pg<Release[]>`140 SELECT141 a.repo_id AS repo,142 a.version,143 bool_or(r.builds_enabled = 0) AS hidden,144 json_agg(145 concat_ws(146 '/',147 'releases',148 a.repo_id,149 a.version,150 a.platform,151 a.filename152 )153 ORDER BY a.platform154 ) AS stored,155 json_build_object(156 'format', 1,157 'repo', a.repo_id,158 'version', a.version,159 'save', (array_agg(b.save_id ORDER BY b.started_at DESC))[1],160 'published_at', max(COALESCE(b.finished_at, a.built_at)),161 'machine', 'kepr2',162 'notes', COALESCE((array_agg(b.release_notes ORDER BY b.started_at DESC))[1], ''),163 'files',164 json_agg(165 json_build_object(166 'platform', a.platform,167 'name',168 CASE169 WHEN a.filename = a.repo_id THEN a.repo_id || '-' || a.platform170 ELSE a.filename171 END,172 'size', a.size_bytes,173 'blake3', a.blake3,174 'sha256', a.sha256175 )176 ORDER BY a.platform177 )178 ) AS record179 FROM release_artifacts a180 JOIN build_runs b ON b.id = a.build_run181 JOIN repos r ON r.name = a.repo_id182 WHERE b.status IN ('success', 'partial')183 GROUP BY a.repo_id, a.version184 ORDER BY a.repo_id, a.version`) {185 const prefix = `${repo}/releases/${version}`;186 try {187 for (const [n, f] of checkRelease(repo, version, record).files.entries()) {188 const path = join(env("KEPR2_DATA"), stored[n] ?? "");189 const file = Bun.file(path);190 const bytes = await ((await file.exists()) ? file : Bun.file(dirname(path))).bytes();191 if (bytesToHex(blake3(bytes)) !== f.blake3) throw new Error(`${f.platform}: hash mismatch`);192 writes.push([`${prefix}/${f.platform}/${f.name}`, bytes]);193 }194 writes.push([`${prefix}/release.json`, JSON.stringify(record)]);195 if (hidden) insert("release_settings", { repo, version, hidden: 1 }, "ON CONFLICT DO NOTHING");196 if (compareVersions(version, newest.get(repo) ?? "") > 0) newest.set(repo, version);197 } catch (error) {198 errors.push(`${prefix}: ${error instanceof Error ? error.message : error}`);199 }200}201202if (errors.length > 0) throw new Error(`not imported, nothing written:\n${errors.join("\n")}`);203for (const [repo, version] of newest) writes.push([`${repo}/releases/latest`, version]);204for (const [key, body] of writes) await put(key, body);205kepr.run(dry ? "ROLLBACK" : "END");206await pg.close();