DDRCompanion/lib/db/db.worker.ts

import sqlWasmUrl from "@jlongster/sql.js/dist/sql-wasm.wasm?url";
import { MigrateDeploy } from "@prisma/migrate";
import initSqlJs from "@jlongster/sql.js";
import { SQLiteFS } from "absurd-sql";
import type { BindParams, Database, Statement } from "sql.js";
import IndexedDBBackend from "absurd-sql/dist/indexeddb-backend";
import { RpcProvider } from "worker-rpc";
import executeMigrations from "./migrations";
import { DbStatusCode } from "./constants";

async function init() {
  const rpcProvider = new RpcProvider((message, transfer) =>
    self.postMessage(message, undefined, transfer),
  );
  self.addEventListener("message", (evt) => rpcProvider.dispatch(evt.data));

  const preparedStatementHandles = new Map<number, Statement>();
  let ctr = 0;
  const fresh = () => {
    return ctr++;
  };

  const SQL = await initSqlJs({
    locateFile: (file) => {
      switch (file) {
        case "sql-wasm.wasm":
          return sqlWasmUrl;
      }
      return file;
    },
  });

  rpcProvider.signal("db", { status: DbStatusCode.LOADED_SQLITE });

  console.log("new SQLITEFS", SQLiteFS, SQL, IndexedDBBackend);
  const sqlFS = new SQLiteFS(SQL.FS, new IndexedDBBackend());
  console.log("REGISTER FOR IDB", sqlFS);
  SQL.register_for_idb(sqlFS);

  console.log("MKDIR SQL");
  SQL.FS.mkdir("/sql");
  console.log("MOUNT SQLFS");
  SQL.FS.mount(sqlFS, {}, "/sql");

  const path = "/sql/db.sqlite";
  if (typeof SharedArrayBuffer === "undefined") {
    const stream = SQL.FS.open(path, "a+");
    await stream.node.contents.readIfFallback();
    SQL.FS.close(stream);
  }

  globalThis.SQL = SQL;
  console.log("OPEN DB", SQL);
  const db: Database = new SQL.Database(path, { filename: true });
  console.log("RUN PRAGMA");
  db.run("PRAGMA journal_mode=MEMORY");
  console.log("DONE");

  rpcProvider.signal("db", { status: DbStatusCode.OPENED_DATABASE });

  await executeMigrations(db);

  rpcProvider.signal("db", { status: DbStatusCode.RAN_MIGRATIONS });

  rpcProvider.registerRpcHandler("run", ({ s, p }: Args) => db.run(s, p));

  rpcProvider.registerRpcHandler("exec", ({ s, p }: Args) => db.exec(s, p));

  rpcProvider.registerRpcHandler("prepare", ({ s, p }: Args) => {
    const preparedStatement = db.prepare(s, p);
    const id = fresh();
    preparedStatementHandles.set(id, preparedStatement);
    return id;
  });

  rpcProvider.registerRpcHandler("preparedFree", ({ h }) => {
    const prepared = preparedStatementHandles.get(h);
    if (!prepared) return;
    prepared.free();
    prepared.freemem();
  });

  rpcProvider.registerRpcHandler("preparedRun", ({ h, p }) => {
    const prepared = preparedStatementHandles.get(h);
    if (!prepared) return;
    prepared.run(p);
  });

  rpcProvider.registerRpcHandler("bulkInsert", ({ s, p, data }) => {
    const preparedStatement = db.prepare(s, p);
    db.run("BEGIN TRANSACTION");

    // TODO: yo how do bulk inserts work again
    for (const entry of data) {
      const mapped = Object.fromEntries(
        Object.entries(entry).map(([key, value]) => [`$${key}`, value]),
      );
      preparedStatement.run(mapped);
    }
    db.run("COMMIT TRANSACTION");
  });

  rpcProvider.signal("db", { status: DbStatusCode.READY });
  rpcProvider.signal(DbStatusCode.READY);
}

init();

interface Args {
  s: string;
  p?: BindParams;
}