commit 13d2e446eefad7cde11257ae43d3c1f5b2f7dfbd · mewsse.pet/pet.mewsse.link

+13

src/db.ts

···

       5
        
       

     

       6
        
       export type DatabaseSchema = {

     

       7
        
           links: Link,

     

       0
        
       
     

       8
        
       }

     

       9
        
       

     

       10
        
       export type Link = {

     
···

       15
        
           tag: string | null,

     

       16
        
           image: string | null,

     

       17
        
           alt: string | null,

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       18
        
           createdAt: string,

     

       19
        
       }

     

       20
        
       

     
···

       38
        
                   .addColumn('alt', 'varchar')

     

       39
        
                   .addColumn('createdAt', 'varchar', (col) => col.notNull())

     

       40
        
                   .execute()

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       41
        
           },

     

       42
        
       

     

       43
        
           async down(db: Kysely<any>) {

     

       44
        
               await db.schema.dropTable('links').execute()

     

       0
        
       
     

       45
        
           },

     

       46
        
       }

     

       47

···

       5
        
       

     

       6
        
       export type DatabaseSchema = {

     

       7
        
           links: Link,

     

       8
       +
           revs: Rev

     

       9
        
       }

     

       10
        
       

     

       11
        
       export type Link = {

     
···

       16
        
           tag: string | null,

     

       17
        
           image: string | null,

     

       18
        
           alt: string | null,

     

       19
       +
           createdAt: string,

     

       20
       +
       }

     

       21
       +
       

     

       22
       +
       export type Rev = {

     

       23
       +
           rkey: string,

     

       24
        
           createdAt: string,

     

       25
        
       }

     

       26
        
       

     
···

       44
        
                   .addColumn('alt', 'varchar')

     

       45
        
                   .addColumn('createdAt', 'varchar', (col) => col.notNull())

     

       46
        
                   .execute()

     

       47
       +
       

     

       48
       +
               await db.schema

     

       49
       +
                   .createTable('revs')

     

       50
       +
                   .addColumn('rkey', 'varchar', (col) => col.primaryKey())

     

       51
       +
                   .addColumn('createdAt', 'varchar', (col) => col.notNull())

     

       52
       +
                   .execute()

     

       53
        
           },

     

       54
        
       

     

       55
        
           async down(db: Kysely<any>) {

     

       56
        
               await db.schema.dropTable('links').execute()

     

       57
       +
               await db.schema.dropTable('rev').execute()

     

       58
        
           },

     

       59
        
       }

     

       60

+39 -15

src/ingester.ts

···

       48
        
                   }

     

       49
        
       

     

       50
        
                   const lastRev = await db

     

       51
       -
                       .selectFrom('links')

     

       52
        
                       .select('rkey')

     

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

     

       54
        
                       .executeTakeFirst()

     
···

       67
        
                   }

     

       68
        
       

     

       69
        
                   await using repo = RepoReader.fromStream(data)

     

       0
        
       
     

       0
        
       
     

       70
        
       

     

       71
        
                   for await (const entry of repo) {

     

       0
        
       
     

       72
        
                       if (entry.collection != "pet.mewsse.link") continue

     

       73
        
                       const link : Link = decode(entry.bytes)

     

       74
        
       

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       75
        
                       await db

     

       76
        
                           .insertInto('links')

     

       77
       -
                           .values({

     

       78
       -
                               rkey: entry.rkey,

     

       79
       -
                               link: link.link,

     

       80
       -
                               title: link.title,

     

       81
       -
                               description: link.description,

     

       82
       -
                               tag: link.tag ?? null,

     

       83
       -
                               image: findImage(did, pds, link),

     

       84
       -
                               alt: link.alt ?? null,

     

       85
       -
                               createdAt: link.createdAt

     

       86
       -
                           })

     

       87
        
                           .onConflict((conflict) => 

     

       88
        
                               conflict.column('rkey').doUpdateSet({

     

       89
        
                                   link: link.link,

     

       90
        
                                   title: link.title,

     

       91
        
                                   description: link.description,

     

       92
       -
                                   tag: link.tag ?? null,

     

       93
       -
                                   image: findImage(did, pds, link),

     

       94
       -
                                   alt: link.alt ?? null,

     

       95
        
                               })

     

       96
        
                           )

     

       97
        
                           .execute()

     

       98
        
                       

     

       99
       -
                       logger.info(`Inserting record ${entry.rkey}`)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       100
        
                   }

     

       101
        
       

     

       102
        
                   logger.info(`Backfilling ended`)

···

       48
        
                   }

     

       49
        
       

     

       50
        
                   const lastRev = await db

     

       51
       +
                       .selectFrom('revs')

     

       52
        
                       .select('rkey')

     

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

     

       54
        
                       .executeTakeFirst()

     
···

       67
        
                   }

     

       68
        
       

     

       69
        
                   await using repo = RepoReader.fromStream(data)

     

       70
       +
                   let links:Array<Link> = []

     

       71
       +
                   let repoRev = null;

     

       72
        
       

     

       73
        
                   for await (const entry of repo) {

     

       74
       +
                       if (!repoRev) repoRev = entry.rkey

     

       75
        
                       if (entry.collection != "pet.mewsse.link") continue

     

       76
        
                       const link : Link = decode(entry.bytes)

     

       77
        
       

     

       78
       +
                       links.push({

     

       79
       +
                           rkey: entry.rkey,

     

       80
       +
                           link: link.link,

     

       81
       +
                           title: link.title,

     

       82
       +
                           description: link.description,

     

       83
       +
                           tag: link.tag ?? null,

     

       84
       +
                           image: findImage(did, pds, link),

     

       85
       +
                           alt: link.alt ?? null,

     

       86
       +
                           createdAt: link.createdAt

     

       87
       +
                       })

     

       88
       +
                   }

     

       89
       +
       

     

       90
       +
                   links = links.sort((a, b) => a.createdAt > b.createdAt ? 1 : -1)

     

       91
       +
       

     

       92
       +
                   for await (const link of links) {

     

       93
        
                       await db

     

       94
        
                           .insertInto('links')

     

       95
       +
                           .values(link)

     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       0
        
       
     

       96
        
                           .onConflict((conflict) => 

     

       97
        
                               conflict.column('rkey').doUpdateSet({

     

       98
        
                                   link: link.link,

     

       99
        
                                   title: link.title,

     

       100
        
                                   description: link.description,

     

       101
       +
                                   tag: link.tag,

     

       102
       +
                                   image: link.image,

     

       103
       +
                                   alt: link.alt,

     

       104
        
                               })

     

       105
        
                           )

     

       106
        
                           .execute()

     

       107
        
                       

     

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

     

       109
       +
                   }

     

       110
       +
       

     

       111
       +
                   if (!repoRev || repoRev === null) {

     

       112
       +
                       logger.error('Backfilling error: no last pds revision found')

     

       113
       +
                       return;

     

       114
       +
                   }

     

       115
       +
       

     

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

     

       117
       +
                       await db

     

       118
       +
                           .insertInto('revs')

     

       119
       +
                           .values({

     

       120
       +
                               rkey: repoRev,

     

       121
       +
                               createdAt: now.toISOString()

     

       122
       +
                           })

     

       123
       +
                           .execute()

     

       124
        
                   }

     

       125
        
       

     

       126
        
                   logger.info(`Backfilling ended`)