wip

packages/pds/src/user-db/index.ts → packages/pds/src/actor-store/actor-db/index.ts

@@ -1,4 +1,5 @@
 import assert from 'assert'
+import path from 'path'
 import {
   Kysely,
   SqliteDialect,
@@ -14,38 +15,38 @@ import SqliteDB from 'better-sqlite3'
 import { DatabaseSchema } from './schema'
 import * as migrations from './migrations'
 import { CtxMigrationProvider } from './migrations/provider'
+export * from './schema'
 
-export class UserDb {
+type CommitHook = () => void
+
+export class ActorDb {
   migrator: Migrator
   destroyed = false
+  commitHooks: CommitHook[] = []
 
-  constructor(public db: Kysely<DatabaseSchema>) {
+  constructor(public did: string, public db: Kysely<DatabaseSchema>) {
     this.migrator = new Migrator({
       db,
       provider: new CtxMigrationProvider(migrations),
     })
   }
 
-  static sqlite(location: string): UserDb {
+  static sqlite(location: string, did: string): ActorDb {
     const db = new Kysely<DatabaseSchema>({
       dialect: new SqliteDialect({
-        database: new SqliteDB(location),
+        database: new SqliteDB(path.join(location, did)),
       }),
     })
-    return new UserDb(db)
-  }
-
-  static memory(): UserDb {
-    return UserDb.sqlite(':memory:')
+    return new ActorDb(did, db)
   }
 
-  async transaction<T>(fn: (db: UserDb) => Promise<T>): Promise<T> {
+  async transaction<T>(fn: (db: ActorDb) => Promise<T>): Promise<T> {
     const leakyTxPlugin = new LeakyTxPlugin()
-    return this.db
+    const { hooks, txRes } = await this.db
       .withPlugin(leakyTxPlugin)
       .transaction()
       .execute(async (txn) => {
-        const dbTxn = new UserDb(txn)
+        const dbTxn = new ActorDb(this.did, txn)
         const txRes = await fn(dbTxn)
           .catch(async (err) => {
             leakyTxPlugin.endTx()
@@ -54,8 +55,16 @@ export class UserDb {
             throw err
           })
           .finally(() => leakyTxPlugin.endTx())
-        return txRes
+        const hooks = dbTxn.commitHooks
+        return { hooks, txRes }
       })
+    hooks.map((hook) => hook())
+    return txRes
+  }
+
+  onCommit(fn: () => void) {
+    this.assertTransaction()
+    this.commitHooks.push(fn)
   }
 
   get isTransaction() {

packages/pds/src/user-db/schema/index.ts → packages/pds/src/actor-store/actor-db/schema/index.ts

@@ -6,6 +6,14 @@ import * as ipldBlock from './ipld-block'
 import * as blob from './blob'
 import * as repoBlob from './repo-blob'
 
+export type { UserPref } from './user-pref'
+export type { RepoRoot } from './repo-root'
+export type { Record } from './record'
+export type { Backlink } from './backlink'
+export type { IpldBlock } from './ipld-block'
+export type { Blob } from './blob'
+export type { RepoBlob } from './repo-blob'
+
 export type DatabaseSchema = userPref.PartialDB &
   repoRoot.PartialDB &
   record.PartialDB &

packages/pds/src/user-db/schema/repo-root.ts → packages/pds/src/actor-store/actor-db/schema/repo-root.ts

@@ -5,6 +5,6 @@ export interface RepoRoot {
   indexedAt: string
 }
 
-export const tableName = 'repo_root'
+const tableName = 'repo_root'
 
 export type PartialDB = { [tableName]: RepoRoot }

packages/pds/src/actor-store/index.ts

@@ -0,0 +1,94 @@
+import { AtpAgent } from '@atproto/api'
+import * as crypto from '@atproto/crypto'
+import { BlobStore } from '@atproto/repo'
+import { ActorDb } from './actor-db'
+import { ActorRepo } from './repo'
+import { ActorRecord } from './record'
+import { ActorLocal } from './local'
+import { ActorPreference } from './preference'
+import { BackgroundQueue } from '../background'
+
+type ActorStoreReaderResources = {
+  repoSigningKey: crypto.Keypair
+  pdsHostname: string
+  appViewAgent?: AtpAgent
+  appViewDid?: string
+  appViewCdnUrlPattern?: string
+}
+
+type ActorStoreResources = ActorStoreReaderResources & {
+  blobstore: BlobStore
+  backgroundQueue: BackgroundQueue
+}
+
+export const createActorStore = (
+  resources: ActorStoreResources,
+): ActorStore => {
+  return {
+    reader: (did: string) => {
+      const db = ActorDb.sqlite('', did)
+      return createActorReader(db, resources)
+    },
+    transact: <T>(did: string, fn: ActorStoreTransactFn<T>) => {
+      const db = ActorDb.sqlite('', did)
+      return db.transaction((dbTxn) => {
+        const store = createActorTransactor(dbTxn, resources)
+        return fn(store)
+      })
+    },
+  }
+}
+
+const createActorTransactor = (
+  db: ActorDb,
+  resources: ActorStoreResources,
+): ActorStoreTransactor => {
+  const { repoSigningKey, blobstore, backgroundQueue } = resources
+  const reader = createActorReader(db, resources)
+  return {
+    ...reader,
+    repo: new ActorRepo(db, repoSigningKey, blobstore, backgroundQueue),
+  }
+}
+
+const createActorReader = (
+  db: ActorDb,
+  resources: ActorStoreReaderResources,
+): ActorStoreReader => {
+  const {
+    repoSigningKey,
+    pdsHostname,
+    appViewAgent,
+    appViewDid,
+    appViewCdnUrlPattern,
+  } = resources
+  return {
+    record: new ActorRecord(db),
+    local: new ActorLocal(
+      db,
+      repoSigningKey,
+      pdsHostname,
+      appViewAgent,
+      appViewDid,
+      appViewCdnUrlPattern,
+    ),
+    pref: new ActorPreference(db),
+  }
+}
+
+export type ActorStore = {
+  reader: (did: string) => ActorStoreReader
+  transact: <T>(did: string, store: ActorStoreTransactFn<T>) => Promise<T>
+}
+
+export type ActorStoreTransactFn<T> = (fn: ActorStoreTransactor) => Promise<T>
+
+export type ActorStoreTransactor = ActorStoreReader & {
+  repo: ActorRepo
+}
+
+export type ActorStoreReader = {
+  record: ActorRecord
+  local: ActorLocal
+  pref: ActorPreference
+}

packages/pds/src/user-store/local.ts → packages/pds/src/actor-store/local.ts

@@ -31,13 +31,13 @@ import {
 import { AtpAgent } from '@atproto/api'
 import { Keypair } from '@atproto/crypto'
 import { createServiceAuthHeaders } from '@atproto/xrpc-server'
-import { UserDb } from '../user-db'
+import { ActorDb } from './actor-db'
 
 type CommonSignedUris = 'avatar' | 'banner' | 'feed_thumbnail' | 'feed_fullsize'
 
-export class LocalService {
+export class ActorLocal {
   constructor(
-    public db: UserDb,
+    public db: ActorDb,
     public signingKey: Keypair,
     public pdsHostname: string,
     public appViewAgent?: AtpAgent,
@@ -52,8 +52,8 @@ export class LocalService {
     appviewDid?: string,
     appviewCdnUrlPattern?: string,
   ) {
-    return (db: UserDb) =>
-      new LocalService(
+    return (db: ActorDb) =>
+      new ActorLocal(
         db,
         signingKey,
         pdsHostname,

packages/pds/src/user-store/preferences.ts → packages/pds/src/actor-store/preference.ts

@@ -1,11 +1,11 @@
 import { InvalidRequestError } from '@atproto/xrpc-server'
-import { UserDb } from '../user-db'
+import { ActorDb } from './actor-db'
 
-export class PreferencesService {
-  constructor(public db: UserDb) {}
+export class ActorPreference {
+  constructor(public db: ActorDb) {}
 
   static creator() {
-    return (db: UserDb) => new PreferencesService(db)
+    return (db: ActorDb) => new ActorPreference(db)
   }
 
   async getPreferences(

packages/pds/src/actor-store/reader.ts

@@ -0,0 +1 @@
+export class ActorStoreReader {}