Migration test

packages/pds/src/db/migrations/20230118T223059239Z-repo-sync-data.ts

@@ -18,9 +18,8 @@ export async function up(db: DatabaseSchema): Promise<void> {
     .execute()
   await db.schema
     .createTable(commitHistoryTable)
-    .addColumn('commit', 'varchar', (col) => col.notNull())
+    .addColumn('commit', 'varchar', (col) => col.primaryKey())
     .addColumn('prev', 'varchar')
-    .addPrimaryKeyConstraint(`${commitHistoryTable}_pkey`, ['commit', 'prev'])
     .execute()
 
   const migrateUser = async (did: string, root: CID) => {

packages/pds/src/db/migrations/index.ts

@@ -9,5 +9,5 @@ export * as _20221209T210026294Z from './20221209T210026294Z-banners'
 export * as _20221212T195416407Z from './20221212T195416407Z-post-media'
 export * as _20221215T220356370Z from './20221215T220356370Z-password-reset-otp'
 export * as _20221226T213635517Z from './20221226T213635517Z-mute-init'
-export * as _20221220T225614428Z from './20221230T215012029Z-moderation-init'
+export * as _20221230T215012029Z from './20221230T215012029Z-moderation-init'
 export * as _20230118T223059239Z from './20230118T223059239Z-repo-sync-data'

packages/pds/src/sql-repo-storage.ts

@@ -19,42 +19,38 @@ export class SqlRepoStorage extends RepoStorage {
   }
 
   async getHead(forUpdate?: boolean): Promise<CID | null> {
-    // if for update, we lock the row & cache the last commit
+    // if for update, we lock the row
+    let builder = this.db.db
+      .selectFrom('repo_root')
+      .selectAll()
+      .where('did', '=', this.did)
+    if (forUpdate && this.db.dialect !== 'sqlite') {
+      // SELECT FOR UPDATE is not supported by sqlite, but sqlite txs are SERIALIZABLE so we don't actually need it
+      builder = builder.forUpdate()
+    }
+    const found = await builder.executeTakeFirst()
+    if (!found) return null
+
+    // if for update, we cache the blocks from last commit
+    // this must be split out into a separate query because of how pg handles SELECT FOR UPDATE with outer joins
     if (forUpdate) {
-      this.db.assertTransaction()
-      let builder = this.db.db
-        .selectFrom('repo_root')
-        .leftJoin(
-          'repo_commit_block',
-          'repo_commit_block.commit',
-          'repo_commit_block.block',
-        )
+      const res = await this.db.db
+        .selectFrom('repo_commit_block')
         .leftJoin('ipld_block', 'ipld_block.cid', 'repo_commit_block.block')
+        .where('repo_commit_block.commit', '=', found.root)
         .select([
-          'repo_root.root as root',
           'ipld_block.cid as blockCid',
           'ipld_block.content as blockBytes',
         ])
-        .where('did', '=', this.did)
-      if (this.db.dialect !== 'sqlite') {
-        // SELECT FOR UPDATE is not supported by sqlite, but sqlite txs are SERIALIZABLE so we don't actually need it
-        builder = builder.forUpdate()
-      }
-      const res = await builder.execute()
+        .execute()
       res.forEach((row) => {
         if (row.blockCid && row.blockBytes) {
           this.cache.set(CID.parse(row.blockCid), row.blockBytes)
         }
       })
-      return res.length > 0 ? CID.parse(res[0].root) : null
-    } else {
-      const found = await this.db.db
-        .selectFrom('repo_root')
-        .selectAll()
-        .where('did', '=', this.did)
-        .executeTakeFirst()
-      return found ? CID.parse(found.root) : null
     }
+
+    return CID.parse(found.root)
   }
 
   async getBytes(cid: CID): Promise<Uint8Array | null> {

packages/pds/tests/migrations/repo-sync-data.test.ts

@@ -0,0 +1,82 @@
+import AtpApi from '@atproto/api'
+import { Database } from '../../src'
+import { SeedClient } from '../seeds/client'
+import { CloseFn, runTestServer } from '../_util'
+import basicSeed from '../seeds/basic'
+import { RepoCommitHistory } from '../../src/db/tables/repo-commit-history'
+import { RepoCommitBlock } from '../../src/db/tables/repo-commit-block'
+
+describe('repo sync data migration', () => {
+  let close: CloseFn
+  let db: Database
+
+  beforeAll(async () => {
+    const server = await runTestServer({
+      dbPostgresSchema: 'migration_repo_sync',
+    })
+    close = server.close
+    db = server.ctx.db
+    const client = AtpApi.service(server.url)
+    const sc = new SeedClient(client)
+    await basicSeed(sc, server.ctx.messageQueue)
+  })
+
+  afterAll(async () => {
+    await close()
+  })
+
+  const getHistory = async () => {
+    return await db.db
+      .selectFrom('repo_commit_history')
+      .selectAll()
+      .orderBy('commit')
+      .orderBy('prev')
+      .execute()
+  }
+
+  const getBlocks = async () => {
+    return await db.db
+      .selectFrom('repo_commit_block')
+      .selectAll()
+      .orderBy('commit')
+      .orderBy('block')
+      .execute()
+  }
+
+  let history: RepoCommitHistory[]
+  let blocks: RepoCommitBlock[]
+
+  it('fetches the current state of the tables', async () => {
+    history = await getHistory()
+    blocks = await getBlocks()
+  })
+
+  it('migrates down', async () => {
+    const migration = await db.migrator.migrateTo('_20221230T215012029Z')
+    expect(migration.error).toBeUndefined()
+    // normal syntax for async exceptions as not catching the error for some reason
+    let err1
+    try {
+      await getHistory()
+    } catch (e) {
+      err1 = e
+    }
+    expect(err1).toBeDefined()
+    let err2
+    try {
+      await getHistory()
+    } catch (e) {
+      err2 = e
+    }
+    expect(err2).toBeDefined()
+  })
+
+  it('migrates up', async () => {
+    const migration = await db.migrator.migrateTo('_20230118T223059239Z')
+    expect(migration.error).toBeUndefined()
+    const migratedHistory = await getHistory()
+    const migratedBlocks = await getBlocks()
+    expect(migratedHistory).toEqual(history)
+    expect(migratedBlocks).toEqual(blocks)
+  })
+})