p2pds / src / sqlite-datastore.ts

/**
 * SQLite-backed Datastore for libp2p.
 *
 * Replaces FsDatastore to avoid filesystem churn.
 * Stores libp2p peer/routing data in a single SQLite table.
 */

import type Database from "better-sqlite3";
import { Key } from "interface-datastore";
import type { Pair, Query, KeyQuery, Batch } from "interface-datastore";
type DatastorePair = Pair;

export class SqliteDatastore {
	private db: Database.Database;

	constructor(db: Database.Database) {
		this.db = db;
		this.db.exec(`
			CREATE TABLE IF NOT EXISTS ipfs_datastore (
				key TEXT PRIMARY KEY,
				value BLOB NOT NULL
			)
		`);
	}

	async put(key: Key, val: Uint8Array): Promise<Key> {
		this.db
			.prepare("INSERT OR REPLACE INTO ipfs_datastore (key, value) VALUES (?, ?)")
			.run(key.toString(), Buffer.from(val));
		return key;
	}

	async get(key: Key): Promise<Uint8Array> {
		const row = this.db
			.prepare("SELECT value FROM ipfs_datastore WHERE key = ?")
			.get(key.toString()) as { value: Buffer } | undefined;
		if (!row) {
			throw new Error(`Key not found: ${key.toString()}`);
		}
		return new Uint8Array(row.value);
	}

	async has(key: Key): Promise<boolean> {
		const row = this.db
			.prepare("SELECT 1 FROM ipfs_datastore WHERE key = ?")
			.get(key.toString());
		return row !== undefined;
	}

	async delete(key: Key): Promise<void> {
		this.db
			.prepare("DELETE FROM ipfs_datastore WHERE key = ?")
			.run(key.toString());
	}

	async * putMany(source: AsyncIterable<DatastorePair> | Iterable<DatastorePair>): AsyncGenerator<Key> {
		for await (const { key, value } of source) {
			await this.put(key, value);
			yield key;
		}
	}

	async * getMany(source: AsyncIterable<Key> | Iterable<Key>): AsyncGenerator<DatastorePair> {
		for await (const key of source) {
			const value = await this.get(key);
			yield { key, value };
		}
	}

	async * deleteMany(source: AsyncIterable<Key> | Iterable<Key>): AsyncGenerator<Key> {
		for await (const key of source) {
			await this.delete(key);
			yield key;
		}
	}

	query(q: Query): AsyncIterable<DatastorePair> {
		const self = this;
		return (async function* () {
			const prefix = q.prefix ?? "/";
			const rows = self.db
				.prepare("SELECT key, value FROM ipfs_datastore WHERE key LIKE ?")
				.all(prefix + "%") as Array<{ key: string; value: Buffer }>;
			for (const row of rows) {
				yield { key: new Key(row.key), value: new Uint8Array(row.value) };
			}
		})();
	}

	queryKeys(q: KeyQuery): AsyncIterable<Key> {
		const self = this;
		return (async function* () {
			const prefix = q.prefix ?? "/";
			const rows = self.db
				.prepare("SELECT key FROM ipfs_datastore WHERE key LIKE ?")
				.all(prefix + "%") as Array<{ key: string }>;
			for (const row of rows) {
				yield new Key(row.key);
			}
		})();
	}

	batch(): Batch {
		const ops: Array<{ type: "put"; key: Key; value: Uint8Array } | { type: "delete"; key: Key }> = [];
		return {
			put: (key: Key, value: Uint8Array) => {
				ops.push({ type: "put", key, value });
			},
			delete: (key: Key) => {
				ops.push({ type: "delete", key });
			},
			commit: async () => {
				const insertStmt = this.db.prepare("INSERT OR REPLACE INTO ipfs_datastore (key, value) VALUES (?, ?)");
				const deleteStmt = this.db.prepare("DELETE FROM ipfs_datastore WHERE key = ?");
				const tx = this.db.transaction(() => {
					for (const op of ops) {
						if (op.type === "put") {
							insertStmt.run(op.key.toString(), Buffer.from(op.value));
						} else {
							deleteStmt.run(op.key.toString());
						}
					}
				});
				tx();
			},
		};
	}
}