feat: Job Queue for Blockchain Operations

__tests__/jobs/service.test.ts

@@ -0,0 +1,98 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const { query } = vi.hoisted(() => ({ query: vi.fn() }))
+vi.mock('@/lib/db', () => ({ sql: { query } }))
+
+import { JobQueueService } from '@/lib/jobs/service'
+
+const row = (overrides: Record<string, unknown> = {}) => ({
+  id: '33333333-3333-4333-8333-333333333333',
+  job_type: 'deadline_check',
+  dedupe_key: 'deadline:2026-09-28T09',
+  status: 'queued',
+  payload: {},
+  result: null,
+  retry_count: 0,
+  max_retries: 3,
+  next_run_at: '2026-09-28T09:00:00Z',
+  completed_at: null,
+  created_at: '2026-09-28T08:00:00Z',
+  updated_at: '2026-09-28T08:00:00Z',
+  ...overrides,
+})
+
+describe('JobQueueService', () => {
+  beforeEach(() => query.mockReset())
+
+  it('queues a job once and returns it without running the handler', async () => {
+    query.mockResolvedValueOnce([row()])
+    const service = new JobQueueService()
+    const result = await service.enqueue({ type: 'deadline_check', dedupeKey: 'deadline:2026-09-28T09' })
+    expect(result.isDuplicate).toBe(false)
+    expect(result.job.state).toBe('pending')
+    expect(query.mock.calls[0][0]).toContain('ON CONFLICT (dedupe_key) DO NOTHING')
+  })
+
+  it('returns the existing job when the same dedupe key is queued again', async () => {
+    query.mockResolvedValueOnce([]).mockResolvedValueOnce([row({ status: 'completed' })])
+    const service = new JobQueueService()
+    const result = await service.enqueue({ type: 'deadline_check', dedupeKey: 'deadline:2026-09-28T09' })
+    expect(result.isDuplicate).toBe(true)
+    expect(result.job.status).toBe('completed')
+    expect(query).toHaveBeenCalledTimes(2)
+  })
+
+  it('rejects a dedupe key already bound to another job type', async () => {
+    query.mockResolvedValueOnce([]).mockResolvedValueOnce([row()])
+    const service = new JobQueueService()
+    await expect(service.enqueue({ type: 'tx_monitor', dedupeKey: 'deadline:2026-09-28T09' }))
+      .rejects.toMatchObject({ code: 'JOB_DEDUPE_CONFLICT', httpStatus: 409 })
+  })
+
+  it('claims due jobs with a row lock so another worker cannot take them', async () => {
+    query.mockResolvedValueOnce([row({ status: 'processing' })])
+    query.mockResolvedValueOnce([])
+    const service = new JobQueueService()
+    const result = await service.processBatch({
+      workerId: 'worker-a',
+      handlers: { deadline_check: async () => ({ remindersSent: 1, overdueFlagged: 0 }) },
+    })
+    expect(query.mock.calls[0][0]).toContain('FOR UPDATE SKIP LOCKED')
+    expect(query.mock.calls[1][0]).toContain("status = 'completed'")
+    expect(result).toMatchObject({ claimed: 1, completed: 1, failed: 0 })
+  })
+
+  it('records the stack and requeues with backoff while retries remain', async () => {
+    query.mockResolvedValueOnce([row()]).mockResolvedValueOnce([{ status: 'queued' }])
+    const service = new JobQueueService()
+    const result = await service.processBatch({
+      workerId: 'worker-a',
+      handlers: { deadline_check: async () => { throw new Error('horizon timeout') } },
+    })
+    const failureSql = query.mock.calls[1][0] as string
+    expect(failureSql).toContain('INSERT INTO background_job_errors')
+    expect(failureSql).toContain('INTERVAL \'1 millisecond\'')
+    expect(query.mock.calls[1][1][3]).toContain('horizon timeout')
+    expect(result).toMatchObject({ retried: 1, failed: 0 })
+  })
+
+  it('marks the job failed once the retry limit is exhausted', async () => {
+    query.mockResolvedValueOnce([row({ retry_count: 2 })]).mockResolvedValueOnce([{ status: 'failed' }])
+    const service = new JobQueueService()
+    const result = await service.processBatch({
+      workerId: 'worker-a',
+      handlers: { deadline_check: async () => { throw new Error('still down') } },
+    })
+    expect(result).toMatchObject({ retried: 0, failed: 1 })
+  })
+
+  it('returns job state and stored errors', async () => {
+    query.mockResolvedValueOnce([row({
+      status: 'failed',
+      errors: [{ attempt: 3, message: 'still down', stack: 'Error: still down', metadata: { workerId: 'worker-a' }, createdAt: '2026-09-28T09:05:00Z' }],
+    })])
+    const job = await new JobQueueService().getJob(row().id)
+    expect(job.state).toBe('failed')
+    expect(job.errors[0]).toMatchObject({ attempt: 3, message: 'still down', stack: 'Error: still down' })
+  })
+})

