fix: db migration issue

docker-compose.yml

@@ -14,10 +14,10 @@ services:
       - postgres-data:/var/lib/postgresql/data
     healthcheck:
       test: ['CMD-SHELL', 'pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB']
-      interval: 5s       # Check every 5 seconds for readiness
-      timeout: 5s        # Allow up to 5 seconds for a response
-      retries: 3         # Fail after 3 unsuccessful attempts
-      start_period: 10s  # Start checks after 10 seconds
+      interval: 5s # Check every 5 seconds for readiness
+      timeout: 5s # Allow up to 5 seconds for a response
+      retries: 3 # Fail after 3 unsuccessful attempts
+      start_period: 10s # Start checks after 10 seconds
     networks:
       - indexer-network
 
@@ -31,33 +31,33 @@ services:
       - '6379:6379'
     healthcheck:
       test: ['CMD-SHELL', 'redis-cli ping']
-      interval: 5s       # Check every 5 seconds
-      timeout: 5s        # Allow up to 5 seconds for a response
-      retries: 3         # Fail after 3 unsuccessful attempts
-      start_period: 5s   # Start health checks after 5 seconds
+      interval: 5s # Check every 5 seconds
+      timeout: 5s # Allow up to 5 seconds for a response
+      retries: 3 # Fail after 3 unsuccessful attempts
+      start_period: 5s # Start health checks after 5 seconds
     networks:
       - indexer-network
 
-  lazy-indexer:
-    build: .
-    restart: unless-stopped
-    depends_on:
-      postgres:
-        condition: service_healthy
-      redis:
-        condition: service_healthy
-    environment:
-      - DATABASE_URL=postgresql://indexer:password@postgres:5432/indexer
-      - REDIS_URL=redis://redis:6379
-      - HUB_REST_URL=https://hub.pinata.cloud
-      - HUB_RPC=hub-grpc.pinata.cloud
-      - HUB_SSL=true
-      - WORKER_CONCURRENCY=5
-      - LOG_LEVEL=debug
-    networks:
-      - indexer-network
-    command: >
-      /bin/sh -c "yarn backfill & yarn stream"
+  # lazy-indexer:
+  #   build: .
+  #   restart: unless-stopped
+  #   depends_on:
+  #     postgres:
+  #       condition: service_healthy
+  #     redis:
+  #       condition: service_healthy
+  #   environment:
+  #     - DATABASE_URL=postgresql://indexer:password@postgres:5432/indexer
+  #     - REDIS_URL=redis://redis:6379
+  #     - HUB_REST_URL=https://hub.pinata.cloud
+  #     - HUB_RPC=hub-grpc.pinata.cloud
+  #     - HUB_SSL=true
+  #     - WORKER_CONCURRENCY=5
+  #     - LOG_LEVEL=debug
+  #   networks:
+  #     - indexer-network
+  #   command: >
+  #     /bin/sh -c "yarn backfill & yarn stream"
 
 volumes:
   postgres-data:

src/api/cast.ts

