feat: queue reliability hardening and migration verification checks

.github/workflows/ci.yml

@@ -100,3 +100,21 @@ jobs:
 
       - name: Run tests
         run: npm run test
+
+  # Fast static verification of the committed migrations: checks that every
+  # migration directory contains a non-empty, structurally valid migration.sql
+  # and that directory names follow the Prisma naming convention. Runs on every
+  # pull request and needs no PostgreSQL connection, so it also catches broken
+  # migrations in containerized runners without database services.
+  migrations-static:
+    name: migrations (static)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 22
+          cache: npm
+      - run: npm ci --include=dev
+      - name: Verify migration files (no database required)
+        run: npm run db:verify:static

CONTRIBUTING.md

@@ -42,8 +42,39 @@ its own microservice.
 2. `npm run db:verify` passes (`scripts/verify-migrations.sh`). See
    [Database migrations](#database-migrations) below.
 3. Database schema changes include a valid Prisma migration folder containing `migration.sql`.
-4. New endpoints are documented with OpenAPI/Swagger decorators.
-5. Cross-repo contracts (response envelope, entity/enum names) still match `astroid-web` and `astroid-sdk`.
+4. `npm run db:verify:static` passes — see [Database migration verification](#database-migration-verification).
+5. New endpoints are documented with OpenAPI/Swagger decorators.
+6. Cross-repo contracts (response envelope, entity/enum names) still match `astroid-web` and `astroid-sdk`.
+
+## Database migration verification
+
+Schema changes in Prisma are strictly verified before they reach production so a
+malformed or drifted migration can never break a deployment. CI runs the static
+check on every pull request, and the database-backed verification runs whenever
+`DATABASE_URL` is available (locally or in CI):
+
+| Job | Requires a database | What it checks |
+|---|---|---|
+| `migrations (static)` (`npm run db:verify:static`) | No | Every migration directory has a non-empty `migration.sql` containing executable SQL (not only comments), quotes/parentheses are balanced, and directory names follow the Prisma convention `<UTC-timestamp>_<snake_case_name>` |
+| `migrations (database)` (`npm run db:verify`) | Yes | Migrations apply cleanly to a fresh PostgreSQL instance and the schema rebuilt from the migrations alone matches `prisma/schema.prisma` (no drift) |
+
+### Naming convention
+
+Migration directories **must** be named `<UTC-timestamp>_<snake_case_name>`
+(e.g. `20260830174000_sync_schema`). CI fails the build otherwise. Create
+migrations with `npm run prisma:migrate` — never by hand.
+
+### Locally
+
+```bash
+npm run db:verify:static   # fast static checks, no database needed
+npm run db:verify          # full verification (needs DATABASE_URL, and
+                           # SHADOW_DATABASE_URL for the drift check)
+```
+
+The static mode is what containerized CI runners without an active PostgreSQL
+connection use; the full script degrades to it automatically when `DATABASE_URL`
+is unset.
 
 ## Database migrations
 

package.json

@@ -26,7 +26,8 @@
     "prisma:seed": "ts-node prisma/seed.ts",
     "db:seed": "ts-node prisma/seed.ts",
     "db:verify": "bash scripts/verify-migrations.sh",
-    "db:migrate": "ts-node src/database/migrate.cli.ts"
+    "db:migrate": "ts-node src/database/migrate.cli.ts",
+    "db:verify:static": "DATABASE_URL= SHADOW_DATABASE_URL= bash scripts/verify-migrations.sh"
   },
   "prisma": {
     "seed": "ts-node prisma/seed.ts"

src/modules/audit/audit.module.ts

@@ -20,6 +20,19 @@ import { AuditCleanupQueue } from './queues/audit-cleanup.queue';
         db: redisConfig().db,
       },
     }),
