···

       
1
       
-
       
import { Kysely, Migrator, PostgresDialect, sql } from 'kysely'

     

       
2
       
 
       
import { Pool as PgPool, types as pgTypes } from 'pg'

     

       
3
       
 
       
import { CID } from 'multiformats/cid'

     

       
4
       
 
       
import { cidForCbor, check } from '@atproto/common'

     
···

       
106
       
 
       
          .insertInto('operations')

     

       
107
       
 
       
          .values({

     

       
108
       
 
       
            did,

     

       
109
       
-
       
            operation: JSON.stringify(proposed),

     

       
110
       
 
       
            cid: cid.toString(),

     

       
111
       
-
       
            nullified: 0,

     

       
112
       
-
       
            createdAt: new Date().toISOString(),

     

       
113
       
 
       
          })

     

       
114
       
 
       
          .execute()

     

       
115
       
 
       


     
···

       
117
       
 
       
          const nullfiedStrs = nullified.map((cid) => cid.toString())

     

       
118
       
 
       
          await tx

     

       
119
       
 
       
            .updateTable('operations')

     

       
120
       
-
       
            .set({ nullified: 1 })

     

       
121
       
 
       
            .where('did', '=', did)

     

       
122
       
 
       
            .where('cid', 'in', nullfiedStrs)

     

       
123
       
 
       
            .execute()

     
···

       
129
       
 
       
          .selectFrom('operations')

     

       
130
       
 
       
          .select('cid')

     

       
131
       
 
       
          .where('did', '=', did)

     

       
132
       
-
       
          .where('nullified', '=', 0)

     

       
133
       
 
       
          .orderBy('createdAt', 'desc')

     

       
134
       
 
       
          .limit(2)

     

       
135
       
 
       
          .execute()

     
···

       
152
       
 
       
      .selectFrom('operations')

     

       
153
       
 
       
      .select('cid')

     

       
154
       
 
       
      .where('did', '=', did)

     

       
155
       
-
       
      .where('nullified', '=', 0)

     

       
156
       
 
       
      .where('cid', 'not in', notIncludedStr)

     

       
157
       
 
       
      .orderBy('createdAt', 'desc')

     

       
158
       
 
       
      .executeTakeFirst()

     
···

       
174
       
 
       
      .selectFrom('operations')

     

       
175
       
 
       
      .selectAll()

     

       
176
       
 
       
      .where('did', '=', did)

     

       
177
       
-
       
      .where('nullified', '=', 0)

     

       
178
       
 
       
      .orderBy('createdAt', 'asc')

     

       
179
       
 
       
      .execute()

     

       
180
       
 
       


     

       
181
       
 
       
    return res.map((row) => ({

     

       
182
       
 
       
      did: row.did,

     

       
183
       
-
       
      operation: JSON.parse(row.operation),

     

       
184
       
 
       
      cid: CID.parse(row.cid),

     

       
185
       
-
       
      nullified: row.nullified === 1,

     

       
186
       
-
       
      createdAt: new Date(row.createdAt),

     

       
187
       
 
       
    }))

     

       
188
       
 
       
  }

     

       
189
       
 
       


     

       
190
       
 
       
  async fullExport(): Promise<Record<string, OpLogExport>> {

     

       
191
       
-
       
    const res = await this.db

     

       
192
       
-
       
      .selectFrom('operations')

     

       
193
       
-
       
      .selectAll()

     

       
194
       
-
       
      .orderBy('did')

     

       
195
       
-
       
      .orderBy('createdAt')

     

       
196
       
-
       
      .execute()

     

       
197
       
-
       
    return res.reduce((acc, cur) => {

     

       
198
       
-
       
      acc[cur.did] ??= []

     

       
199
       
-
       
      acc[cur.did].push({

     

       
200
       
-
       
        op: JSON.parse(cur.operation),

     

       
201
       
-
       
        nullified: cur.nullified === 1,

     

       
202
       
-
       
        createdAt: cur.createdAt,

     

       
203
       
-
       
      })

     

       
204
       
-
       
      return acc

     

       
205
       
-
       
    }, {} as Record<string, OpLogExport>)

     

       
0
       
 
       

     

       
206
       
 
       
  }

     

       
207
       
 
       
}

     

       
208
       
 
       


     
···

       
210
       
 
       


     

       
211
       
 
       
interface OperationsTable {

     

       
212
       
 
       
  did: string

     

       
213
       
-
       
  operation: string

     

       
214
       
 
       
  cid: string

     

       
215
       
-
       
  nullified: 0 | 1

     

       
216
       
-
       
  createdAt: string

     

       
217
       
 
       
}

     

       
218
       
 
       


     

       
219
       
 
       
interface DatabaseSchema {

     

···

       
1
       
+
       
import { Generated, Kysely, Migrator, PostgresDialect, sql } from 'kysely'

     

       
2
       
 
       
import { Pool as PgPool, types as pgTypes } from 'pg'

     

       
3
       
 
       
import { CID } from 'multiformats/cid'

     

       
4
       
 
       
import { cidForCbor, check } from '@atproto/common'

     
···

       
106
       
 
       