@@ -4,7 +4,7 @@ import {
   isCastRemoveMessage,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { getBackfillQueue, queueBackfillJob } from '../lib/backfill.js'
 import { log } from '../lib/logger.js'
 import {

src/api/fid.ts

@@ -5,7 +5,7 @@ import {
   isIdRegisterOnChainEvent,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { hubClient } from '../lib/hub-client.js'
 import { getOnChainEventsByFidInBatchesOf } from '../lib/paginate.js'
 import { MAX_PAGE_SIZE, NULL_ETH_ADDRESS } from '../lib/utils.js'

src/api/hub.ts

@@ -1,6 +1,6 @@
 import { ContactInfoContentBody } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { log } from '../lib/logger.js'
 import { breakIntoChunks, formatHubs } from '../lib/utils.js'
 

src/api/link.ts

@@ -4,7 +4,7 @@ import {
   isLinkRemoveMessage,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { getBackfillQueue, queueBackfillJob } from '../lib/backfill.js'
 import { log } from '../lib/logger.js'
 import { isRootTarget } from '../lib/targets.js'

src/api/reaction.ts

@@ -4,7 +4,7 @@ import {
   isReactionRemoveMessage,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { log } from '../lib/logger.js'
 import {
   breakIntoChunks,

src/api/signer.ts

@@ -6,7 +6,7 @@ import {
 } from '@farcaster/hub-nodejs'
 import { bytesToHex, decodeAbiParameters } from 'viem'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { hubClient } from '../lib/hub-client.js'
 import { getOnChainEventsByFidInBatchesOf } from '../lib/paginate.js'
 import { MAX_PAGE_SIZE } from '../lib/utils.js'

src/api/storage.ts

@@ -4,7 +4,7 @@ import {
   isStorageRentOnChainEvent,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { hubClient } from '../lib/hub-client.js'
 import { getOnChainEventsByFidInBatchesOf } from '../lib/paginate.js'
 import { MAX_PAGE_SIZE, farcasterTimeToDate } from '../lib/utils.js'

src/api/user-data.ts

@@ -1,6 +1,6 @@
 import { Message } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { log } from '../lib/logger.js'
 import { formatUserDatas } from '../lib/utils.js'
 

src/api/verification.ts

@@ -4,7 +4,7 @@ import {
   isVerificationRemoveMessage,
 } from '@farcaster/hub-nodejs'
 
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { log } from '../lib/logger.js'
 import { farcasterTimeToDate, formatVerifications } from '../lib/utils.js'
 

src/db/db.ts

@@ -0,0 +1,40 @@
+import 'dotenv/config'
+import { CamelCasePlugin, Kysely, PostgresDialect } from 'kysely'
+import pg from 'pg'
+
+import { Tables } from './db.types'
+
+const { Pool } = pg
+
+export const getDbClient = (
+  connectionString: string | undefined = process.env.DATABASE_URL
+) => {
+  if (!connectionString) {
+    throw new Error('DATABASE_URL is not set')
+  }
+
+  return new Kysely<Tables>({
+    dialect: new PostgresDialect({
+      pool: new Pool({
+        max: 20,
+        connectionString,
+      }),
+    }),
+    plugins: [new CamelCasePlugin()],
+    log: ['error'],
+  })
+}
+
+export const db = getDbClient()
+
+// export const db = new Kysely<Tables>({
+//   dialect: new PostgresDialect({
+//     pool: new Pool({
+//       connectionString: process.env.DATABASE_URL,
+//       log(...messages) {
+//         console.log('DB:', ...messages)
+//       },
+//     }),
+//   }),
+//   plugins: [new CamelCasePlugin()],
+// })

src/db/migrator.ts

@@ -5,9 +5,11 @@ import * as path from 'path'
 import { fileURLToPath } from 'url'
 
 import { log } from '../lib/logger.js'
-import { db } from './kysely.js'
+import { getDbClient } from './db.js'
 
 export async function migrateToLatest() {
+  const db = getDbClient()
+
   const migrator = new Migrator({
     db,
     provider: new FileMigrationProvider({

src/index.ts

@@ -1,12 +1,12 @@
 import 'dotenv/config'
 
 import { getLatestEvent } from './api/event.js'
+import { migrateToLatest } from './db/migrator.js'
 import { backfill } from './lib/backfill.js'
 import { initExpressApp } from './lib/express.js'
 import { subscribe } from './lib/subscriber.js'
-import { migrateToLatest } from './db/migrator.js'
 
-migrateToLatest()
+await migrateToLatest()
 
 if (process.argv[2] === '--backfill') {
   initExpressApp()

src/lib/links-utils.ts

@@ -15,6 +15,7 @@ import fastq from 'fastq'
 import { createPublicClient, http, parseAbi } from 'viem'
 import { optimism } from 'viem/chains'
 
+import { log } from './logger.js'
 import { getAllLinksByFid } from './paginate.js'
 
 const HUB_URL = process.env.HUB_REST_URL!
@@ -218,6 +219,8 @@ export async function _getAllLinksByFid(
       })
     : await getAllLinksByFid({ fid })
 
+  log.info(`Got ${linksMessages.length} link messages for fid ${fid}`)
+
   const linksSet = new Set<number>()
 
   linksMessages.forEach((linkMessage) =>

src/lib/targets.ts

@@ -1,4 +1,4 @@
-import { db } from '../db/kysely.js'
+import { db } from '../db/db.js'
 import { redis } from './redis.js'
 
 export const rootTargetsKey = 'targets:root'