p2pds / src / sqlite-blockstore.ts

/**
 * SQLite-backed Blockstore for Helia.
 *
 * Replaces FsBlockstore to avoid creating thousands of tiny files
 * that hammer macOS fseventsd. All blocks stored in a single SQLite table.
 */

import type Database from "better-sqlite3";
import { CID } from "multiformats";

export class SqliteBlockstore {
	private db: Database.Database;

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

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

	async * get(key: CID): AsyncGenerator<Uint8Array> {
		const row = this.db
			.prepare("SELECT bytes FROM ipfs_blocks WHERE cid = ?")
			.get(key.toString()) as { bytes: Buffer } | undefined;
		if (!row) {
			throw new Error(`Block not found: ${key.toString()}`);
		}
		yield new Uint8Array(row.bytes);
	}

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

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

	async * putMany(source: AsyncIterable<{ cid: CID; block: Uint8Array }> | Iterable<{ cid: CID; block: Uint8Array }>): AsyncGenerator<CID> {
		for await (const { cid, block } of source) {
			await this.put(cid, block);
			yield cid;
		}
	}

	async * getMany(source: AsyncIterable<CID> | Iterable<CID>): AsyncGenerator<{ cid: CID; block: Uint8Array }> {
		for await (const cid of source) {
			let block: Uint8Array = new Uint8Array(0);
			for await (const chunk of this.get(cid)) {
				block = chunk;
			}
			yield { cid, block };
		}
	}

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

	/**
	 * Delete all blocks from the blockstore.
	 * Used during full disconnect to wipe the node clean.
	 */
	clear(): void {
		this.db.prepare("DELETE FROM ipfs_blocks").run();
	}

	async * getAll(): AsyncGenerator<{ cid: CID; block: Uint8Array }> {
		const rows = this.db
			.prepare("SELECT cid, bytes FROM ipfs_blocks")
			.all() as Array<{ cid: string; bytes: Buffer }>;
		for (const row of rows) {
			yield { cid: CID.parse(row.cid), block: new Uint8Array(row.bytes) };
		}
	}
}