jobyssey/apps/worker/scripts/import-postings.ts

#!/usr/bin/env bun
// Turns the local jobsdb snapshot (documents/postings/jobs.db) into D1 import
// chunks: .import/0000-schema.sql (a copy of schema.sql) followed by
// .import/NNNN.sql files of INSERT OR REPLACE statements for companies,
// postings and postings_fts. Apply them in name order with
// `wrangler d1 execute jobyssey-postings --remote --file=<chunk>`; the script
// prints that loop, or runs it itself with --apply.
//
//   bun run scripts/import-postings.ts [path/to/jobs.db] [--apply]
//   JOBSDB_PATH=... bun run scripts/import-postings.ts
//
// Runs under Bun only (bun:sqlite); never imported by the Worker.
import { Database } from "bun:sqlite";
import { existsSync, mkdirSync, readdirSync, rmSync, writeFileSync } from "node:fs";
import { join, resolve } from "node:path";

const ROWS_PER_CHUNK = 2000;
// Multi-row VALUES keep the statement count (and wrangler's round trips) low;
// 50 rows stays well under D1's per-statement size limit for these ~350-byte rows.
const ROWS_PER_STATEMENT = 50;
const DATABASE_NAME = "jobyssey-postings";

const worker = resolve(import.meta.dir, "..");
const repo = resolve(worker, "../..");
const outDir = join(worker, ".import");

const args = process.argv.slice(2);
const apply = args.includes("--apply");
const dbArg = args.find((a) => !a.startsWith("--"));
const dbPath = resolve(dbArg ?? process.env.JOBSDB_PATH ?? join(repo, "documents/postings/jobs.db"));
if (!existsSync(dbPath)) {
  console.error(`database not found at ${dbPath} (pass a path or set JOBSDB_PATH)`);
  process.exit(1);
}

type Cell = string | number | bigint | null;

function literal(v: Cell): string {
  if (v === null) return "NULL";
  if (typeof v === "string") return `'${v.replace(/'/g, "''")}'`;
  return String(v);
}

/** One or more multi-row INSERT OR REPLACE statements for `rows`. */
function insertStatements(table: string, columns: string[], rows: Cell[][]): string[] {
  const head = `INSERT OR REPLACE INTO ${table} (${columns.join(", ")}) VALUES\n`;
  const out: string[] = [];
  for (let i = 0; i < rows.length; i += ROWS_PER_STATEMENT) {
    const values = rows.slice(i, i + ROWS_PER_STATEMENT).map((r) => `(${r.map(literal).join(", ")})`);
    out.push(`${head}${values.join(",\n")};\n`);
  }
  return out;
}

const POSTING_COLUMNS = [
  "posting_id",
  "company_id",
  "company_name",
  "company_domain",
  "title",
  "department",
  "function",
  "seniority",
  "posted_date",
  "location_text",
  "remote_policy_stated",
  "location_class",
  "location_country",
  "salary_min",
  "salary_max",
  "salary_currency",
  "url",
  "source_ats",
  "source_tier",
  "scrape_timestamp",
  "classification_rationale",
];
const COMPANY_COLUMNS = ["company_id", "name", "target_sector", "crypto_native", "priority_tier", "homepage_domain"];
const FTS_COLUMNS = ["rowid", "posting_id", "title", "company_name", "location_text"];

type PostingRow = Record<string, Cell> & { rid: number };
type CompanyRow = Record<string, Cell>;

const db = new Database(dbPath, { readonly: true });

// The source `companies` table has no homepage column; postings carry
// company_domain, so a company's domain is its postings' most common one.
const companies = db
  .query<CompanyRow, []>(
    `SELECT c.company_id, c.company_name AS name, c.target_sector, c.crypto_native, c.priority_tier,
       (SELECT p.company_domain FROM postings p
        WHERE p.company_id = c.company_id AND p.company_domain IS NOT NULL AND p.company_domain != ''
        GROUP BY p.company_domain ORDER BY count(*) DESC, p.company_domain LIMIT 1) AS homepage_domain
     FROM companies c ORDER BY c.company_id`,
  )
  .all();
// The snapshot re-scrapes postings across days, so posting_id repeats (~87k rows,
// ~76k ids). D1 keys on posting_id: keep each id's latest scrape only, so the
// FTS rows (one per kept source rowid) line up 1:1 with postings.
const postings = db
  .query<PostingRow, []>(
    `SELECT rid, ${POSTING_COLUMNS.join(", ")} FROM (
       SELECT rowid AS rid, *, row_number() OVER (PARTITION BY posting_id ORDER BY scrape_timestamp DESC, rowid DESC) AS rn
       FROM postings
     ) WHERE rn = 1 ORDER BY rid`,
  )
  .all();
db.close();

rmSync(outDir, { recursive: true, force: true });
mkdirSync(outDir, { recursive: true });
const files: string[] = [];
function emit(name: string, statements: string[]): void {
  const file = join(outDir, name);
  writeFileSync(file, statements.join("\n"));
  files.push(file);
}
let chunk = 0;
const chunkName = () => `${String(++chunk).padStart(4, "0")}.sql`;

emit("0000-schema.sql", [await Bun.file(join(worker, "schema.sql")).text()]);

for (let i = 0; i < companies.length; i += ROWS_PER_CHUNK) {
  const rows = companies.slice(i, i + ROWS_PER_CHUNK).map((c) => COMPANY_COLUMNS.map((col) => c[col] ?? null));
  emit(chunkName(), insertStatements("companies", COMPANY_COLUMNS, rows));
}

for (let i = 0; i < postings.length; i += ROWS_PER_CHUNK) {
  const slice = postings.slice(i, i + ROWS_PER_CHUNK);
  const rows = slice.map((p) => POSTING_COLUMNS.map((col) => p[col] ?? null));
  // The FTS row reuses the source rowid so a re-run replaces instead of duplicating.
  const fts = slice.map((p) => [p.rid, p.posting_id ?? null, p.title ?? null, p.company_name ?? null, p.location_text ?? null]);
  emit(chunkName(), [...insertStatements("postings", POSTING_COLUMNS, rows), ...insertStatements("postings_fts", FTS_COLUMNS, fts)]);
}

console.log(
  `wrote ${files.length} files to ${outDir}: schema + ${companies.length} companies + ${postings.length} postings (${ROWS_PER_CHUNK}/chunk)`,
);

const wrangler = join(worker, "node_modules/.bin/wrangler");
if (!apply) {
  console.log(`\nApply in order (from ${worker}):\n`);
  console.log(`  for f in .import/*.sql; do\n    ${wrangler} d1 execute ${DATABASE_NAME} --remote --yes --file="$f" || break\n  done\n`);
  console.log("or re-run with --apply to do that here.");
  process.exit(0);
}

for (const [n, file] of readdirSync(outDir)
  .filter((f) => f.endsWith(".sql"))
  .sort()
  .map((f) => join(outDir, f))
  .entries()) {
  console.log(`[${n + 1}/${files.length}] ${file}`);
  const proc = Bun.spawn([wrangler, "d1", "execute", DATABASE_NAME, "--remote", "--yes", `--file=${file}`], {
    cwd: worker,
    stdout: "inherit",
    stderr: "inherit",
  });
  const code = await proc.exited;
  if (code !== 0) {
    console.error(`wrangler exited ${code} on ${file}; fix and re-run --apply (chunks are INSERT OR REPLACE, so replaying is safe)`);
    process.exit(code);
  }
}
console.log("import complete");