/**
 * The worker-side server: owns sqlite-wasm connections keyed by name and
 * answers WorkerRequests. Runs identically inside a real Worker
 * (src/worker.ts) and in-process under Node tests — that's the point.
 *
 * Storage: 'opfs' uses the OPFS SyncAccessHandle pool VFS (persistent, no
 * COOP/COEP headers needed, worker-only — the reason the whole seam is
 * async). 'memory' is explicit and non-persistent; opening with 'opfs'
 * where OPFS is unavailable is a loud error, never a silent downgrade.
 */
import type {
  Database,
  PreparedStatement,
  SqlValue as WasmSqlValue,
  Sqlite3Static,
} from '@sqlite.org/sqlite-wasm';
import type { SqlValue } from '@remelondb/core';
import type { Endpoint, WorkerRequest, WorkerResponse } from './protocol';
import { OpfsPoolHeldError } from './errors';

export const POOL_RETRY_DELAYS_MS = [50, 100, 250, 500, 1000, 2000] as const;

interface Connection {
  db: Database;
  statements: Map<string, PreparedStatement>;
  storage: 'opfs' | 'memory';
}

// booleans are stored as 0/1 — the seam-wide convention.
// (bind() throws on parameterless statements, hence the length guard)
const bind = (
  statement: PreparedStatement,
  args: readonly SqlValue[],
): void => {
  if (args.length > 0) {
    statement.bind(
      args.map((value) =>
        value === true ? 1 : value === false ? 0 : (value as WasmSqlValue),
      ),
    );
  }
};

const fromColumn = (value: unknown): SqlValue => {
  if (typeof value === 'bigint') {
    if (
      value > BigInt(Number.MAX_SAFE_INTEGER) ||
      value < BigInt(Number.MIN_SAFE_INTEGER)
    ) {
      throw new Error(
        `SqliteWorkerServer: the INTEGER ${value.toString()} is outside ` +
          `JavaScript's safe integer range and would read back wrong as a ` +
          `number — store values this large as TEXT`,
      );
    }
    return Number(value);
  }
  // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- wasm sqlite types column values as `unknown`; SqlValue is the vocabulary the seam defines for them, bigint having been handled above.
  return value as SqlValue;
};

export class SqliteWorkerServer {
  private connections = new Map<string, Connection>();
  private poolUtil: Awaited<
    ReturnType<Sqlite3Static['installOpfsSAHPoolVfs']>
  > | null = null;

  constructor(private readonly sqlite3: Sqlite3Static) {}

  /**
   * Acquire the SAH pool, absorbing the holder-death race (remelonDB#3):
   * when a worker dies holding the pool, the browser releases its
   * handles asynchronously — seconds later under a real tab close. A
   * held pool is transient, so retry with backoff before declaring
   * storage unavailable. Chromium and Firefox report it as
   * NoModificationAllowedError; WebKit reports InvalidStateError
   * (remelonDB#67). Other failures (no OPFS, private mode) fail fast.
   */
  private async installPool(): Promise<
    Awaited<ReturnType<Sqlite3Static['installOpfsSAHPoolVfs']>>
  > {
    for (let attempt = 0; ; attempt++) {
      try {
        return await this.sqlite3.installOpfsSAHPoolVfs({
          initialCapacity: 32, // db + journal per open database
        });
      } catch (error) {
        const text = String(error);
        const transient =
          text.includes('NoModificationAllowedError') ||
          text.includes('InvalidStateError');
        const delay = POOL_RETRY_DELAYS_MS[attempt];
        if (!transient) {
          throw new Error(
            `OPFS storage is unavailable here (${String(error)}) — ` +
              `if this app is open in another tab, that tab holds the ` +
              `storage (open with { takeover: true } to take it over); ` +
              `pass storage: 'memory' if non-persistent storage is intended`,
          );
        }
        if (delay === undefined) {
          throw new OpfsPoolHeldError(String(error));
        }
        await new Promise((resolve) => setTimeout(resolve, delay));
      }
    }
  }

  private connection(name: string): Connection {
    const connection = this.connections.get(name);
    if (!connection) {
      throw new Error(`database '${name}' is not open`);
    }
    return connection;
  }

  private prepared(connection: Connection, sql: string): PreparedStatement {
    let statement = connection.statements.get(sql);
    if (!statement) {
      statement = connection.db.prepare(sql);
      connection.statements.set(sql, statement);
    } else {
      statement.reset();
      statement.clearBindings();
    }
    return statement;
  }

  private closeConnection(connection: Connection): void {
    for (const statement of connection.statements.values()) {
      statement.finalize();
    }
    connection.statements.clear();
    connection.db.close();
  }

  async open(
    name: string,
    storage: 'opfs' | 'memory',
  ): Promise<{ userVersion: number }> {
    if (this.connections.has(name)) {
      throw new Error(`database '${name}' is already open`);
    }
    let db: Database;
    if (storage === 'opfs') {
      if (!this.poolUtil) {
        this.poolUtil = await this.installPool();
      }
      db = new this.poolUtil.OpfsSAHPoolDb(name);
    } else {
      db = new this.sqlite3.oo1.DB(':memory:', 'c');
    }
    const connection: Connection = { db, statements: new Map(), storage };
    this.connections.set(name, connection);
    const userVersion = Number(db.selectValue('pragma user_version') ?? 0);
    return { userVersion };
  }

