Scaffold Pwned Passwords clone: React/wouter frontend, Koa/pg backend, import CLI

- Backend: Koa + @koa/router serving the HIBP-style /range/:prefix k-anonymity
  API, backed by Postgres via pg, with migrate/seed scripts run through tsx.
- Frontend: React + wouter, bundled with webpack (dev server + prod build),
  thin client that hashes passwords locally and only sends the hash prefix.
- Import package: CLI to bulk-load a wordlist or hash list into the database,
  plus a separate script to pull the full dataset from the HIBP range API.
  Both support parallelism (worker threads for wordlist hashing, concurrency
  limiter for DB/HTTP fan-out).
- Dev tooling: docker compose for a local Postgres, .env.local/.env.example,
  DATABASE_URL support as an alternative to discrete PG* vars.

packages/import/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "import",
+  "version": "1.0.0",
+  "description": "Bulk-loads a wordlist or hash list into the pwned_passwords table",
+  "type": "module",
+  "main": "dist/index.js",
+  "scripts": {
+    "start": "tsx --env-file=../../.env.local src/index.ts",
+    "import-hibp": "tsx --env-file=../../.env.local src/importHibp.ts",
+    "build": "tsc -p tsconfig.json",
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "license": "ISC",
+  "dependencies": {
+    "pg": "^8.22.0"
+  },
+  "devDependencies": {
+    "@types/pg": "^8.20.0"
+  }
+}

packages/import/src/concurrencyLimit.ts

@@ -0,0 +1,23 @@
+// Bounds how many async tasks run at once. Used both for fanning out
+// concurrent DB upserts and concurrent HTTP requests.
+export function createLimiter(concurrency: number) {
+  let active = 0;
+  const queue: Array<() => void> = [];
+
+  function release() {
+    active--;
+    queue.shift()?.();
+  }
+
+  return async function limit<T>(fn: () => Promise<T>): Promise<T> {
+    if (active >= concurrency) {
+      await new Promise<void>((resolve) => queue.push(resolve));
+    }
+    active++;
+    try {
+      return await fn();
+    } finally {
+      release();
+    }
+  };
+}

packages/import/src/db.ts

@@ -0,0 +1,20 @@
+import { Pool } from "pg";
+
+// Concurrent imports open several connections at once; bump via PGPOOL_MAX
+// if running with a high --jobs/--concurrency value.
+const max = process.env.PGPOOL_MAX ? Number(process.env.PGPOOL_MAX) : 20;
+
+// DATABASE_URL (e.g. postgres://user:pass@host:5432/db) takes precedence over
+// the discrete PG* vars, for connecting to an externally hosted database.
+export const pool = new Pool(
+  process.env.DATABASE_URL
+    ? { connectionString: process.env.DATABASE_URL, max }
+    : {
+        host: process.env.PGHOST,
+        port: process.env.PGPORT ? Number(process.env.PGPORT) : undefined,
+        user: process.env.PGUSER,
+        password: process.env.PGPASSWORD,
+        database: process.env.PGDATABASE,
+        max,
+      },
+);

packages/import/src/hashWorker.ts

@@ -0,0 +1,15 @@
+import { parentPort } from "node:worker_threads";
+import { createHash } from "node:crypto";
+import type { Entry } from "./upsertBatch.ts";
+
+function sha1(value: string): string {
+  return createHash("sha1").update(value).digest("hex").toUpperCase();
+}
+
+parentPort?.on("message", (lines: string[]) => {
+  const entries: Entry[] = lines.map((line) => {
+    const hash = sha1(line);
+    return { prefix: hash.slice(0, 5), suffix: hash.slice(5), count: 1 };
+  });
+  parentPort!.postMessage(entries);
+});

packages/import/src/importHibp.ts

