SqliteObjectStorage.ts

import type { Statement, Database } from 'better-sqlite3';
import { guid } from './utils/index.ts';
import type { IContainer } from 'node-cqrs';
import type { IObjectStorage, Identifier } from '../interfaces/index.ts';
import { AbstractSqliteAccessor } from './AbstractSqliteAccessor.ts';
import { assertDefined, assertFunction } from '../utils/assert.ts';

export class SqliteObjectStorage<TRecord> extends AbstractSqliteAccessor implements IObjectStorage<TRecord> {

	#tableName: string;
	#getQuery!: Statement<[Buffer], { data: string, version: number }>;
	#insertQuery!: Statement<[Buffer, string], void>;
	#updateByIdAndVersionQuery!: Statement<[string, Buffer, number], void>;
	#deleteQuery!: Statement<[Buffer], void>;

	constructor(o: Pick<IContainer, 'viewModelSqliteDb' | 'viewModelSqliteDbFactory'> & {
		tableName: string
	}) {
		super(o);

		this.#tableName = o.tableName;
	}

	protected initialize(db: Database) {
		db.exec(`CREATE TABLE IF NOT EXISTS ${this.#tableName} (
			id BLOB PRIMARY KEY,
			version INTEGER DEFAULT 1,
			data TEXT NOT NULL
		);`);

		this.#getQuery = db.prepare(`
			SELECT data, version
			FROM ${this.#tableName}
			WHERE id = ?
		`);

		this.#insertQuery = db.prepare(`
			INSERT INTO ${this.#tableName} (id, data)
			VALUES (?, ?)
		`);

		this.#updateByIdAndVersionQuery = db.prepare(`
			UPDATE ${this.#tableName}
			SET
				data = ?,
				version = version + 1
			WHERE
				id = ?
				AND version = ?
		`);

		this.#deleteQuery = db.prepare(`
			DELETE FROM ${this.#tableName}
			WHERE id = ?
		`);
	}

	async get(id: Identifier): Promise<TRecord | undefined> {
		assertDefined(id, 'id');
		await this.assertConnection();

		const r = this.#getQuery.get(guid(id));
		if (!r)
			return undefined;

		return JSON.parse(r.data);
	}

	getSync(id: Identifier): TRecord | undefined {
		assertDefined(id, 'id');
		const r = this.#getQuery.get(guid(id));
		if (!r)
			return undefined;

		return JSON.parse(r.data);
	}

	async create(id: Identifier, data: TRecord) {
		assertDefined(id, 'id');
		await this.assertConnection();

		this.#createSync(id, data);
	}

	#createSync(id: Identifier, data: TRecord) {
		const r = this.#insertQuery.run(guid(id), JSON.stringify(data));
		if (r.changes !== 1)
			throw new Error(`Record '${id}' could not be created`);
	}

	async update(id: Identifier, update: (r: TRecord) => TRecord) {
		assertDefined(id, 'id');
		assertFunction(update, 'update');

		await this.assertConnection();

		this.#updateSync(id, update);
	}

	#updateSync(id: Identifier, update: (r: TRecord) => TRecord) {
		const gid = guid(id);
		const record = this.#getQuery.get(gid);
		if (!record)
			throw new Error(`Record '${id}' does not exist`);

		this.#updateExistingSync(id, record, update);
	}

	#updateExistingSync(id: Identifier, record: { data: string, version: number }, update: (r: TRecord) => TRecord) {
		const gid = guid(id);
		const data = JSON.parse(record.data);
		const updatedData = update(data);
		const updatedJson = JSON.stringify(updatedData);

		// Version check is implemented to ensure the record isn't modified by another process.
		// A conflict resolution strategy could potentially be passed as an option to this method,
		// but for now, conflict resolution should happen outside this class.
		const r = this.#updateByIdAndVersionQuery.run(updatedJson, gid, record.version);
		if (r.changes !== 1)
			throw new Error(`Record '${id}' could not be updated`);
	}

	async updateEnforcingNew(id: Identifier, update: (r?: TRecord) => TRecord) {
		assertDefined(id, 'id');
		assertFunction(update, 'update');

		await this.assertConnection();

		const record = this.#getQuery.get(guid(id));
		if (record)
			this.#updateExistingSync(id, record, update as (r: TRecord) => TRecord);
		else
			this.#createSync(id, update());
	}

	async delete(id: Identifier): Promise<boolean> {
		assertDefined(id, 'id');
		await this.assertConnection();

		const r = this.#deleteQuery.run(guid(id));
		return r.changes === 1;
	}
}