  /** Close every database before handing this worker's OPFS pool back. */
  releasePool(): void {
    for (const connection of this.connections.values()) {
      this.closeConnection(connection);
    }
    this.connections.clear();
    this.poolUtil?.pauseVfs();
  }

  handle(request: WorkerRequest): unknown {
    switch (request.op) {
      case 'close': {
        this.closeConnection(this.connection(request.name));
        this.connections.delete(request.name);
        return null;
      }
      case 'query': {
        const connection = this.connection(request.name);
        const statement = this.prepared(connection, request.sql);
        bind(statement, request.args);
        const columns = statement.getColumnNames();
        const rows: Record<string, SqlValue>[] = [];
        while (statement.step()) {
          const values = statement.get([]) as unknown[];
          const row: Record<string, SqlValue> = {};
          columns.forEach((column, index) => {
            row[column] = fromColumn(values[index]);
          });
          rows.push(row);
        }
        statement.reset();
        return rows;
      }
      case 'execute': {
        const connection = this.connection(request.name);
        const statement = this.prepared(connection, request.sql);
        bind(statement, request.args);
        statement.step();
        statement.reset();
        return null;
      }
      case 'executeBatch': {
        const connection = this.connection(request.name);
        const { db } = connection;
        db.exec('begin');
        try {
          for (const [sql, argSets] of request.statements) {
            const statement = this.prepared(connection, sql);
            for (const args of argSets) {
              statement.reset();
              statement.clearBindings();
              bind(statement, args);
              statement.step();
            }
            statement.reset();
          }
        } catch (error) {
          db.exec('rollback');
          throw error;
        }
        db.exec('commit');
        return null;
      }
      case 'setUserVersion': {
        const version = request.version;
        if (!Number.isInteger(version) || version < 0) {
          throw new Error(`invalid user_version ${version}`);
        }
        this.connection(request.name).db.exec(
          `pragma user_version = ${version}`,
        );
        return null;
      }
      case 'destroy': {
        const connection = this.connections.get(request.name);
        const storage = connection?.storage;
        if (connection) {
          this.closeConnection(connection);
          this.connections.delete(request.name);
        }
        if (storage === 'opfs' && this.poolUtil) {
          // pool filenames are absolute ('/name'), plus journal sidecars
          for (const file of this.poolUtil.getFileNames()) {
            if (
              file === `/${request.name}` ||
              file.startsWith(`/${request.name}-`)
            ) {
              this.poolUtil.unlink(file);
            }
          }
        }
        return null;
      }
      case 'ping':
        return null;
      case 'releasePool':
        this.releasePool();
        return null;
      case 'acquireSlot':
      case 'releaseSlot':
      case 'publishChanges':
      case 'syncTurn':
        throw new Error(
          `${request.op} is broker-only and must not reach the worker`,
        );
      case 'open':
        throw new Error('open is handled asynchronously'); // see serve()
    }
  }
}

/**
 * One server, any number of endpoints (the broker's adopted port serves
 * alongside the worker scope). Requests arriving before sqlite-wasm has
 * loaded are answered once it has (promise chaining preserves order).
 */
export function createSqliteWorkerServing(
  loadSqlite: () => Promise<Sqlite3Static>,
): (endpoint: Endpoint) => void {
  const serverPromise = loadSqlite().then(
    (sqlite3) => new SqliteWorkerServer(sqlite3),
  );
  // Requests run strictly in arrival order, across every endpoint this
  // server answers (the worker scope plus the broker's adopted ports).
  // `open` is async — it awaits the OPFS pool install — so without a
  // single shared queue a later request could run while an earlier one
  // is parked at that await and race it on the pool (double install, two
  // opens past the `already open` guard). The queue never rejects: each
  // step turns its own failure into an error response.
  let queue: Promise<unknown> = serverPromise;
  return (endpoint) => {
    endpoint.addMessageListener((message) => {
      // eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion -- an endpoint delivers `unknown`; the protocol module is the contract, and the `op` check below drops anything else.
      const request = message as WorkerRequest;
      if (typeof (request as { op?: unknown }).op !== 'string') {
        return; // control traffic (port adoption), not a request
      }
      if (request.op === 'ping') {
        // Liveness is answered out-of-band, never queued. A real op can
        // stall the queue for seconds (e.g. an OPFS pool install retrying
        // a transient error), and a queued ping behind it would time out
        // — making a busy-but-alive worker look dead and tripping the
        // broker's "went away" respawn. Answering immediately reports the
        // truth: the worker is here.
        endpoint.postMessage({ id: request.id, ok: true, result: null });
        return;
      }
      queue = queue.then(async () => {
        const server = await serverPromise;
        try {
          const result =
            request.op === 'open'
              ? await server.open(request.name, request.storage)
              : server.handle(request);
          const response: WorkerResponse = { id: request.id, ok: true, result };
          endpoint.postMessage(response);
        } catch (error: unknown) {
          const poolHeld =
            error instanceof OpfsPoolHeldError
              ? { code: error.code, diagnostic: error.diagnostic }
              : {};
          const response: WorkerResponse = {
            id: request.id,
            ok: false,
            error: error instanceof Error ? error.message : String(error),
            ...poolHeld,
          };
          endpoint.postMessage(response);
        }
      });
    });
  };
}

/** Wire a server to a single endpoint (see createSqliteWorkerServing). */
export function serveSqliteWorker(
  endpoint: Endpoint,
  loadSqlite: () => Promise<Sqlite3Static>,
): void {
  createSqliteWorkerServing(loadSqlite)(endpoint);
}