@@ -0,0 +1,112 @@
+import { parseArgs } from "node:util";
+import { pool } from "./db.ts";
+import { upsertBatch, type Entry } from "./upsertBatch.ts";
+import { createLimiter } from "./concurrencyLimit.ts";
+
+// Mirrors the HaveIBeenPwned Pwned Passwords range API on the way in:
+// https://haveibeenpwned.com/API/v3#PwnedPasswords
+// There's no bulk file for anonymous use, so this walks all 16^5 = 1,048,576
+// 5-hex-character prefixes and pulls each range individually.
+const HIBP_RANGE_URL = "https://api.pwnedpasswords.com/range/";
+const PREFIX_COUNT = 0x100000; // 16^5
+const MAX_RETRIES = 5;
+
+function usage(): never {
+  console.error(
+    `Usage: npm run import-hibp -w import -- [options]\n\n` +
+      `Options:\n` +
+      `  --concurrency, -c   parallel HTTP requests to api.pwnedpasswords.com (default 16)\n` +
+      `  --start             starting prefix index, for resuming (default 0)\n` +
+      `  --end               ending prefix index, exclusive (default ${PREFIX_COUNT})\n`,
+  );
+  process.exit(1);
+}
+
+function prefixFromIndex(index: number): string {
+  return index.toString(16).toUpperCase().padStart(5, "0");
+}
+
+function parseRangeBody(prefix: string, body: string): Entry[] {
+  const entries: Entry[] = [];
+  for (const line of body.split(/\r?\n/)) {
+    if (!line) continue;
+    const [suffix, count] = line.split(":");
+    entries.push({ prefix, suffix, count: Number(count) });
+  }
+  return entries;
+}
+
+async function fetchRange(prefix: string): Promise<Entry[]> {
+  for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
+    const response = await fetch(`${HIBP_RANGE_URL}${prefix}`, {
+      headers: { "Add-Padding": "false" },
+    });
+
+    if (response.ok) {
+      return parseRangeBody(prefix, await response.text());
+    }
+
+    if (response.status === 429) {
+      const retryAfterSeconds = Number(response.headers.get("retry-after")) || 2 ** attempt;
+      console.warn(`rate limited on ${prefix}, retrying in ${retryAfterSeconds}s`);
+      await new Promise((resolve) => setTimeout(resolve, retryAfterSeconds * 1000));
+      continue;
+    }
+
+    throw new Error(`unexpected status ${response.status} for prefix ${prefix}`);
+  }
+
+  throw new Error(`giving up on prefix ${prefix} after ${MAX_RETRIES} retries`);
+}
+
+async function main() {
+  const { values } = parseArgs({
+    options: {
+      concurrency: { type: "string", short: "c", default: "16" },
+      start: { type: "string", default: "0" },
+      end: { type: "string" },
+    },
+  });
+
+  const concurrency = Number(values.concurrency);
+  const start = Number(values.start);
+  const end = values.end ? Number(values.end) : PREFIX_COUNT;
+  if (!Number.isInteger(start) || !Number.isInteger(end) || start < 0 || end > PREFIX_COUNT || start >= end) {
+    usage();
+  }
+
+  const limit = createLimiter(concurrency);
+  let completed = 0;
+  const total = end - start;
+  const startedAt = Date.now();
+
+  const tasks = [];
+  for (let i = start; i < end; i++) {
+    const prefix = prefixFromIndex(i);
+    tasks.push(
+      limit(async () => {
+        const entries = await fetchRange(prefix);
+        await upsertBatch(pool, entries);
+        completed++;
+        if (completed % 1000 === 0) {
+          const elapsedSeconds = (Date.now() - startedAt) / 1000;
+          const rate = completed / elapsedSeconds;
+          const etaMinutes = (total - completed) / rate / 60;
+          console.log(
+            `${completed.toLocaleString()}/${total.toLocaleString()} prefixes ` +
+              `(${rate.toFixed(0)}/s, ~${etaMinutes.toFixed(1)}m remaining)`,
+          );
+        }
+      }),
+    );
+  }
+
+  await Promise.all(tasks);
+  console.log(`done: imported ${total.toLocaleString()} prefixes`);
+  await pool.end();
+}
+
+main().catch((err) => {
+  console.error(err);
+  process.exitCode = 1;
+});

packages/import/src/index.ts

