commit 197a66ae8278ea248e0e7892e5c3e038747f5ac1 · mewsse.pet/pet.mewsse.link

+53 -9
src/ingester.ts
···

       4
       4
        
       

     

       5
       5
        
       import { Client, simpleFetchHandler } from '@atcute/client'

     

       6
       6
        
       import { findUserPDS, getUserDID } from './id-resolver.ts'

     

       7
       7
       -
       import { Jetstream } from '@skyware/jetstream'

     

       7
       7
       +
       import { CommitType, Jetstream } from '@skyware/jetstream'

     

       8
       8
        
       import { RepoReader } from '@atcute/car/v4'

     

       9
       9
        
       import { decode } from '@atcute/cbor'

     

       10
       10
        
       import { logger } from "./lib/logger.ts"

     
···

       112
       112
        
               logger.info(`Inserting record ${link.rkey}`)

     

       113
       113
        
             }

     

       114
       114
        
       

     

       115
       115
       -
             if (!repoRev || repoRev === null) {

     

       116
       116
       -
               logger.error('Backfilling error: no last pds revision found')

     

       117
       117
       -
               return;

     

       118
       118
       -
             }

     

       119
       119
       -
       

     

       120
       120
       -
             if (repoRev != lastRev?.rkey) {

     

       115
       115
       +
             if (repoRev && repoRev != lastRev?.rkey) {

     

       121
       116
        
               await db

     

       122
       117
        
                 .insertInto('revs')

     

       123
       118
        
                 .values({

     
···

       130
       125
        
             logger.info(`Backfilling ended`)

     

       131
       126
        
           },

     

       132
       127
        
       

     

       128
       128
       +
           async setLastRevision(rkey: string, createdAt: string): Promise<void> {

     

       129
       129
       +
               await db

     

       130
       130
       +
                 .deleteFrom('revs')

     

       131
       131
       +
                 .execute()

     

       132
       132
       +
       

     

       133
       133
       +
               await db

     

       134
       134
       +
                 .insertInto('revs')

     

       135
       135
       +
                 .values({

     

       136
       136
       +
                   rkey,

     

       137
       137
       +
                   createdAt

     

       138
       138
       +
                 })

     

       139
       139
       +
                 .execute()

     

       140
       140
       +
           },

     

       141
       141
       +
       

     

       142
       142
       +
           async rollbackRevision(): Promise<void> {

     

       143
       143
       +
               const lastInsert = await db

     

       144
       144
       +
                 .selectFrom('links')

     

       145
       145
       +
                 .select(['rkey', 'createdAt'])

     

       146
       146
       +
                 .orderBy('createdAt', 'desc')

     

       147
       147
       +
                 .limit(1)

     

       148
       148
       +
                 .executeTakeFirst()

     

       149
       149
       +
       

     

       150
       150
       +
               if (lastInsert !== undefined) {

     

       151
       151
       +
                 await db

     

       152
       152
       +
                   .deleteFrom('revs')

     

       153
       153
       +
                   .execute()

     

       154
       154
       +
       

     

       155
       155
       +
                 await db

     

       156
       156
       +
                   .insertInto('revs')

     

       157
       157
       +
                   .values({

     

       158
       158
       +
                     rkey: lastInsert.rkey,

     

       159
       159
       +
                     createdAt: lastInsert.createdAt

     

       160
       160
       +
                   })

     

       161
       161
       +
                   .execute()

     

       162
       162
       +
               }

     

       163
       163
       +
           },

     

       164
       164
       +
       

     

       133
       165
        
           async jetstream(): Promise<Jetstream> {

     

       134
       166
        
             const did = getUserDID()

     

       135
       167
        
             const pds = await findUserPDS()

     

       136
       168
        
       

     

       137
       169
        
             const jetstream = new Jetstream({

     

       138
       138
       -
               wantedCollections: ['pet.mewsse.link'],

     

       139
       139
       -
               wantedDids: [did]

     

       170
       170
       +
               wantedDids: [did],

     

       171
       171
       +
             })

     

       172
       172
       +
       

     

       173
       173
       +
             jetstream.on("commit", async (event) => {

     

       174
       174
       +
               if (event.commit.operation === CommitType.Create) {

     

       175
       175
       +
                 const date = new Date()

     

       176
       176
       +
                 await this.setLastRevision(event.commit.rev, date.toISOString())

     

       177
       177
       +
                 return

     

       178
       178
       +
               }

     

       179
       179
       +
       

     

       180
       180
       +
               if (event.commit.operation === CommitType.Delete) {

     

       181
       181
       +
                 await this.rollbackRevision()

     

       182
       182
       +
                 return

     

       183
       183
       +
               }

     

       140
       184
        
             })

     

       141
       185
        
       

     

       142
       186
        
             jetstream.onCreate('pet.mewsse.link', async (event) => {