move did_cache to own db

packages/pds/src/context.ts

@@ -16,7 +16,7 @@ import { BlobStore } from '@atproto/repo'
 import { Services, createServices } from './services'
 import { Sequencer } from './sequencer'
 import { BackgroundQueue } from './background'
-import DidSqlCache from './did-cache'
+import { DidSqliteCache } from './did-cache'
 import { Crawlers } from './crawlers'
 import { DiskBlobStore } from './disk-blobstore'
 import { getRedisClient } from './redis'
@@ -31,7 +31,7 @@ export type AppContextOptions = {
   localViewer: (did: string) => Promise<LocalViewer>
   mailer: ServerMailer
   moderationMailer: ModerationMailer
-  didCache: DidSqlCache
+  didCache: DidSqliteCache
   idResolver: IdResolver
   plcClient: plc.Client
   services: Services
@@ -53,7 +53,7 @@ export class AppContext {
   public localViewer: (did: string) => Promise<LocalViewer>
   public mailer: ServerMailer
   public moderationMailer: ModerationMailer
-  public didCache: DidSqlCache
+  public didCache: DidSqliteCache
   public idResolver: IdResolver
   public plcClient: plc.Client
   public services: Services
@@ -119,11 +119,13 @@ export class AppContext {
 
     const moderationMailer = new ModerationMailer(modMailTransport, cfg)
 
-    const didCache = new DidSqlCache(
-      db,
+    const didCache = new DidSqliteCache(
+      path.join(cfg.db.directory, 'did_cache.sqlite'),
       cfg.identity.cacheStaleTTL,
       cfg.identity.cacheMaxTTL,
     )
+    await didCache.migrateOrThrow()
+
     const idResolver = new IdResolver({
       plcUrl: cfg.identity.plcUrl,
       didCache,

packages/pds/src/did-cache/db/index.ts

@@ -0,0 +1,11 @@
+import { Database, Migrator } from '../../db'
+import { DidCacheSchema } from './schema'
+import migrations from './migrations'
+
+export * from './schema'
+
+export type DidCacheDb = Database<DidCacheSchema>
+
+export const getMigrator = (db: DidCacheDb) => {
+  return new Migrator(db.db, migrations)
+}

packages/pds/src/did-cache/db/migrations.ts

@@ -0,0 +1,17 @@
+import { Kysely } from 'kysely'
+
+export default {
+  '001': {
+    up: async (db: Kysely<unknown>) => {
+      await db.schema
+        .createTable('did_doc')
+        .addColumn('did', 'varchar', (col) => col.primaryKey())
+        .addColumn('doc', 'text', (col) => col.notNull())
+        .addColumn('updatedAt', 'bigint', (col) => col.notNull())
+        .execute()
+    },
+    down: async (db: Kysely<unknown>) => {
+      await db.schema.dropTable('did_doc').execute()
+    },
+  },
+}

packages/pds/src/did-cache/db/schema.ts

@@ -0,0 +1,9 @@
+export interface DidDoc {
+  did: string
+  doc: string // json representation of DidDocument
+  updatedAt: number
+}
+
+export type DidCacheSchema = {
+  did_doc: DidDoc
+}

packages/pds/src/did-cache.ts → packages/pds/src/did-cache/index.ts

@@ -1,23 +1,26 @@
 import PQueue from 'p-queue'
 import { CacheResult, DidCache, DidDocument } from '@atproto/identity'
-import { excluded } from './db/util'
-import { dbLogger } from './logger'
-import { ServiceDb } from './service-db'
+import { excluded } from '../db/util'
+import { didCacheLogger } from '../logger'
+import { DidCacheDb, getMigrator } from './db'
+import { Database } from '../db'
 
-export class DidSqlCache implements DidCache {
+export class DidSqliteCache implements DidCache {
+  db: DidCacheDb
   public pQueue: PQueue | null //null during teardown
 
   constructor(
-    public db: ServiceDb,
+    dbLocation: string,
     public staleTTL: number,
     public maxTTL: number,
   ) {
+    this.db = Database.sqlite(dbLocation)
     this.pQueue = new PQueue()
   }
 
   async cacheDid(did: string, doc: DidDocument): Promise<void> {
     await this.db.db
-      .insertInto('did_cache')
+      .insertInto('did_doc')
       .values({ did, doc: JSON.stringify(doc), updatedAt: Date.now() })
       .onConflict((oc) =>
         oc.column('did').doUpdateSet({
@@ -41,14 +44,14 @@ export class DidSqlCache implements DidCache {
           await this.clearEntry(did)
         }
       } catch (err) {
-        dbLogger.error({ did, err }, 'refreshing did cache failed')
+        didCacheLogger.error({ did, err }, 'refreshing did cache failed')
       }
     })
   }
 
   async checkCache(did: string): Promise<CacheResult | null> {
     const res = await this.db.db
-      .selectFrom('did_cache')
+      .selectFrom('did_doc')
       .where('did', '=', did)
       .selectAll()
       .executeTakeFirst()
@@ -72,19 +75,23 @@ export class DidSqlCache implements DidCache {
 
   async clearEntry(did: string): Promise<void> {
     await this.db.db
-      .deleteFrom('did_cache')
+      .deleteFrom('did_doc')
       .where('did', '=', did)
       .executeTakeFirst()
   }
 
   async clear(): Promise<void> {
-    await this.db.db.deleteFrom('did_cache').execute()
+    await this.db.db.deleteFrom('did_doc').execute()
   }
 
   async processAll() {
     await this.pQueue?.onIdle()
   }
 
+  async migrateOrThrow() {
+    await getMigrator(this.db).migrateToLatestOrThrow()
+  }
+
   async destroy() {
     const pQueue = this.pQueue
     this.pQueue = null
@@ -93,5 +100,3 @@ export class DidSqlCache implements DidCache {
     await pQueue?.onIdle()
   }
 }
-
-export default DidSqlCache

packages/pds/src/logger.ts

@@ -5,11 +5,12 @@ import * as jwt from 'jsonwebtoken'
 import { parseBasicAuth } from './auth-verifier'
 
 export const dbLogger = subsystemLogger('pds:db')
+export const didCacheLogger = subsystemLogger('pds:did-cache')
 export const readStickyLogger = subsystemLogger('pds:read-sticky')
 export const redisLogger = subsystemLogger('pds:redis')
 export const seqLogger = subsystemLogger('pds:sequencer')
 export const mailerLogger = subsystemLogger('pds:mailer')
-export const labelerLogger = subsystemLogger('pds:labler')
+export const labelerLogger = subsystemLogger('pds:labeler')
 export const crawlerLogger = subsystemLogger('pds:crawler')
 export const httpLogger = subsystemLogger('pds')
 

packages/pds/src/service-db/migrations/001-init.ts

@@ -17,13 +17,6 @@ export async function up(db: Kysely<unknown>): Promise<void> {
     .addPrimaryKeyConstraint('app_password_pkey', ['did', 'name'])
     .execute()
 
-  await db.schema
-    .createTable('did_cache')
-    .addColumn('did', 'varchar', (col) => col.primaryKey())
-    .addColumn('doc', 'text', (col) => col.notNull())
-    .addColumn('updatedAt', 'bigint', (col) => col.notNull())
-    .execute()
-
   await db.schema
     .createTable('did_handle')
     .addColumn('did', 'varchar', (col) => col.primaryKey())
@@ -126,7 +119,6 @@ export async function down(db: Kysely<unknown>): Promise<void> {
   await db.schema.dropTable('invite_code_use').execute()
   await db.schema.dropTable('invite_code').execute()
   await db.schema.dropTable('did_handle').execute()
-  await db.schema.dropTable('did_cache').execute()
   await db.schema.dropTable('app_password').execute()
   await db.schema.dropTable('app_migration').execute()
 }

packages/pds/src/service-db/schema/index.ts

@@ -1,7 +1,6 @@
 import * as userAccount from './user-account'
 import * as didHandle from './did-handle'
 import * as repoRoot from './repo-root'
-import * as didCache from './did-cache'
 import * as refreshToken from './refresh-token'
 import * as appPassword from './app-password'
 import * as inviteCode from './invite-code'
@@ -14,14 +13,12 @@ export type DatabaseSchema = appMigration.PartialDB &
   refreshToken.PartialDB &
   appPassword.PartialDB &
   repoRoot.PartialDB &
-  didCache.PartialDB &
   inviteCode.PartialDB &
   emailToken.PartialDB
 
 export type { UserAccount, UserAccountEntry } from './user-account'
 export type { DidHandle } from './did-handle'
 export type { RepoRoot } from './repo-root'
-export type { DidCache } from './did-cache'
 export type { RefreshToken } from './refresh-token'
 export type { AppPassword } from './app-password'
 export type { InviteCode, InviteCodeUse } from './invite-code'