feat: add worker domain events and processing metrics

apps/worker/package.json

@@ -23,6 +23,10 @@
   "dependencies": {
     "@flakemetry/contracts": "workspace:*",
     "@flakemetry/core": "workspace:*",
-    "@flakemetry/db": "workspace:*"
+    "@flakemetry/db": "workspace:*",
+    "@opentelemetry/api": "^1.9.0",
+    "@opentelemetry/exporter-metrics-otlp-http": "^0.220.0",
+    "@opentelemetry/resources": "^2.0.0",
+    "@opentelemetry/sdk-metrics": "^2.0.0"
   }
 }

apps/worker/src/__tests__/events.test.ts

@@ -0,0 +1,66 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import { createEventBus } from '../events'
+
+describe('createEventBus', () => {
+  it('delivers a payload to every subscriber of that event', () => {
+    const bus = createEventBus()
+    const first = vi.fn()
+    const second = vi.fn()
+    bus.on('run.processed', first)
+    bus.on('run.processed', second)
+
+    bus.emit('run.processed', {
+      runId: 'r1',
+      projectId: 'p1',
+      executions: 2,
+      newIdentities: 1,
+      movedIdentities: 0,
+    })
+
+    expect(first).toHaveBeenCalledTimes(1)
+    expect(second).toHaveBeenCalledTimes(1)
+    expect(first.mock.calls[0]?.[0]).toMatchObject({ runId: 'r1', executions: 2 })
+  })
+
+  it('does not deliver to subscribers of other events', () => {
+    const bus = createEventBus()
+    const handler = vi.fn()
+    bus.on('identity.created', handler)
+
+    bus.emit('score.updated', {
+      testIdentityId: 't1',
+      projectId: 'p1',
+      score: 0.4,
+      quarantineCandidate: false,
+    })
+
+    expect(handler).not.toHaveBeenCalled()
+  })
+
+  it('stops delivering after unsubscribe', () => {
+    const bus = createEventBus()
+    const handler = vi.fn()
+    const off = bus.on('identity.moved', handler)
+    off()
+
+    bus.emit('identity.moved', { testIdentityId: 't1', projectId: 'p1', alias: 'a' })
+
+    expect(handler).not.toHaveBeenCalled()
+  })
+
+  it('isolates a throwing subscriber from the rest', () => {
+    const onError = vi.fn()
+    const bus = createEventBus(onError)
+    const healthy = vi.fn()
+    bus.on('identity.created', () => {
+      throw new Error('boom')
+    })
+    bus.on('identity.created', healthy)
+
+    bus.emit('identity.created', { testIdentityId: 't1', projectId: 'p1', fingerprint: 'fp' })
+
+    expect(healthy).toHaveBeenCalledTimes(1)
+    expect(onError).toHaveBeenCalledTimes(1)
+  })
+})

apps/worker/src/__tests__/processor.test.ts

@@ -2,6 +2,7 @@ import type { IngestRunBatch } from '@flakemetry/contracts'
 import { PrismaClient } from '@flakemetry/db'
 import { afterAll, beforeEach, describe, expect, it } from 'vitest'
 
+import { createEventBus, type DomainEventMap } from '../events'
 import { processJob } from '../processor'
 
 const hasDb = Boolean(process.env.DATABASE_URL)
@@ -96,6 +97,63 @@ describe.skipIf(!hasDb)('processJob', () => {
     expect(await prisma.testIdentity.count()).toBe(1)
   })
 
