feat: rocksdb persistence and make CastSet a LWW CRDT (#107)

* chore: clean up jest

* feat: initial leveldb commit

* feat: refactor signer set to use db

* refactor: fix mock test

* feat: start adding rocksdb wrapper

* intermediate commit

* feat: integrate all sets with db

* feat: update rocksdb wrapper to create directory if missing

* chore: extend test timeout for rpcSync and engine mock

* chore: improve db/cast.test

* chore: add documentation to db/ files

* chore: update set comments

.gitignore

@@ -20,4 +20,8 @@ node_modules/
 .console-history
 
 # Code Coverage Reports
-coverage/
+coverage/
+
+# Databases
+.level/
+.rocks/

jest.config.ts

@@ -7,6 +7,8 @@ export default async (): Promise<Config.InitialOptions> => {
     moduleNameMapper: {
       '^~/(.*)$': '<rootDir>/src/$1',
     },
+    coveragePathIgnorePatterns: ['<rootDir>/build/', '<rootDir>/node_modules/'],
+    testPathIgnorePatterns: ['<rootDir>/build', '<rootDir>/node_modules'],
     // transform ts files with ts-jest and enable ESM
     transform: {
       '^.+\\.tsx?$': [

package.json

@@ -12,8 +12,8 @@
     "lint:fix": "npm run lint -- --fix",
     "server": "npm run build",
     "start": "npm run server",
-    "test": "rm -rf ./build && NODE_OPTIONS=--experimental-vm-modules jest",
-    "test:ci": "ENVIRONMENT=test NODE_OPTIONS=--experimental-vm-modules jest src/ --ci --runInBand --forceExit --coverage",
+    "test": "NODE_OPTIONS=--experimental-vm-modules jest",
+    "test:ci": "ENVIRONMENT=test NODE_OPTIONS=--experimental-vm-modules jest --ci --runInBand --forceExit --coverage",
     "typecheck": "tsc --noEmit"
   },
   "repository": {
@@ -62,7 +62,6 @@
     "@noble/ed25519": "^1.6.1",
     "caip": "^1.1.0",
     "canonicalize": "^1.0.8",
-    "colors": "^1.4.0",
     "ethereum-cryptography": "^1.1.2",
     "ethers": "^5.6.1",
     "faker": "5.5.3",
@@ -71,6 +70,7 @@
     "log-update": "5.0.1",
     "neverthrow": "^5.0.0",
     "node-fetch": "^3.2.10",
+    "rocksdb": "^5.2.1",
     "tiny-typed-emitter": "^2.1.0",
     "undici": "^5.10.0",
     "uri-js": "^4.4.1"

src/abstract-leveldown.d.ts

@@ -0,0 +1,119 @@
+// Type definitions for abstract-leveldown 7.2
+// Project: https://github.com/Level/abstract-leveldown
+// Definitions by: Meirion Hughes <https://github.com/MeirionHughes>
+//                 Daniel Byrne <https://github.com/danwbyrne>
+//                 Steffen Park <https://github.com/istherepie>
+//                 Paul Fletcher-Hill <https://github.com/pfletcherhill>
+// Definitions: https://github.com/DefinitelyTyped/DefinitelyTyped
+// TypeScript Version: 2.3
+
+export interface AbstractOptions {
+  readonly [k: string]: any;
+}
+
+export type ErrorCallback = (err: Error | undefined) => void;
+export type ErrorValueCallback<V> = (err: Error | undefined, value: V) => void;
+export type ErrorKeyValueCallback<K, V> = (err: Error | undefined, key: K, value: V) => void;
+
+export interface AbstractOpenOptions extends AbstractOptions {
+  createIfMissing?: boolean | undefined;
+  errorIfExists?: boolean | undefined;
+}
+
+export interface AbstractGetOptions extends AbstractOptions {
+  asBuffer?: boolean | undefined;
+}
+
+export interface AbstractLevelDOWN<K = any, V = any> extends AbstractOptions {
+  open(cb: ErrorCallback): void;
+  open(options: AbstractOpenOptions, cb: ErrorCallback): void;
+
+  close(cb: ErrorCallback): void;
+
+  get(key: K, cb: ErrorValueCallback<V>): void;
+  get(key: K, options: AbstractGetOptions, cb: ErrorValueCallback<V>): void;
+
+  put(key: K, value: V, cb: ErrorCallback): void;
+  put(key: K, value: V, options: AbstractOptions, cb: ErrorCallback): void;
+
+  del(key: K, cb: ErrorCallback): void;
+  del(key: K, options: AbstractOptions, cb: ErrorCallback): void;
+
+  getMany(key: K[], cb: ErrorValueCallback<V[]>): void;
+  getMany(key: K[], options: AbstractGetOptions, cb: ErrorValueCallback<V[]>): void;
+
+  batch(): AbstractChainedBatch<K, V>;
+  batch(array: ReadonlyArray<AbstractBatch<K, V>>, cb: ErrorCallback): AbstractChainedBatch<K, V>;
+  batch(
+    array: ReadonlyArray<AbstractBatch<K, V>>,
+    options: AbstractOptions,
+    cb: ErrorCallback
+  ): AbstractChainedBatch<K, V>;
+
+  iterator(options?: AbstractIteratorOptions<K>): AbstractIterator<K, V>;
+
+  readonly status: 'new' | 'opening' | 'open' | 'closing' | 'closed';
+  readonly location: string;
+  isOperational(): boolean;
+}
+
+export interface AbstractLevelDOWNConstructor {
+  new <K = any, V = any>(location: string): AbstractLevelDOWN<K, V>;
+  <K = any, V = any>(location: string): AbstractLevelDOWN<K, V>;
+}
+
+export interface AbstractIteratorOptions<K = any> extends AbstractOptions {
+  gt?: K | undefined;
+  gte?: K | undefined;
+  lt?: K | undefined;
+  lte?: K | undefined;
+  reverse?: boolean | undefined;
+  limit?: number | undefined;
+  keys?: boolean | undefined;
+  values?: boolean | undefined;
+  keyAsBuffer?: boolean | undefined;
+  valueAsBuffer?: boolean | undefined;
+}
+
+export type AbstractBatch<K = any, V = any> = PutBatch<K, V> | DelBatch<K, V>;
+
+export interface PutBatch<K = any, V = any> {
+  readonly type: 'put';
+  readonly key: K;
+  readonly value: V;
+}
+
+export interface DelBatch<K = any> {
+  readonly type: 'del';
+  readonly key: K;
+}
+
+export interface AbstractChainedBatch<K = any, V = any> extends AbstractOptions {
+  put: (key: K, value: V) => this;
+  del: (key: K) => this;
+  clear: () => this;
+  write(cb: ErrorCallback): any;
+  write(options: any, cb: ErrorCallback): any;
+  db: () => AbstractLevelDOWN;
+}
+
+export interface AbstractChainedBatchConstructor {
+  new <K = any, V = any>(db: any): AbstractChainedBatch<K, V>;
+  <K = any, V = any>(db: any): AbstractChainedBatch<K, V>;
+}
+
+export interface AbstractIterator<K, V> extends AbstractOptions {
+  db: AbstractLevelDOWN<K, V>;
+  next(cb: ErrorKeyValueCallback<K, V>): this;
+  end(cb: ErrorCallback): void;
+  [Symbol.asyncIterator](): AsyncGenerator<any, void, unknown>;
+}
+
+export interface AbstractIteratorConstructor {
+  new <K = any, V = any>(db: any): AbstractIterator<K, V>;
+  <K = any, V = any>(db: any): AbstractIterator<K, V>;
+}
+
+export const AbstractLevelDOWN: AbstractLevelDOWNConstructor;
+export const AbstractIterator: AbstractIteratorConstructor;
+export const AbstractChainedBatch: AbstractChainedBatchConstructor;

src/db/cast.test.ts

@@ -0,0 +1,120 @@
+import Faker from 'faker';
+import CastDB from '~/db/cast';
+import { Factories } from '~/factories';
+import { CastRecast, CastRemove, CastShort } from '~/types';
+import { jestRocksDB } from '~/db/jestUtils';
+import { NotFoundError } from '~/errors';
+
+const rocks = jestRocksDB('db.cast.test');
+const db = new CastDB(rocks);
+
+/** Test data */
+const fid = Faker.datatype.number();
+const target = Faker.internet.url();
+
+let cast1: CastShort;
+let recast1: CastRecast;
+let remove1: CastRemove;
+
+beforeAll(async () => {
+  cast1 = await Factories.CastShort.create({ data: { fid, body: { targetUri: target } } });
+  recast1 = await Factories.CastRecast.create({ data: { fid, body: { targetCastUri: target } } });
+  remove1 = await Factories.CastRemove.create({ data: { fid, body: { targetHash: cast1.hash } } });
+});
+
+describe('putCastAdd', () => {
+  describe('CastShort', () => {
+    test('stores cast', async () => {
+      await expect(db.putCastAdd(cast1)).resolves.toEqual(undefined);
+      await expect(db.getCastAdd(cast1.data.fid, cast1.hash)).resolves.toEqual(cast1);
+      await expect(db.getCastShortsByTarget(target)).resolves.toEqual([cast1]);
+      await expect(db.getCastRecastsByTarget(target)).resolves.toEqual([]);
+    });
+
+    test('indexes cast by target if targetUri is present', async () => {
+      await expect(db.putCastAdd(cast1)).resolves.toEqual(undefined);
+      await expect(db.getCastShortsByTarget(target)).resolves.toEqual([cast1]);
+    });
+
+    test('does not index by target if targetUri is blank', async () => {
+      const cast1NoTarget: CastShort = {
+        ...cast1,
+        data: { ...cast1.data, body: { ...cast1.data.body, targetUri: undefined } },
+      };
+      await expect(db.putCastAdd(cast1NoTarget)).resolves.toEqual(undefined);
+      await expect(db.getCastShortsByTarget(target)).resolves.toEqual([]);
+    });
+
+    test('deletes associated CastRemove if present', async () => {
+      await db.putCastRemove(remove1);
+      await expect(db.getCastRemove(cast1.data.fid, cast1.hash)).resolves.toEqual(remove1);
+      await expect(db.putCastAdd(cast1)).resolves.toEqual(undefined);
+      await expect(db.getCastRemove(cast1.data.fid, cast1.hash)).rejects.toThrow(NotFoundError);
+    });
+  });
+
+  describe('CastRecast', () => {
+    test('stores cast and indexes it by target', async () => {
+      await expect(db.putCastAdd(recast1)).resolves.toEqual(undefined);
+      await expect(db.getCastAdd(recast1.data.fid, recast1.hash)).resolves.toEqual(recast1);
+    });
+
+    test('indexes cast by target', async () => {
+      await expect(db.putCastAdd(recast1)).resolves.toEqual(undefined);
+      await expect(db.getCastRecastsByTarget(target)).resolves.toEqual([recast1]);
+    });
+  });
+});
+
+describe('putCastRemove', () => {
+  test('stores CastRemove', async () => {
+    await expect(db.putCastRemove(remove1)).resolves.toEqual(undefined);
+    await expect(db.getCastRemove(remove1.data.fid, remove1.data.body.targetHash)).resolves.toEqual(remove1);
+  });
+
+  test('deletes associated cast if present', async () => {
+    await db.putCastAdd(cast1);
+    await expect(db.getCastAdd(remove1.data.fid, cast1.hash)).resolves.toEqual(cast1);
+    await expect(db.putCastRemove(remove1)).resolves.toEqual(undefined);
+    await expect(db.getCastAdd(remove1.data.fid, cast1.hash)).rejects.toThrow(NotFoundError);
+  });
+});
+
+describe('getCastAdd', () => {
+  test('returns a cast', async () => {
+    await db.putCastAdd(cast1);
+    await expect(db.getCastAdd(cast1.data.fid, cast1.hash)).resolves.toEqual(cast1);
+  });
+
+  test('fails if cast not found', async () => {
+    await expect(db.getCastAdd(cast1.data.fid, cast1.hash)).rejects.toThrow();
+  });
+});
+
+describe('getAllCastMessagesByUser', () => {
+  test('returns array of messages', async () => {
+    await db.putCastAdd(recast1);
+    await db.putCastRemove(remove1);
+    const messages = await db.getAllCastMessagesByUser(fid);
+    expect(new Set(messages)).toEqual(new Set([recast1, remove1]));
+  });
+
+  test('returns empty array without messages', async () => {
+    await expect(db.getAllCastMessagesByUser(fid)).resolves.toEqual([]);
+  });
+});
+
+describe('deleteAllCastMessagesBySigner', () => {
+  test('deletes all messages from a signer', async () => {
+    await db.putCastAdd(recast1);
+    await db.putCastRemove(remove1);
+
+    await expect(db.getMessagesBySigner(fid, recast1.signer)).resolves.toEqual([recast1]);
+    await expect(db.deleteAllCastMessagesBySigner(fid, recast1.signer)).resolves.toEqual(undefined);
+    await expect(db.getMessagesBySigner(fid, recast1.signer)).resolves.toEqual([]);
+
+    await expect(db.getMessagesBySigner(fid, remove1.signer)).resolves.toEqual([remove1]);
+    await expect(db.deleteAllCastMessagesBySigner(fid, remove1.signer)).resolves.toEqual(undefined);
+    await expect(db.getMessagesBySigner(fid, remove1.signer)).resolves.toEqual([]);
+  });
+});