app/api/jobs/[id]/route.ts

@@ -0,0 +1,26 @@
+export const dynamic = 'force-dynamic'
+
+import { NextRequest, NextResponse } from 'next/server'
+import { withAuthCtx } from '@/lib/auth/middleware'
+import { JobQueueError, jobQueueService } from '@/lib/jobs'
+
+export const GET = withAuthCtx(async (
+  _request: NextRequest,
+  _auth,
+  context: { params: Promise<{ id: string }> },
+) => {
+  const { id } = await context.params
+  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id)) {
+    return NextResponse.json({ error: 'Job id must be a UUID.', code: 'INVALID_JOB_ID' }, { status: 422 })
+  }
+
+  try {
+    return NextResponse.json({ data: await jobQueueService.getJob(id) })
+  } catch (error) {
+    if (error instanceof JobQueueError) {
+      return NextResponse.json({ error: error.message, code: error.code }, { status: error.httpStatus })
+    }
+    console.error('[api/jobs/[id]] Error:', error)
+    return NextResponse.json({ error: 'Failed to load job', code: 'JOB_QUERY_FAILED' }, { status: 500 })
+  }
+})

app/api/jobs/process/route.ts

@@ -0,0 +1,33 @@
+export const dynamic = 'force-dynamic'
+
+import { timingSafeEqual } from 'crypto'
+import { NextRequest, NextResponse } from 'next/server'
+import { createJobHandlers, jobQueueService } from '@/lib/jobs'
+
+function authorized(request: NextRequest): boolean {
+  const expected = process.env.JOB_WORKER_SECRET
+  const provided = request.headers.get('authorization')?.replace(/^Bearer\s+/i, '')
+  if (!expected || !provided) return false
+  const a = Buffer.from(expected)
+  const b = Buffer.from(provided)
+  return a.length === b.length && timingSafeEqual(a, b)
+}
+
+export async function POST(request: NextRequest) {
+  if (!authorized(request)) {
+    return NextResponse.json({ error: 'Unauthorized', code: 'AUTH_REQUIRED' }, { status: 401 })
+  }
+  try {
+    const body = await request.json().catch(() => ({}))
+    const batchSize = Number.isInteger(body.batchSize) ? body.batchSize : 10
+    const data = await jobQueueService.processBatch({
+      workerId: typeof body.workerId === 'string' && body.workerId.trim() ? body.workerId.trim() : 'api-worker',
+      batchSize,
+      handlers: createJobHandlers(),
+    })
+    return NextResponse.json({ data })
+  } catch (error) {
+    console.error('[api/jobs/process] Error:', error)
+    return NextResponse.json({ error: 'Job processing failed', code: 'JOB_PROCESS_FAILED' }, { status: 500 })
+  }
+}

lib/contract-sync/service.ts

@@ -85,6 +85,11 @@ export class ContractSyncService {
     return this.listener
   }
 
