feat(space-store): add sqlite implementation

packages/common/space-store/package.json

@@ -8,7 +8,9 @@
     ".": "./index.ts",
     "./idb": "./src/impls/idb/index.ts",
     "./idb/v1": "./src/impls/idb/v1/index.ts",
-    "./cloud": "./src/impls/cloud/index.ts"
+    "./cloud": "./src/impls/cloud/index.ts",
+    "./sqlite": "./src/impls/sqlite/index.ts",
+    "./sqlite/v1": "./src/impls/sqlite/v1/index.ts"
   },
   "dependencies": {
     "@toeverything/infra": "workspace:*",
@@ -18,12 +20,16 @@
     "yjs": "patch:yjs@npm%3A13.6.18#~/.yarn/patches/yjs-npm-13.6.18-ad0d5f7c43.patch"
   },
   "devDependencies": {
+    "@affine/electron-api": "workspace:*",
     "@affine/graphql": "workspace:*",
+    "@affine/native": "workspace:*",
     "idb": "^8.0.0",
     "socket.io-client": "^4.7.5"
   },
   "peerDependencies": {
+    "@affine/electron-api": "workspace:*",
     "@affine/graphql": "workspace:*",
+    "@affine/native": "workspace:*",
     "idb": "^8.0.0",
     "socket.io-client": "^4.7.5"
   }

packages/common/space-store/src/impls/sqlite/blob.ts

@@ -0,0 +1,39 @@
+import { share } from '../../connection';
+import {
+  type BlobRecord,
+  BlobStorage,
+  type BlobStorageOptions,
+} from '../../storage';
+import { NativeDBConnection } from './db';
+
+interface SqliteBlobStorageOptions extends BlobStorageOptions {
+  dbPath: string;
+}
+
+export class SqliteBlobStorage extends BlobStorage<SqliteBlobStorageOptions> {
+  override connection = share(new NativeDBConnection(this.options.dbPath));
+
+  get db() {
+    return this.connection.inner;
+  }
+
+  override async get(key: string) {
+    return this.db.getBlob(key);
+  }
+
+  override async set(blob: BlobRecord) {
+    await this.db.setBlob(blob);
+  }
+
+  override async delete(key: string, permanently: boolean) {
+    await this.db.deleteBlob(key, permanently);
+  }
+
+  override async release() {
+    await this.db.releaseBlobs();
+  }
+
+  override async list() {
+    return this.db.listBlobs();
+  }
+}

packages/common/space-store/src/impls/sqlite/db.ts

@@ -0,0 +1,23 @@
+import { DocStorage as NativeDocStorage } from '@affine/native';
+
+import { Connection } from '../../connection';
+
+export class NativeDBConnection extends Connection<NativeDocStorage> {
+  constructor(private readonly dbPath: string) {
+    super();
+  }
+
+  override get shareId(): string {
+    return `sqlite:${this.dbPath}`;
+  }
+
+  override async doConnect() {
+    const conn = new NativeDocStorage(this.dbPath);
+    await conn.init();
+    return conn;
+  }
+
+  override async doDisconnect(conn: NativeDocStorage) {
+    await conn.close();
+  }
+}

packages/common/space-store/src/impls/sqlite/doc.ts

@@ -0,0 +1,86 @@
+import { share } from '../../connection';
+import {
+  type DocClocks,
+  type DocRecord,
+  DocStorage,
+  type DocStorageOptions,
+  type DocUpdate,
+} from '../../storage';
+import { NativeDBConnection } from './db';
+
+interface SqliteDocStorageOptions extends DocStorageOptions {
+  dbPath: string;
+}
+
+export class SqliteDocStorage extends DocStorage<SqliteDocStorageOptions> {
+  get name() {
+    return 'sqlite';
+  }
+  override connection = share(new NativeDBConnection(this.options.dbPath));
+
+  get db() {
+    return this.connection.inner;
+  }
+
+  override async pushDocUpdate(update: DocUpdate) {
+    const timestamp = await this.db.pushUpdate(update.docId, update.bin);
+
+    return { docId: update.docId, timestamp };
+  }
+
+  override async deleteDoc(docId: string) {
+    await this.db.deleteDoc(docId);
+  }
+
+  override async getDocTimestamps(after?: Date) {
+    const clocks = await this.db.getDocClocks(
+      after ? new Date(after) : undefined
+    );
+
+    return clocks.reduce((ret, cur) => {
+      ret[cur.docId] = cur.timestamp;
+      return ret;
+    }, {} as DocClocks);
+  }
+
+  protected override async getDocSnapshot(docId: string) {
+    const snapshot = await this.db.getDocSnapshot(docId);
+
+    if (!snapshot) {
+      return null;
+    }
+
+    return {
+      docId,
+      bin: snapshot.data,
+      timestamp: snapshot.timestamp,
+    };
+  }
+
+  protected override async setDocSnapshot(
+    snapshot: DocRecord
+  ): Promise<boolean> {
+    return this.db.setDocSnapshot({
+      docId: snapshot.docId,
+      data: Buffer.from(snapshot.bin),
+      timestamp: new Date(snapshot.timestamp),
+    });
+  }
+
+  protected override async getDocUpdates(docId: string) {
+    return this.db.getDocUpdates(docId).then(updates =>
+      updates.map(update => ({
+        docId,
+        bin: update.data,
+        timestamp: update.createdAt,
+      }))
+    );
+  }
+
+  protected override markUpdatesMerged(docId: string, updates: DocRecord[]) {
+    return this.db.markUpdatesMerged(
+      docId,
+      updates.map(update => update.timestamp)
+    );
+  }
+}

