Spaces:

automindy
/

fiemwl

Build error

fiemwl / src /lib /db /adapters /libsqlAdapter.ts

Upload 1980 files

6111b2b verified 5 months ago

6.73 kB

// 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;
    },
  };
}

	// 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;
	},
	};
	}