@@ -0,0 +1,155 @@
+import { createHash } from "node:crypto";
+import { availableParallelism } from "node:os";
+import { parseArgs } from "node:util";
+import { pool } from "./db.ts";
+import { readLines } from "./lines.ts";
+import { upsertBatch, type Entry } from "./upsertBatch.ts";
+import { createLimiter } from "./concurrencyLimit.ts";
+import { WorkerPool } from "./workerPool.ts";
+
+const HASH_LINE = /^([0-9A-Fa-f]{40})(?::(\d+))?$/;
+const CHUNK_SIZE = 2000;
+
+function usage(): never {
+  console.error(
+    `Usage: npm run start -w import -- <file> --format=<wordlist|hashes> [options]\n\n` +
+      `  wordlist        file contains one plaintext password per line\n` +
+      `  hashes          file contains one SHA-1 hash per line, optionally "HASH:COUNT"\n` +
+      `  file may be gzip-compressed (.gz)\n\n` +
+      `Options:\n` +
+      `  --batch-size    rows per upsert statement (default 5000)\n` +
+      `  --jobs, -j      parallelism: worker threads for wordlist hashing, or\n` +
+      `                  concurrent DB upserts for hashes (default: all CPU cores)\n`,
+  );
+  process.exit(1);
+}
+
+function sha1(value: string): string {
+  return createHash("sha1").update(value).digest("hex").toUpperCase();
+}
+
+function toEntry(hash: string, count: number): Entry {
+  return { prefix: hash.slice(0, 5), suffix: hash.slice(5), count };
+}
+
+function parseHashLine(line: string): Entry | null {
+  const match = HASH_LINE.exec(line.trim());
+  if (!match) return null;
+  const [, hash, count] = match;
+  return toEntry(hash.toUpperCase(), count ? Number(count) : 1);
+}
+
+// `.ts` when running under tsx, `.js` once built with tsc.
+const hashWorkerUrl = new URL(`./hashWorker.${import.meta.url.endsWith(".ts") ? "ts" : "js"}`, import.meta.url);
+
+async function importWordlist(file: string, batchSize: number, jobs: number) {
+  let processed = 0;
+
+  if (jobs <= 1) {
+    let batch: Entry[] = [];
+    for await (const line of readLines(file)) {
+      batch.push(toEntry(sha1(line), 1));
+      processed++;
+      if (batch.length >= batchSize) {
+        await upsertBatch(pool, batch);
+        batch = [];
+        console.log(`imported ${processed.toLocaleString()} lines`);
+      }
+    }
+    await upsertBatch(pool, batch);
+    return processed;
+  }
+
+  const workers = new WorkerPool<string[], Entry[]>(hashWorkerUrl, jobs);
+  const limit = createLimiter(jobs);
+  const inFlight: Promise<void>[] = [];
+
+  let chunk: string[] = [];
+  const submit = (lines: string[]) =>
+    limit(async () => {
+      const entries = await workers.run(lines);
+      await upsertBatch(pool, entries);
+      processed += lines.length;
+      console.log(`imported ${processed.toLocaleString()} lines`);
+    });
+
+  for await (const line of readLines(file)) {
+    chunk.push(line);
+    if (chunk.length >= CHUNK_SIZE) {
+      inFlight.push(submit(chunk));
+      chunk = [];
+    }
+  }
+  if (chunk.length > 0) inFlight.push(submit(chunk));
+
+  await Promise.all(inFlight);
+  await workers.close();
+  return processed;
+}
+
+async function importHashes(file: string, batchSize: number, jobs: number) {
+  const limit = createLimiter(Math.max(1, jobs));
+  const inFlight: Promise<void>[] = [];
+
+  let batch: Entry[] = [];
+  let processed = 0;
+  let skipped = 0;
+
+  const flush = (rows: Entry[]) => limit(() => upsertBatch(pool, rows));
+
+  for await (const line of readLines(file)) {
+    const entry = parseHashLine(line);
+    if (!entry) {
+      skipped++;
+      continue;
+    }
+
+    batch.push(entry);
+    processed++;
+
+    if (batch.length >= batchSize) {
+      inFlight.push(flush(batch));
+      batch = [];
+      console.log(`queued ${processed.toLocaleString()} lines (${skipped} skipped)`);
+    }
+  }
+  if (batch.length > 0) inFlight.push(flush(batch));
+
+  await Promise.all(inFlight);
+  return { processed, skipped };
+}
+
+async function main() {
+  const { positionals, values } = parseArgs({
+    allowPositionals: true,
+    options: {
+      format: { type: "string" },
+      "batch-size": { type: "string", default: "5000" },
+      jobs: { type: "string", short: "j" },
+    },
+  });
+
+  const [file] = positionals;
+  const format = values.format;
+  if (!file || (format !== "wordlist" && format !== "hashes")) {
+    usage();
+  }
+
+  const batchSize = Number(values["batch-size"]);
+  const jobs = values.jobs ? Number(values.jobs) : availableParallelism();
+
+  if (format === "wordlist") {
+    const processed = await importWordlist(file, batchSize, jobs);
+    console.log(`done: imported ${processed.toLocaleString()} lines`);
+  } else {
+    const { processed, skipped } = await importHashes(file, batchSize, jobs);
+    console.log(`done: imported ${processed.toLocaleString()} lines, skipped ${skipped}`);
+  }
+
+  await pool.end();
+}
+
+main().catch((err) => {
+  console.error(err);
+  process.exitCode = 1;
+});