          .insertInto('operations')

     

       
107
       
 
       
          .values({

     

       
108
       
 
       
            did,

     

       
109
       
+
       
            operation: proposed,

     

       
110
       
 
       
            cid: cid.toString(),

     

       
111
       
+
       
            nullified: false,

     

       
0
       
 
       

     

       
112
       
 
       
          })

     

       
113
       
 
       
          .execute()

     

       
114
       
 
       


     
···

       
116
       
 
       
          const nullfiedStrs = nullified.map((cid) => cid.toString())

     

       
117
       
 
       
          await tx

     

       
118
       
 
       
            .updateTable('operations')

     

       
119
       
+
       
            .set({ nullified: true })

     

       
120
       
 
       
            .where('did', '=', did)

     

       
121
       
 
       
            .where('cid', 'in', nullfiedStrs)

     

       
122
       
 
       
            .execute()

     
···

       
128
       
 
       
          .selectFrom('operations')

     

       
129
       
 
       
          .select('cid')

     

       
130
       
 
       
          .where('did', '=', did)

     

       
131
       
+
       
          .where('nullified', '=', false)

     

       
132
       
 
       
          .orderBy('createdAt', 'desc')

     

       
133
       
 
       
          .limit(2)

     

       
134
       
 
       
          .execute()

     
···

       
151
       
 
       
      .selectFrom('operations')

     

       
152
       
 
       
      .select('cid')

     

       
153
       
 
       
      .where('did', '=', did)

     

       
154
       
+
       
      .where('nullified', '=', false)

     

       
155
       
 
       
      .where('cid', 'not in', notIncludedStr)

     

       
156
       
 
       
      .orderBy('createdAt', 'desc')

     

       
157
       
 
       
      .executeTakeFirst()

     
···

       
173
       
 
       
      .selectFrom('operations')

     

       
174
       
 
       
      .selectAll()

     

       
175
       
 
       
      .where('did', '=', did)

     

       
176
       
+
       
      .where('nullified', '=', false)

     

       
177
       
 
       
      .orderBy('createdAt', 'asc')

     

       
178
       
 
       
      .execute()

     

       
179
       
 
       


     

       
180
       
 
       
    return res.map((row) => ({

     

       
181
       
 
       
      did: row.did,

     

       
182
       
+
       
      operation: row.operation,

     

       
183
       
 
       
      cid: CID.parse(row.cid),

     

       
184
       
+
       
      nullified: row.nullified,

     

       
185
       
+
       
      createdAt: row.createdAt,

     

       
186
       
 
       
    }))

     

       
187
       
 
       
  }

     

       
188
       
 
       


     

       
189
       
 
       
  async fullExport(): Promise<Record<string, OpLogExport>> {

     

       
190
       
+
       
    return {}

     

       
191
       
+
       
    //   const res = await this.db

     

       
192
       
+
       
    //     .selectFrom('operations')

     

       
193
       
+
       
    //     .selectAll()

     

       
194
       
+
       
    //     .orderBy('did')

     

       
195
       
+
       
    //     .orderBy('createdAt')

     

       
196
       
+
       
    //     .execute()

     

       
197
       
+
       
    //   return res.reduce((acc, cur) => {

     

       
198
       
+
       
    //     acc[cur.did] ??= []

     

       
199
       
+
       
    //     acc[cur.did].push({

     

       
200
       
+
       
    //       op: cur.operation),

     

       
201
       
+
       
    //       nullified: cur.nullified === 1,

     

       
202
       
+
       
    //       createdAt: cur.createdAt,

     

       
203
       
+
       
    //     })

     

       
204
       
+
       
    //     return acc

     

       
205
       
+
       
    //   }, {} as Record<string, OpLogExport>)

     

       
206
       
 
       
  }

     

       
207
       
 
       
}

     

       
208
       
 
       


     
···

       
210
       
 
       


     

       
211
       
 
       
interface OperationsTable {

     

       
212
       
 
       
  did: string

     

       
213
       
+
       
  operation: plc.CompatibleOpOrTombstone

     

       
214
       
 
       
  cid: string

     

       
215
       
+
       
  nullified: boolean

     

       
216
       
+
       
  createdAt: Generated<Date>

     

       
217
       
 
       
}

     

       
218
       
 
       


     

       
219
       
 
       
interface DatabaseSchema {

     

···

       
22
       
 
       
    createdAt: row.createdAt,

     

       
23
       
 
       
  }))

     

       
24
       
 
       


     

       
25
       
-
       
  await db.insertInto('operations_new').values(vals).execute()

     

       
0
       
 
       

     

       
0
       
 
       

     

       
26
       
 
       


     

       
27
       
 
       
  await db.schema.dropTable('operations').execute()

     

       
28
       
 
       


     

···

       
22
       
 
       
    createdAt: row.createdAt,

     

       
23
       
 
       
  }))

     

       
24
       
 
       


     

       
25
       
