Merge pull request #9069 from bluesky-social/migrate-blink

Start migrating blink. Also add better shutdown handling for ogcard

.github/workflows/build-and-push-link-aws.yaml

@@ -1,6 +1,11 @@
 name: build-and-push-link-aws
 on:
   workflow_dispatch:
+  pull_request:
+    paths:
+      - 'bskylink/**'
+      - 'Dockerfile.bskylink'
+      - '.github/workflows/build-and-push-link-aws.yaml'
 
 env:
   REGISTRY: ${{ secrets.AWS_ECR_REGISTRY_USEAST2_PACKAGES_REGISTRY }}
@@ -45,7 +50,7 @@ jobs:
         uses: docker/build-push-action@v4
         with:
           context: .
-          push: ${{ github.event_name != 'pull_request' }}
+          push: true
           file: ./Dockerfile.bskylink
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}

bskylink/src/bin.ts

@@ -1,29 +1,62 @@
 import {Database, envToCfg, httpLogger, LinkService, readEnv} from './index.js'
+
 async function main() {
-  const env = readEnv()
-  const cfg = envToCfg(env)
-  if (cfg.db.migrationUrl) {
-    const migrateDb = Database.postgres({
-      url: cfg.db.migrationUrl,
-      schema: cfg.db.schema,
+  try {
+    httpLogger.info('Starting blink service')
+
+    const env = readEnv()
+    const cfg = envToCfg(env)
+
+    httpLogger.info(
+      {
+        port: cfg.service.port,
+        safelinkEnabled: cfg.service.safelinkEnabled,
+        hasDbUrl: !!cfg.db.url,
+        hasDbMigrationUrl: !!cfg.db.migrationUrl,
+      },
+      'Configuration loaded',
+    )
+
+    if (cfg.db.migrationUrl) {
+      httpLogger.info('Running database migrations')
+      const migrateDb = Database.postgres({
+        url: cfg.db.migrationUrl,
+        schema: cfg.db.schema,
+      })
+      await migrateDb.migrateToLatestOrThrow()
+      await migrateDb.close()
+      httpLogger.info('Database migrations completed')
+    }
+
+    httpLogger.info('Creating LinkService')
+    const link = await LinkService.create(cfg)
+
+    if (link.ctx.cfg.service.safelinkEnabled) {
+      httpLogger.info('Starting Safelink client')
+      link.ctx.safelinkClient.runFetchEvents()
+    }
+
+    await link.start()
+    httpLogger.info('Link service is running')
+
+    process.on('SIGTERM', async () => {
+      httpLogger.info('Link service is stopping')
+      await link.destroy()
+      httpLogger.info('Link service is stopped')
     })
-    await migrateDb.migrateToLatestOrThrow()
-    await migrateDb.close()
+  } catch (error) {
+    httpLogger.error(
+      {
+        error: String(error),
+        stack: error instanceof Error ? error.stack : undefined,
+      },
+      'Failed to start blink service',
+    )
+    process.exit(1)
   }
-
-  const link = await LinkService.create(cfg)
-
-  if (link.ctx.cfg.service.safelinkEnabled) {
-    link.ctx.safelinkClient.runFetchEvents()
-  }
-
-  await link.start()
-  httpLogger.info('link service is running')
-  process.on('SIGTERM', async () => {
-    httpLogger.info('link service is stopping')
-    await link.destroy()
-    httpLogger.info('link service is stopped')
-  })
 }
 
-main()
+main().catch(error => {
+  console.error('Unhandled startup error:', error)
+  process.exit(1)
+})

bskylink/src/db/index.ts

@@ -34,6 +34,16 @@ export class Database {
 
   static postgres(opts: PgOptions): Database {
     const {schema, url, txLockNonce} = opts
+    log.info(
+      {
+        schema,
+        poolSize: opts.poolSize,
+        poolMaxUses: opts.poolMaxUses,
+        poolIdleTimeoutMs: opts.poolIdleTimeoutMs,
+      },
+      'Creating database connection',
+    )
+
     const pool =
       opts.pool ??
       new Pg.Pool({

bskyogcard/src/index.ts

@@ -62,7 +62,10 @@ export class CardService {
     // Start main application server
     this.server = this.app.listen(this.ctx.cfg.service.port)
     this.server.keepAliveTimeout = 90000
-    this.terminator = createHttpTerminator({server: this.server})
+    this.terminator = createHttpTerminator({
+      server: this.server,
+      gracefulTerminationTimeout: 15000, // 15s timeout for in-flight requests
+    })
     await events.once(this.server, 'listening')
 
     // Start separate metrics server
@@ -73,13 +76,32 @@ export class CardService {
     })
 
     this.metricsServer = metricsApp.listen(this.ctx.cfg.service.metricsPort)
-    this.metricsTerminator = createHttpTerminator({server: this.metricsServer})
+    this.metricsTerminator = createHttpTerminator({
+      server: this.metricsServer,
+      gracefulTerminationTimeout: 2000, // 2s timeout for metrics server
+    })
     await events.once(this.metricsServer, 'listening')
   }
 
   async destroy() {
+    const startTime = Date.now()
+
     this.ctx.abortController.abort()
-    await this.terminator?.terminate()
-    await this.metricsTerminator?.terminate()
+
+    const shutdownPromises = []
+
+    if (this.terminator) {
+      shutdownPromises.push(this.terminator.terminate())
+    }
+
+    if (this.metricsTerminator) {
+      shutdownPromises.push(this.metricsTerminator.terminate())
+    }
+
+    await Promise.all(shutdownPromises)
+
+    const elapsed = Date.now() - startTime
+    const {httpLogger} = await import('./logger.js')
+    httpLogger.info(`Graceful shutdown completed in ${elapsed}ms`)
   }
 }