packages/common/space-store/src/impls/sqlite/index.ts

@@ -0,0 +1,3 @@
+export * from './blob';
+export * from './doc';
+export * from './sync';

packages/common/space-store/src/impls/sqlite/sync.ts

@@ -0,0 +1,48 @@
+import { share } from '../../connection';
+import {
+  type DocClock,
+  type DocClocks,
+  SyncStorage,
+  type SyncStorageOptions,
+} from '../../storage';
+import { NativeDBConnection } from './db';
+
+export interface SqliteSyncStorageOptions extends SyncStorageOptions {
+  dbPath: string;
+}
+
+export class SqliteSyncStorage extends SyncStorage<SqliteSyncStorageOptions> {
+  override connection = share(new NativeDBConnection(this.options.dbPath));
+
+  get db() {
+    return this.connection.inner;
+  }
+
+  override async getPeerClocks(peer: string) {
+    const records = await this.db.getPeerClocks(peer);
+    return records.reduce((clocks, { docId, timestamp }) => {
+      clocks[docId] = timestamp;
+      return clocks;
+    }, {} as DocClocks);
+  }
+
+  override async setPeerClock(peer: string, clock: DocClock) {
+    await this.db.setPeerClock(peer, clock.docId, clock.timestamp);
+  }
+
+  override async getPeerPushedClocks(peer: string) {
+    const records = await this.db.getPeerPushedClocks(peer);
+    return records.reduce((clocks, { docId, timestamp }) => {
+      clocks[docId] = timestamp;
+      return clocks;
+    }, {} as DocClocks);
+  }
+
+  override async setPeerPushedClock(peer: string, clock: DocClock) {
+    await this.db.setPeerPushedClock(peer, clock.docId, clock.timestamp);
+  }
+
+  override async clearClocks() {
+    await this.db.clearClocks();
+  }
+}

packages/common/space-store/src/impls/sqlite/v1/blob.ts

@@ -0,0 +1,66 @@
+import { apis } from '@affine/electron-api';
+
+import { DummyConnection, share } from '../../../connection';
+import { BlobStorage, type BlobStorageOptions } from '../../../storage';
+
+interface SqliteV1BlobStorageOptions extends BlobStorageOptions {
+  dbPath: string;
+}
+
+/**
+ * @deprecated readonly
+ */
+export class SqliteV1BlobStorage extends BlobStorage<SqliteV1BlobStorageOptions> {
+  override connection = share(new DummyConnection());
+
+  get db() {
+    if (!apis) {
+      throw new Error('Not in electron context.');
+    }
+
+    return apis.db;
+  }
+
+  override async get(key: string) {
+    const data: Uint8Array | null = await this.db.getBlob(
+      this.spaceType,
+      this.spaceId,
+      key
+    );
+
+    if (!data) {
+      return null;
+    }
+
+    return {
+      key,
+      data,
+      mime: '',
+      createdAt: new Date(),
+    };
+  }
+
+  override async delete(key: string, permanently: boolean) {
+    if (permanently) {
+      await this.db.deleteBlob(this.spaceType, this.spaceId, key);
+    }
+  }
+
+  override async list() {
+    const keys = await this.db.getBlobKeys(this.spaceType, this.spaceId);
+
+    return keys.map(key => ({
+      key,
+      mime: '',
+      size: 0,
+      createdAt: new Date(),
+    }));
+  }
+
+  override async set() {
+    // no more writes
+  }
+  override async release() {
+    // no more writes
+  }
+}

packages/common/space-store/src/impls/sqlite/v1/doc.ts

@@ -0,0 +1,76 @@
+import { apis } from '@affine/electron-api';
+
+import { DummyConnection, share } from '../../../connection';
+import {
+  type DocRecord,
+  DocStorage,
+  type DocStorageOptions,
+  type DocUpdate,
+} from '../../../storage';
+
+interface SqliteV1DocStorageOptions extends DocStorageOptions {
+  dbPath: string;
+}
+
+/**
+ * @deprecated readonly
+ */
+export class SqliteV1DocStorage extends DocStorage<SqliteV1DocStorageOptions> {
+  override connection = share(new DummyConnection());
+
+  get name() {
+    return 'sqlite(old)';
+  }
+
+  get db() {
+    if (!apis) {
+      throw new Error('Not in electron context.');
+    }
+
+    return apis.db;
+  }
+
+  override async pushDocUpdate(update: DocUpdate) {
+    // no more writes
+
+    return { docId: update.docId, timestamp: new Date() };
+  }
+
+  override async getDoc(docId: string) {
+    const bin = await this.db.getDocAsUpdates(
+      this.spaceType,
+      this.spaceId,
+      docId
+    );
+
+    return {
+      docId,
+      bin,
+      timestamp: new Date(),
+    };
+  }
+
+  override async deleteDoc(docId: string) {
+    await this.db.deleteDoc(this.spaceType, this.spaceId, docId);
+  }
+
+  protected override async getDocSnapshot() {
+    return null;
+  }
+
+  override async getDocTimestamps() {
+    return {};
+  }
+
+  protected override async setDocSnapshot(): Promise<boolean> {
+    return false;
+  }
+
+  protected override async getDocUpdates(): Promise<DocRecord[]> {
+    return [];
+  }
+
+  protected override async markUpdatesMerged(): Promise<number> {
+    return 0;
+  }
+}

packages/common/space-store/src/impls/sqlite/v1/index.ts

@@ -0,0 +1,2 @@
+export * from './blob';
+export * from './doc';