+
       
  if (vals.length > 0) {

     

       
26
       
+
       
    await db.insertInto('operations_new').values(vals).execute()

     

       
27
       
+
       
  }

     

       
28
       
 
       


     

       
29
       
 
       
  await db.schema.dropTable('operations').execute()

     

       
30
       
 
       


     

···

       
1
       
 
       
import { cidForCbor, DAY } from '@atproto/common'

     

       
2
       
 
       
import { Secp256k1Keypair } from '@atproto/crypto'

     

       
3
       
 
       
import * as plc from '@did-plc/lib'

     

       
0
       
 
       

     

       
4
       
 
       
import { Database } from '../../src'

     

       
5
       
 
       


     

       
6
       
 
       
describe('refactor migration', () => {

     

       
7
       
 
       
  let db: Database

     

       
0
       
 
       

     

       
8
       
 
       


     

       
9
       
 
       
  beforeAll(async () => {

     

       
10
       
 
       
    const dbPostgresUrl = process.env.DB_POSTGRES_URL

     
···

       
17
       
 
       
    })

     

       
18
       
 
       


     

       
19
       
 
       
    await db.migrateToOrThrow('_20221020T204908820Z')

     

       
0
       
 
       

     

       
20
       
 
       
  })

     

       
21
       
 
       


     

       
22
       
 
       
  afterAll(async () => {

     
···

       
53
       
 
       
        createdAt: time,

     

       
54
       
 
       
      })

     

       
55
       
 
       
    }

     

       
56
       
-
       
    await db.db.insertInto('operations').values(ops).execute()

     

       
57
       
 
       


     

       
58
       
-
       
    before = await db.db

     

       
59
       
 
       
      .selectFrom('operations')

     

       
60
       
 
       
      .selectAll()

     

       
61
       
 
       
      .orderBy('did', 'asc')

     
···

       
67
       
 
       
  })

     

       
68
       
 
       


     

       
69
       
 
       
  it('correctly migrated all data', async () => {

     

       
70
       
-
       
    const migrated = await db.db

     

       
71
       
 
       
      .selectFrom('operations')

     

       
72
       
 
       
      .selectAll()

     

       
73
       
 
       
      .orderBy('did', 'asc')

     
···

       
87
       
 
       


     

       
88
       
 
       
  it('migrates down', async () => {

     

       
89
       
 
       
    await db.migrateToOrThrow('_20221020T204908820Z')

     

       
90
       
-
       
    const migratedBack = await db.db

     

       
91
       
 
       
      .selectFrom('operations')

     

       
92
       
 
       
      .selectAll()

     

       
93
       
 
       
      .orderBy('did', 'asc')

     

···

       
1
       
 
       
import { cidForCbor, DAY } from '@atproto/common'

     

       
2
       
 
       
import { Secp256k1Keypair } from '@atproto/crypto'

     

       
3
       
 
       
import * as plc from '@did-plc/lib'

     

       
4
       
+
       
import { Kysely } from 'kysely'

     

       
5
       
 
       
import { Database } from '../../src'

     

       
6
       
 
       


     

       
7
       
 
       
describe('refactor migration', () => {

     

       
8
       
 
       
  let db: Database

     

       
9
       
+
       
  let rawDb: Kysely<any>

     

       
10
       
 
       


     

       
11
       
 
       
  beforeAll(async () => {

     

       
12
       
 
       
    const dbPostgresUrl = process.env.DB_POSTGRES_URL

     
···

       
19
       
 
       
    })

     

       
20
       
 
       


     

       
21
       
 
       
    await db.migrateToOrThrow('_20221020T204908820Z')

     

       
22
       
+
       
    rawDb = db.db

     

       
23
       
 
       
  })

     

       
24
       
 
       


     

       
25
       
 
       
  afterAll(async () => {

     
···

       
56
       
 
       
        createdAt: time,

     

       
57
       
 
       
      })

     

       
58
       
 
       
    }

     

       
59
       
+
       
    await rawDb.insertInto('operations').values(ops).execute()

     

       
60
       
 
       


     

       
61
       
+
       
    before = await rawDb

     

       
62
       
 
       
      .selectFrom('operations')

     

       
63
       
 
       
      .selectAll()

     

       
64
       
 
       
      .orderBy('did', 'asc')

     
···

       
70
       
 
       
  })

     

       
71
       
 
       


     

       
72
       
 
       
  it('correctly migrated all data', async () => {

     

       
73
       
+
       
    const migrated = await rawDb

     

       
74
       
 
       
      .selectFrom('operations')

     

       
75
       
 
       
      .selectAll()

     

       
76
       
 
       
      .orderBy('did', 'asc')

     
···

       
90
       
 
       


     

       
91
       
 
       
  it('migrates down', async () => {

     

       
92
       
 
       
    await db.migrateToOrThrow('_20221020T204908820Z')

     

       
93
       
+
       
    const migratedBack = await rawDb

     

       
94
       
 
       
      .selectFrom('operations')

     

       
95
       
 
       
      .selectAll()

     

       
96
       
 
       
      .orderBy('did', 'asc')