+  /** Applies one Soroban event through the existing sync path. */
+  async applyEvent(payload: SorobanEventPayload): Promise<void> {
+    await this.processSync({ id: buildSyncDedupeKey(payload), payload })
+  }
+
   private async onEvent(payload: SorobanEventPayload): Promise<void> {
     const dedupeKey = buildSyncDedupeKey(payload)
 

lib/jobs/handlers.ts

@@ -0,0 +1,65 @@
+import type { JobHandler, JobType } from './types'
+import type { SorobanEventPayload } from '@/lib/contract-sync/types'
+
+export function createJobHandlers(): Record<JobType, JobHandler> {
+  return {
+    tx_monitor: async (job) => {
+      const { txRecoveryService } = await import('@/lib/tx-recovery')
+      const requested = job.payload.batchSize
+      const batchSize = typeof requested === 'number' ? requested : 25
+      const result = await txRecoveryService.pollPendingQueue(batchSize)
+      return { ...result }
+    },
+    notification: async (job) => {
+      const { createNotification } = await import('@/lib/notifications')
+      const { userId, title, message, type, eventType, payload } = job.payload
+      if (typeof userId !== 'number' && typeof userId !== 'string') {
+        throw new Error('notification job requires a userId')
+      }
+      if (typeof title !== 'string' || typeof message !== 'string' || typeof type !== 'string' || typeof eventType !== 'string') {
+        throw new Error('notification job requires title, message, type, and eventType')
+      }
+      const notification = await createNotification({
+        userId: userId as number,
+        title,
+        message,
+        type,
+        eventType,
+        payload: payload && typeof payload === 'object' && !Array.isArray(payload)
+          ? payload as Record<string, unknown>
+          : {},
+      })
+      return { notificationId: notification.id }
+    },
+    contract_sync: async (job) => {
+      const { ContractSyncService } = await import('@/lib/contract-sync')
+      await new ContractSyncService().applyEvent(readSyncPayload(job.payload))
+      return { synced: true }
+    },
+    deadline_check: async () => {
+      const { DeadlineMonitorService } = await import('@/lib/deadline-monitor')
+      const result = await new DeadlineMonitorService().runCheck()
+      return { ...result }
+    },
+  }
+}
+
+function readSyncPayload(payload: Record<string, unknown>): SorobanEventPayload {
+  const { event, contractAddress, txHash, data } = payload
+  if (typeof event !== 'string' || typeof contractAddress !== 'string' || typeof txHash !== 'string' || !Array.isArray(data)) {
+    throw new Error('contract_sync job requires event, contractAddress, txHash, and data')
+  }
+  return {
+    event: event as SorobanEventPayload['event'],
+    contractAddress,
+    txHash,
+    data,
+    ledgerSequence: typeof payload.ledgerSequence === 'number' ? payload.ledgerSequence : 0,
+    timestamp: typeof payload.timestamp === 'number' ? payload.timestamp : Date.now(),
+    milestoneId: typeof payload.milestoneId === 'number' ? payload.milestoneId : undefined,
+    disputeId: typeof payload.disputeId === 'number' ? payload.disputeId : undefined,
+    amount: typeof payload.amount === 'string' ? payload.amount : undefined,
+    actor: typeof payload.actor === 'string' ? payload.actor : undefined,
+    recipient: typeof payload.recipient === 'string' ? payload.recipient : undefined,
+  }
+}

lib/jobs/index.ts

@@ -0,0 +1,13 @@
+export { JobQueueService, jobQueueService } from './service'
+export { createJobHandlers } from './handlers'
+export { JOB_TYPES, JOB_STATE, JobQueueError, getJobBackoffMs } from './types'
+export type {
+  BackgroundJob,
+  EnqueueJobInput,
+  JobBatchResult,
+  JobErrorRecord,
+  JobHandler,
+  JobState,
+  JobStatus,
+  JobType,
+} from './types'