src/lib/db/adapters/libsqlAdapter.ts

// src/lib/db/adapters/libsqlAdapter.ts
import { Worker, MessageChannel, receiveMessageOnPort } from "node:worker_threads";
import type { SqliteAdapter, PreparedStatement, RunResult } from "./types";

const workerCode = `
const { parentPort } = require('worker_threads');
const { createClient } = require('@libsql/client');

let client;

parentPort.on('message', async (msg) => {
  if (msg.type === 'init') {
    try {
      client = createClient({
        url: msg.url,
        syncUrl: msg.syncUrl,
        authToken: msg.authToken,
        syncInterval: msg.syncInterval,
      });
      if (msg.syncUrl) {
        await client.sync();
      }
      parentPort.postMessage({ type: 'init_done' });
    } catch (err) {
      parentPort.postMessage({ type: 'init_failed', error: err.message });
    }
    return;
  }

  const { type, sql, params, sharedBuffer, port } = msg;
  const sharedArray = new Int32Array(sharedBuffer);

  try {
    if (type === 'sync') {
      if (client.sync) {
        await client.sync();
      }
      port.postMessage({ success: true });
    } else if (type === 'exec') {
      await client.execute(sql);
      port.postMessage({ success: true });
    } else if (type === 'close') {
      if (client.close) {
        client.close();
      }
      port.postMessage({ success: true });
    } else {
      const rs = await client.execute({ sql, args: params || [] });
      if (type === 'run') {
        const result = {
          changes: rs.rowsAffected || 0,
          lastInsertRowid: rs.lastInsertRowid !== undefined ? String(rs.lastInsertRowid) : 0
        };
        port.postMessage({ success: true, result });
      } else {
        const rows = rs.rows.map(row => {
          const obj = {};
          for (const col of rs.columns) {
            obj[col] = row[col];
          }
          return obj;
        });
        if (type === 'get') {
          port.postMessage({ success: true, result: rows[0] });
        } else {
          port.postMessage({ success: true, result: rows });
        }
      }
    }
  } catch (err) {
    port.postMessage({ success: false, error: err.message });
  } finally {
    Atomics.store(sharedArray, 0, 1);
    Atomics.notify(sharedArray, 0);
  }
});
`;

export function createLibsqlAdapter(
  filePath: string,
  options?: { url?: string; syncUrl?: string; authToken?: string; syncInterval?: number }
): SqliteAdapter {
  const worker = new Worker(workerCode, { eval: true });

  const url = options?.url || `file:${filePath}`;
  const syncUrl = options?.syncUrl;
  const authToken = options?.authToken;
  const syncInterval = options?.syncInterval || 30000;

  // Initialize the worker synchronously (block until done)
  const sharedBuffer = new SharedArrayBuffer(4);
  const sharedArray = new Int32Array(sharedBuffer);
  sharedArray[0] = 0;

  let initError: string | null = null;
  let initDone = false;

  worker.on("message", (msg) => {
    if (msg.type === "init_done") {
      initDone = true;
      Atomics.store(sharedArray, 0, 1);
      Atomics.notify(sharedArray, 0);
    } else if (msg.type === "init_failed") {
      initError = msg.error;
      Atomics.store(sharedArray, 0, 1);
      Atomics.notify(sharedArray, 0);
    }
  });

  worker.postMessage({
    type: "init",
    url,
    syncUrl,
    authToken,
    syncInterval,
  });

  // Block until initialized
  Atomics.wait(sharedArray, 0, 0);

  if (initError) {
    worker.terminate();
    throw new Error(`[DB] LibSQL initialization failed: ${initError}`);
  }

  let _isOpen = true;

  function querySync(type: string, sql: string, params?: unknown[]): any {
    if (!_isOpen) {
      throw new Error("[DB] Database is closed");
    }

    const channel = new MessageChannel();
    const queryBuffer = new SharedArrayBuffer(4);
    const queryArray = new Int32Array(queryBuffer);
    queryArray[0] = 0;

    worker.postMessage(
      {
        type,
        sql,
        params,
        sharedBuffer: queryBuffer,
        port: channel.port2,
      },
      [channel.port2]
    );

    // Block main thread
    Atomics.wait(queryArray, 0, 0);

    const msg = receiveMessageOnPort(channel.port1);
    if (!msg) {
      throw new Error("[DB] LibSQL worker did not return a response");
    }

    const { success, result, error } = msg.message;
    if (!success) {
      throw new Error(`[DB] LibSQL Query Error: ${error}`);
    }

    return result;
  }

  function runSavepoint<T>(fn: (...args: unknown[]) => T, ...args: unknown[]): T {
    const sp = `sp_${Math.random().toString(36).slice(2)}`;
    querySync("exec", `SAVEPOINT "${sp}"`);
    try {
      const result = fn(...args);
      querySync("exec", `RELEASE "${sp}"`);
      return result;
    } catch (err) {
      try {
        querySync("exec", `ROLLBACK TO "${sp}"`);
        querySync("exec", `RELEASE "${sp}"`);
      } catch {}
      throw err;
    }
  }

  return {
    driver: "libsql" as any,

    get open() {
      return _isOpen;
    },

    get name() {
      return filePath;
    },

    prepare(sql: string): PreparedStatement {
      return {
        run(...params: unknown[]): RunResult {
          const r = querySync("run", sql, params);
          return {
            changes: Number(r.changes ?? 0),
            lastInsertRowid: typeof r.lastInsertRowid === "string" ? BigInt(r.lastInsertRowid) : BigInt(r.lastInsertRowid ?? 0),
          };
        },
        get(...params: unknown[]): unknown {
          return querySync("get", sql, params);
        },
        all(...params: unknown[]): unknown[] {
          return querySync("all", sql, params);
        },
      };
    },

    exec(sql: string): void {
      querySync("exec", sql);
    },

    pragma(pragmaStr: string, options?: { simple?: boolean }): unknown {
      const sql = `PRAGMA ${pragmaStr}`;
      if (options?.simple) {
        const row = querySync("get", sql) as Record<string, unknown> | undefined;
        if (!row) return null;
        return Object.values(row)[0] ?? null;
      }
      return querySync("all", sql);
    },

    transaction<T>(fn: (...args: unknown[]) => T): (...args: unknown[]) => T {
      return (...args: unknown[]) => runSavepoint(fn, ...args);
    },

    immediate(fn: () => void): void {
      runSavepoint(() => fn());
    },

    async backup(destination: string): Promise<void> {
      // For remote or replica, we trigger a manual sync
      querySync("sync", "");
    },

    checkpoint(mode = "TRUNCATE"): void {
      // WAL checkpoints are managed on Turso side
    },

    close(): void {
      if (_isOpen) {
        try {
          querySync("close", "");
        } catch {}
        worker.terminate();
        _isOpen = false;
      }
    },

    get raw() {
      return null;
    },
  };
}