+    // The dedicated `audit` queue persists audit entries asynchronously via the
+    // AuditWorker (src/workers/audit.worker.ts). Bounded retries with
+    // exponential backoff ride out transient database outages without dropping
+    // entries; terminal failures land in the dead-letter queue.
+    BullModule.registerQueue({
+      name: Queues.Audit,
+      defaultJobOptions: {
+        attempts: 5,
+        backoff: { type: 'exponential', delay: 1000 },
+        removeOnComplete: { count: 1000 },
+        removeOnFail: { age: 7 * 24 * 3600 },
+      },
+    }),
     BullModule.registerQueue({
       name: Queues.AuditCleanup,
       defaultJobOptions: {

src/modules/health/health.controller.ts

@@ -1,14 +1,20 @@
-import { Controller, Get, Res, HttpStatus } from '@nestjs/common';
-import { ApiOperation, ApiResponse, ApiTags } from '@nestjs/swagger';
+import { Controller, Get, HttpException, HttpStatus, Res, UseGuards } from '@nestjs/common';
+import { ApiOperation, ApiResponse, ApiTags, ApiBearerAuth } from '@nestjs/swagger';
 import { HealthIndicatorResult } from '@nestjs/terminus';
 import { SkipThrottle } from '@nestjs/throttler';
 import { Response } from 'express';
 import { Public } from '../../common/decorators/public.decorator';
 import { SkipAudit } from '../../common/decorators/skip-audit.decorator';
+import { JwtAuthGuard } from '../../common/guards/jwt-auth.guard';
+import { RolesGuard } from '../../common/guards/roles.guard';
+import { Roles } from '../../common/decorators/roles.decorator';
+import { UserRole } from '@prisma/client';
+import { SkipPublicRateLimit } from '../../common/decorators/skip-public-rate-limit.decorator';
 import { PrismaHealthIndicator } from './indicators/prisma.health';
 import { RedisHealthIndicator } from './indicators/redis.health';
 import { StellarHealthIndicator } from './indicators/stellar.health';
 import { DatabaseMigrationHealthIndicator } from './indicators/database-migration.health';
+import { BullMQHealthIndicator, QueuesHealthReport } from './indicators/bullmq.health';
 
 /** Per-dependency report shape returned under `services` in the readiness body. */
 interface ReadinessServiceReport {
@@ -98,7 +104,7 @@ export class HealthController {
     };
   }
 
-  @Get(['ready', 'readiness'])
+  @Get('readiness')
   @ApiOperation({ summary: 'Application readiness check' })
   @ApiResponse({ status: 200, description: 'Application is ready' })
   @ApiResponse({ status: 503, description: 'Application is not ready' })
@@ -192,3 +198,52 @@ function unwrap(result: HealthIndicatorResult, key: string): ReadinessServiceRep
     ...(message ?? error ? { error: message ?? error } : {}),
   };
 }
+
+/**
+ * BullMQ queue diagnostics. Separate from `HealthController` so the queue
+ * internals stay protected: this controller is NOT marked `@Public()` and
+ * requires an authenticated OWNER, ADMIN, DEVELOPER or AUDITOR.
+ */
+@ApiTags('health')
+@Controller('health')
+@SkipAudit()
+@SkipThrottle({ api: true, auth: true })
+@SkipPublicRateLimit()
+export class QueuesHealthController {
+  constructor(private readonly bullmqHealthIndicator: BullMQHealthIndicator) {}
+
+  @Get('queues')
+  @UseGuards(JwtAuthGuard, RolesGuard)
+  @Roles(UserRole.OWNER, UserRole.ADMIN, UserRole.DEVELOPER, UserRole.AUDITOR)
+  @ApiBearerAuth('access-token')
+  @ApiOperation({
+    summary: 'BullMQ queue health check',
+    description:
+      'Inspects every registered BullMQ queue (notifications, webhooks, ' +
+      'stellar-sync, analytics, reports, outbox-events, stellar-fee-bump, ' +
+      'transactions, risk-analysis, dead-letter, audit-cleanup, audit) and ' +
+      'returns waiting/active/failed/delayed/completed/paused job counts plus ' +
+      'Redis connectivity. Used by Kubernetes probes and dashboards to monitor ' +
+      'asynchronous worker health.',
+  })
+  @ApiResponse({ status: 200, description: 'Per-queue job counts and Redis connectivity' })
+  @ApiResponse({ status: 401, description: 'Not authenticated' })
+  @ApiResponse({ status: 403, description: 'Insufficient permissions' })
+  @ApiResponse({ status: 503, description: 'Redis unreachable or all queues failing' })
+  async checkQueuesHealth(): Promise<QueuesHealthReport> {
+    const report = await this.bullmqHealthIndicator.checkHealth();
+
+    if (report.status === 'down') {
+      throw new HttpException(
+        {
+          statusCode: 503,
+          message: 'Redis unreachable or all BullMQ queues failing health probes',
+          report,
+        },
+        HttpStatus.SERVICE_UNAVAILABLE,
+      );
+    }
+
+    return report;
+  }
+}

src/modules/health/health.module.ts

@@ -1,28 +1,31 @@
-import { Module } from '@nestjs/common';
-import { TerminusModule } from '@nestjs/terminus';
-import { HealthController } from './health.controller';
-import { PrismaHealthIndicator } from './indicators/prisma.health';
-import { RedisHealthIndicator } from './indicators/redis.health';
-import { StellarHealthIndicator } from './indicators/stellar.health';
-import { DatabaseMigrationHealthIndicator } from './indicators/database-migration.health';
-import { DatabaseModule } from '../../database/database.module';
-
-@Module({
-  // TerminusModule supplies `HealthCheckService` and the indicator base class
-  // used by PrismaHealthIndicator.
-  imports: [DatabaseModule, TerminusModule],
-  controllers: [HealthController],
-  providers: [
-    PrismaHealthIndicator,
-    RedisHealthIndicator,
-    StellarHealthIndicator,
-    DatabaseMigrationHealthIndicator,
-  ],
-  exports: [
-    PrismaHealthIndicator,
-    RedisHealthIndicator,
-    StellarHealthIndicator,
-    DatabaseMigrationHealthIndicator,
-  ],
-})
-export class HealthModule {}
+import { Module } from '@nestjs/common';
+import { TerminusModule } from '@nestjs/terminus';
+import { HealthController, QueuesHealthController } from './health.controller';
+import { PrismaHealthIndicator } from './indicators/prisma.health';
+import { RedisHealthIndicator } from './indicators/redis.health';
+import { StellarHealthIndicator } from './indicators/stellar.health';
+import { DatabaseMigrationHealthIndicator } from './indicators/database-migration.health';
+import { BullMQHealthIndicator } from './indicators/bullmq.health';
+import { DatabaseModule } from '../../database/database.module';
+
+@Module({
+  // TerminusModule supplies `HealthCheckService` and the indicator base class
+  // used by PrismaHealthIndicator.
+  imports: [DatabaseModule, TerminusModule],
+  controllers: [HealthController, QueuesHealthController],
+  providers: [
+    PrismaHealthIndicator,
+    RedisHealthIndicator,
+    StellarHealthIndicator,
+    DatabaseMigrationHealthIndicator,
+    BullMQHealthIndicator,
+  ],
+  exports: [
+    PrismaHealthIndicator,
+    RedisHealthIndicator,
+    StellarHealthIndicator,
+    DatabaseMigrationHealthIndicator,
+    BullMQHealthIndicator,
+  ],
+})
+export class HealthModule {}

src/modules/health/index.ts

@@ -1,6 +1,7 @@
-export * from './health.module';
-export * from './health.controller';
-export * from './indicators/prisma.health';
-export * from './indicators/redis.health';
-export * from './indicators/stellar.health';
-export * from './indicators/database-migration.health';
+export * from './health.module';
+export * from './health.controller';
+export * from './indicators/prisma.health';
+export * from './indicators/redis.health';
+export * from './indicators/stellar.health';
+export * from './indicators/database-migration.health';
+export * from './indicators/bullmq.health';