packages/import/src/lines.ts

@@ -0,0 +1,17 @@
+import { createReadStream } from "node:fs";
+import { createGunzip } from "node:zlib";
+import { createInterface } from "node:readline";
+
+export async function* readLines(filePath: string): AsyncGenerator<string> {
+  let stream: NodeJS.ReadableStream = createReadStream(filePath);
+  if (filePath.endsWith(".gz")) {
+    stream = stream.pipe(createGunzip());
+  }
+
+  const rl = createInterface({ input: stream, crlfDelay: Infinity });
+  for await (const line of rl) {
+    if (line.length > 0) {
+      yield line;
+    }
+  }
+}

packages/import/src/upsertBatch.ts

@@ -0,0 +1,35 @@
+import type { Pool } from "pg";
+
+export interface Entry {
+  prefix: string;
+  suffix: string;
+  count: number;
+}
+
+export async function upsertBatch(pool: Pool, entries: Entry[]): Promise<void> {
+  if (entries.length === 0) return;
+
+  // Merge duplicate prefix/suffix pairs within the batch: ON CONFLICT can't
+  // touch the same row twice in one statement.
+  const merged = new Map<string, Entry>();
+  for (const entry of entries) {
+    const key = `${entry.prefix}:${entry.suffix}`;
+    const existing = merged.get(key);
+    if (existing) {
+      existing.count += entry.count;
+    } else {
+      merged.set(key, { ...entry });
+    }
+  }
+  // Concurrent upserts (--jobs > 1) may target overlapping rows across
+  // statements; a consistent lock order avoids deadlocks between them.
+  const rows = [...merged.values()].sort((a, b) => (a.prefix + a.suffix < b.prefix + b.suffix ? -1 : 1));
+
+  await pool.query(
+    `INSERT INTO pwned_passwords (prefix, suffix, count)
+     SELECT * FROM UNNEST($1::char(5)[], $2::char(35)[], $3::bigint[])
+     ON CONFLICT (prefix, suffix) DO UPDATE
+       SET count = pwned_passwords.count + EXCLUDED.count`,
+    [rows.map((r) => r.prefix), rows.map((r) => r.suffix), rows.map((r) => r.count)],
+  );
+}

packages/import/src/workerPool.ts

@@ -0,0 +1,55 @@
+import { Worker } from "node:worker_threads";
+
+interface Task<T, R> {
+  payload: T;
+  resolve: (result: R) => void;
+  reject: (err: unknown) => void;
+}
+
+// A fixed-size pool of worker_threads that each run `workerUrl`, used to move
+// CPU-bound work (hashing) off the main thread. Workers are stateless
+// request/response: send a payload, get a result back.
+export class WorkerPool<T, R> {
+  private workers: Worker[] = [];
+  private idle: Worker[] = [];
+  private queue: Task<T, R>[] = [];
+  private pending = new Map<Worker, Task<T, R>>();
+
+  constructor(workerUrl: URL, size: number) {
+    for (let i = 0; i < size; i++) {
+      const worker = new Worker(workerUrl);
+      worker.on("message", (result: R) => {
+        this.pending.get(worker)?.resolve(result);
+        this.pending.delete(worker);
+        this.idle.push(worker);
+        this.dispatch();
+      });
+      worker.on("error", (err) => {
+        this.pending.get(worker)?.reject(err);
+        this.pending.delete(worker);
+      });
+      this.workers.push(worker);
+      this.idle.push(worker);
+    }
+  }
+
+  run(payload: T): Promise<R> {
+    return new Promise((resolve, reject) => {
+      this.queue.push({ payload, resolve, reject });
+      this.dispatch();
+    });
+  }
+
+  private dispatch() {
+    while (this.idle.length > 0 && this.queue.length > 0) {
+      const worker = this.idle.shift()!;
+      const task = this.queue.shift()!;
+      this.pending.set(worker, task);
+      worker.postMessage(task.payload);
+    }
+  }
+
+  async close(): Promise<void> {
+    await Promise.all(this.workers.map((worker) => worker.terminate()));
+  }
+}

packages/import/tsconfig.json

@@ -0,0 +1,19 @@
+{
+  "compilerOptions": {
+    "target": "ES2023",
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "lib": ["ES2023"],
+    "outDir": "dist",
+    "rootDir": "src",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "forceConsistentCasingInFileNames": true,
+    "rewriteRelativeImportExtensions": true,
+    "resolveJsonModule": true,
+    "declaration": false,
+    "sourceMap": true
+  },
+  "include": ["src"]
+}