+  it('emits domain events for identities, scores and the processed run', async () => {
+    const events = createEventBus()
+    const created: DomainEventMap['identity.created'][] = []
+    const scored: DomainEventMap['score.updated'][] = []
+    const processed: DomainEventMap['run.processed'][] = []
+    events.on('identity.created', (payload) => created.push(payload))
+    events.on('score.updated', (payload) => scored.push(payload))
+    events.on('run.processed', (payload) => processed.push(payload))
+
+    const ctx = { ...(await seedProject()), now: NOW, events }
+    const result = await processJob(prisma, batch(), ctx)
+
+    expect(created).toHaveLength(1)
+    expect(created[0]?.fingerprint).toBeTruthy()
+    expect(scored).toHaveLength(1)
+    expect(scored[0]?.testIdentityId).toBe(created[0]?.testIdentityId)
+    expect(processed).toEqual([
+      {
+        runId: result.runId,
+        projectId: ctx.projectId,
+        executions: 2,
+        newIdentities: 1,
+        movedIdentities: 0,
+      },
+    ])
+  })
+
+  it('emits identity.moved when a test file moves', async () => {
+    const events = createEventBus()
+    const moved: DomainEventMap['identity.moved'][] = []
+    events.on('identity.moved', (payload) => moved.push(payload))
+
+    const ctx = { ...(await seedProject()), now: NOW, events }
+    await processJob(prisma, batch(), ctx)
+    await processJob(
+      prisma,
+      batch({
+        idempotencyKey: 'run-000002',
+        executions: [
+          {
+            filePath: 'e2e/auth/login.spec.ts',
+            suite: 'auth',
+            title: 'logs in',
+            status: 'pass',
+            attempt: 1,
+            startedAt: new Date('2026-07-16T11:00:00Z'),
+            durationMs: 1600,
+          },
+        ],
+      }),
+      ctx,
+    )
+
+    expect(moved).toHaveLength(1)
+    expect(moved[0]?.alias).toBeTruthy()
+  })
+
   it('stitches history across a file move via L2 identity resolution', async () => {
     const ctx = { ...(await seedProject()), now: NOW }
     await processJob(prisma, batch(), ctx)

apps/worker/src/events.ts

@@ -0,0 +1,60 @@
+export interface DomainEventMap {
+  'run.processed': {
+    runId: string
+    projectId: string
+    executions: number
+    newIdentities: number
+    movedIdentities: number
+  }
+  'identity.created': {
+    testIdentityId: string
+    projectId: string
+    fingerprint: string
+  }
+  'identity.moved': {
+    testIdentityId: string
+    projectId: string
+    alias: string
+  }
+  'score.updated': {
+    testIdentityId: string
+    projectId: string
+    score: number
+    quarantineCandidate: boolean
+  }
+}
+
+export type DomainEventName = keyof DomainEventMap
+
+export type DomainEventHandler<K extends DomainEventName> = (payload: DomainEventMap[K]) => void
+
+export interface EventBus {
+  emit: <K extends DomainEventName>(name: K, payload: DomainEventMap[K]) => void
+  on: <K extends DomainEventName>(name: K, handler: DomainEventHandler<K>) => () => void
+}
+
+export const createEventBus = (
+  onHandlerError: (error: unknown) => void = () => undefined,
+): EventBus => {
+  const handlers = new Map<DomainEventName, Set<DomainEventHandler<DomainEventName>>>()
+
+  return {
+    emit(name, payload) {
+      for (const handler of handlers.get(name) ?? []) {
+        try {
+          ;(handler as DomainEventHandler<typeof name>)(payload)
+        } catch (error) {
+          onHandlerError(error)
+        }
+      }
+    },
+    on(name, handler) {
+      const set = handlers.get(name) ?? new Set()
+      set.add(handler as DomainEventHandler<DomainEventName>)
+      handlers.set(name, set)
+      return () => {
+        set.delete(handler as DomainEventHandler<DomainEventName>)
+      }
+    },
+  }
+}

apps/worker/src/index.ts

@@ -1,16 +1,32 @@
 import { getPrismaClient, IngestionQueue } from '@flakemetry/db'
 
+import { createEventBus } from './events'
 import { createWorker } from './runner'
+import { initSelfTelemetry, observeQueueDepth } from './telemetry'
 
 const prisma = getPrismaClient()
 const queue = new IngestionQueue(prisma)
+
+const selfOtelEndpoint = process.env.FLAKEMETRY_SELF_OTEL_ENDPOINT
+const shutdownTelemetry = selfOtelEndpoint
+  ? initSelfTelemetry({ endpoint: selfOtelEndpoint })
+  : undefined
+
+observeQueueDepth(() => queue.depth())
+
+const events = createEventBus((error) => {
+  process.stderr.write(`worker: event handler failed ${String(error)}\n`)
+})
+
 const worker = createWorker(prisma, queue, {
   pollIntervalMs: Number(process.env.POLL_INTERVAL_MS ?? 1_000),
+  events,
 })
 
 const shutdown = () => {
   process.stdout.write('worker: shutting down\n')
   worker.stop()
+  void shutdownTelemetry?.()
 }
 process.on('SIGINT', shutdown)
 process.on('SIGTERM', shutdown)

apps/worker/src/processor.ts

@@ -9,12 +9,15 @@ import {
 } from '@flakemetry/core'
 import type { Prisma, PrismaClient } from '@flakemetry/db'
 
+import type { EventBus } from './events'
+
 export interface ProcessContext {
   orgId: string
   projectId: string
   now: Date
   threshold?: number
   minSamples?: number
+  events?: EventBus
 }
 
 export interface ProcessResult {
@@ -38,6 +41,8 @@ export const processJob = async (
   const finishedAt = batch.run.finishedAt ?? null
 
   const affected = new Set<string>()
+  const createdEvents: { testIdentityId: string; projectId: string; fingerprint: string }[] = []
+  const movedEvents: { testIdentityId: string; projectId: string; alias: string }[] = []
   let newIdentities = 0
   let movedIdentities = 0
 
@@ -120,6 +125,11 @@ export const processJob = async (
         })
         const entry = existing.find((item) => item.id === identityId)
         if (entry) entry.aliases = [...entry.aliases, resolution.addAlias]
+        movedEvents.push({
+          testIdentityId: identityId,
+          projectId: ctx.projectId,
+          alias: resolution.addAlias,
+        })
       } else {
         newIdentities += 1
         const created = await tx.testIdentity.upsert({
@@ -146,6 +156,7 @@ export const processJob = async (
           paramsHash,
           aliases: [],
         })
+        createdEvents.push({ testIdentityId: identityId, projectId: ctx.projectId, fingerprint })
       }
 
       affected.add(identityId)
@@ -175,10 +186,21 @@ export const processJob = async (
     return run.id
   })
 
+  for (const event of createdEvents) ctx.events?.emit('identity.created', event)
+  for (const event of movedEvents) ctx.events?.emit('identity.moved', event)
+
   for (const identityId of affected) {
     await scoreIdentity(prisma, identityId, ctx)
   }
 
+  ctx.events?.emit('run.processed', {
+    runId,
+    projectId: ctx.projectId,
+    executions: batch.executions.length,
+    newIdentities,
+    movedIdentities,
+  })
+
   return {
     runId,
     executions: batch.executions.length,
@@ -260,4 +282,11 @@ const scoreIdentity = async (
     create: { testIdentityId: identityId, ...data },
     update: data,
   })
+
+  ctx.events?.emit('score.updated', {
+    testIdentityId: identityId,
+    projectId: ctx.projectId,
+    score: result.score,
+    quarantineCandidate: result.quarantineCandidate,
+  })
 }

apps/worker/src/runner.ts

@@ -1,12 +1,15 @@
 import { ingestRunBatchSchema } from '@flakemetry/contracts'
 import type { IngestionQueue, PrismaClient } from '@flakemetry/db'
 
+import type { EventBus } from './events'
 import { processJob } from './processor'
+import { workerMetrics } from './telemetry'
 
 export interface WorkerOptions {
   pollIntervalMs?: number
   batchSize?: number
   now?: () => Date
+  events?: EventBus
 }
 
 export interface Worker {
@@ -30,16 +33,28 @@ export const createWorker = (
   const tick = async (): Promise<number> => {
     const jobs = await queue.dequeue(batchSize)
     for (const job of jobs) {
+      const pickedUpAt = Date.now()
+      workerMetrics.processingLag.record(Math.max(0, pickedUpAt - job.createdAt.getTime()))
       try {
         const batch = ingestRunBatchSchema.parse(job.payload)
-        await processJob(prisma, batch, { orgId: job.orgId, projectId: job.projectId, now: now() })
+        await processJob(prisma, batch, {
+          orgId: job.orgId,
+          projectId: job.projectId,
+          now: now(),
+          events: options.events,
+        })
         await queue.complete(job.id)
+        workerMetrics.jobsProcessed.add(1)
       } catch (error) {
         const outcome = await queue.fail(
           job.id,
           error instanceof Error ? error.message : String(error),
         )
+        workerMetrics.jobsFailed.add(1, { outcome })
+        if (outcome === 'dead') workerMetrics.jobsDeadLettered.add(1)
         process.stderr.write(`worker: job ${job.id} failed (${outcome}): ${String(error)}\n`)
+      } finally {
+        workerMetrics.processingDuration.record(Date.now() - pickedUpAt)
       }
     }
     return jobs.length