test: add identity property tests and full-chain otlp e2e

apps/api/package.json

@@ -10,6 +10,10 @@
     ".": {
       "types": "./dist/index.d.ts",
       "default": "./dist/index.js"
+    },
+    "./app": {
+      "types": "./dist/app.d.ts",
+      "default": "./dist/app.js"
     }
   },
   "scripts": {

apps/api/tsup.config.ts

@@ -1,7 +1,7 @@
 import { defineConfig } from 'tsup'
 
 export default defineConfig({
-  entry: ['src/index.ts'],
+  entry: ['src/index.ts', 'src/app.ts'],
   format: ['esm'],
   target: 'es2022',
   outDir: 'dist',

apps/worker/package.json

@@ -13,7 +13,9 @@
     "test": "vitest run"
   },
   "devDependencies": {
+    "@flakemetry/api": "workspace:*",
     "@flakemetry/eslint-config": "workspace:*",
+    "@flakemetry/sdk": "workspace:*",
     "@flakemetry/tsconfig": "workspace:*",
     "@flakemetry/tsup-config": "workspace:*",
     "@types/node": "^22.10.5",

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

@@ -0,0 +1,155 @@
+import { buildApp } from '@flakemetry/api/app'
+import { generateToken, hashToken, IngestionQueue, PrismaClient } from '@flakemetry/db'
+import { exportRunOverOtlp, TestRunRecorder } from '@flakemetry/sdk'
+import { afterAll, afterEach, beforeEach, describe, expect, it } from 'vitest'
+
+import { createEventBus, type DomainEventMap } from '../events'
+import { createWorker, type Worker } from '../runner'
+
+const hasDb = Boolean(process.env.DATABASE_URL)
+const prisma = new PrismaClient()
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
+
+const drain = async (worker: Worker): Promise<number> => {
+  let processed = 0
+  for (let attempt = 0; attempt < 20; attempt += 1) {
+    processed += await worker.tick()
+    const pending = await prisma.ingestionJob.count({ where: { status: { in: ['pending'] } } })
+    if (pending === 0) return processed
+    await sleep(25)
+  }
+  return processed
+}
+
+const recordRun = (commitSha: string, failFirstAttempt: boolean) => {
+  const recorder = new TestRunRecorder({
+    project: 'acme/web',
+    commitSha,
+    branch: 'main',
+    ciProvider: 'github_actions',
+    trigger: 'push',
+    ciRunId: `run-${commitSha}`,
+  })
+  recorder.startRun(new Date('2026-07-16T10:00:00Z'))
+  recorder.record({
+    filePath: 'e2e/login.spec.ts',
+    suite: 'auth',
+    title: 'logs in',
+    status: failFirstAttempt ? 'fail' : 'pass',
+    attempt: 1,
+    startedAt: new Date('2026-07-16T10:00:01Z'),
+    durationMs: 1800,
+    ...(failFirstAttempt
+      ? { error: { type: 'TimeoutError', message: 'Timeout 30000ms exceeded', stack: 'at login' } }
+      : {}),
+  })
+  if (failFirstAttempt) {
+    recorder.record({
+      filePath: 'e2e/login.spec.ts',
+      suite: 'auth',
+      title: 'logs in',
+      status: 'flaky',
+      attempt: 2,
+      retryOfIndex: 0,
+      startedAt: new Date('2026-07-16T10:00:03Z'),
+      durationMs: 1400,
+    })
+  }
+  recorder.finishRun(failFirstAttempt ? 'failed' : 'passed', new Date('2026-07-16T10:00:05Z'))
+  return recorder
+}
+
+describe.skipIf(!hasDb)('full ingestion chain', () => {
+  let app: ReturnType<typeof buildApp>
+  let endpoint: string
+  let token: string
+
+  beforeEach(async () => {
+    await prisma.flakyScore.deleteMany()
+    await prisma.testExecution.deleteMany()
+    await prisma.testIdentity.deleteMany()
+    await prisma.run.deleteMany()
+    await prisma.ingestionJob.deleteMany()
+    await prisma.ingestToken.deleteMany()
+    await prisma.project.deleteMany()
+    await prisma.org.deleteMany()
+
+    const org = await prisma.org.create({ data: { name: 'Acme', slug: `acme-${Date.now()}` } })
+    const project = await prisma.project.create({
+      data: { orgId: org.id, name: 'Web', slug: 'web' },
+    })
+    token = generateToken()
+    await prisma.ingestToken.create({
+      data: { orgId: org.id, projectId: project.id, name: 'ci', tokenHash: hashToken(token) },
+    })
+
+    app = buildApp({ prisma })
+    endpoint = await app.listen({ port: 0, host: '127.0.0.1' })
+  })
+
+  afterEach(async () => {
+    await app.close()
+  })
+
+  afterAll(async () => {
+    await prisma.$disconnect()
+  })
+
+  it('carries a real OTLP export through queue and worker into scored rows', async () => {
+    const events = createEventBus()
+    const processedEvents: DomainEventMap['run.processed'][] = []
+    events.on('run.processed', (payload) => processedEvents.push(payload))
+
+    const worker = createWorker(prisma, new IngestionQueue(prisma), { events })
+
+    await exportRunOverOtlp(recordRun('aaa1111', true), 'e2e-run-000001', { endpoint, token })
+    expect(await prisma.ingestionJob.count()).toBe(1)
+
+    expect(await drain(worker)).toBe(1)
+
+    const run = await prisma.run.findFirstOrThrow()
+    expect(run.commitSha).toBe('aaa1111')
+    expect(run.branch).toBe('main')
+    expect(run.status).toBe('failed')
+
+    const executions = await prisma.testExecution.findMany({ orderBy: { attempt: 'asc' } })
+    expect(executions).toHaveLength(2)
+    expect(executions[0]?.status).toBe('fail')
+    expect(executions[0]?.errorMessage).toBe('Timeout 30000ms exceeded')
+    expect(executions[1]?.status).toBe('flaky')
+    expect(executions[1]?.retryOf).toBe(executions[0]?.id)
+
+    const identity = await prisma.testIdentity.findFirstOrThrow()
+    expect(identity.filePath).toBe('e2e/login.spec.ts')
+    expect(identity.title).toBe('logs in')
+
+    const score = await prisma.flakyScore.findFirstOrThrow()
+    expect(score.testIdentityId).toBe(identity.id)
+    expect(score.passOnRerunRate).toBeGreaterThan(0)
+    expect((score.reasonCodes as { code: string }[]).length).toBeGreaterThan(0)
+
+    expect(processedEvents).toHaveLength(1)
+    expect(processedEvents[0]?.executions).toBe(2)
+  })
+
+  it('accumulates history across runs and stays idempotent on re-delivery', async () => {
+    const worker = createWorker(prisma, new IngestionQueue(prisma))
+
+    await exportRunOverOtlp(recordRun('aaa1111', true), 'e2e-run-000001', { endpoint, token })
+    await drain(worker)
+    await exportRunOverOtlp(recordRun('bbb2222', false), 'e2e-run-000002', { endpoint, token })
+    await drain(worker)
+
+    expect(await prisma.run.count()).toBe(2)
+    expect(await prisma.testIdentity.count()).toBe(1)
+    expect(await prisma.testExecution.count()).toBe(3)
+
+    await exportRunOverOtlp(recordRun('bbb2222', false), 'e2e-run-000002', { endpoint, token })
+    expect(await prisma.ingestionJob.count()).toBe(2)
+    await drain(worker)
+
+    expect(await prisma.run.count()).toBe(2)
+    expect(await prisma.testExecution.count()).toBe(3)
+  })
+})

packages/core/package.json

@@ -40,6 +40,7 @@
     "@flakemetry/tsconfig": "workspace:*",
     "@flakemetry/tsup-config": "workspace:*",
     "@types/node": "^22.10.5",
+    "fast-check": "^3.23.2",
     "vitest": "^3.2.4"
   },
   "dependencies": {

packages/core/src/__tests__/identity.property.test.ts

@@ -0,0 +1,118 @@
+import fc from 'fast-check'
+import { describe, expect, it } from 'vitest'
+
+import { computeFingerprint, hashParams, normalizeFilePath } from '../fingerprint'
+import { type ExistingIdentity, resolveIdentity } from '../identity'
+
+const nonEmpty = fc.string({ minLength: 1, maxLength: 40 }).filter((s) => s.trim().length > 0)
+const filePath = fc
+  .array(nonEmpty, { minLength: 1, maxLength: 4 })
+  .map((segments) => `${segments.join('/')}.spec.ts`)
+
+const asExisting = (
+  candidate: { fingerprint: string; suite: string; title: string; paramsHash: string | null },
+  id = 'identity-1',
+): ExistingIdentity => ({ ...candidate, id, aliases: [] })
+
+const candidateFor = (path: string, suite: string, title: string, paramsHash: string | null) => ({
+  fingerprint: computeFingerprint({ filePath: path, suite, title, paramsHash }),
+  suite,
+  title,
+  paramsHash,
+})
+
+describe('identity properties', () => {
+  it('resolves a moved file to the same identity for any path change', () => {
+    fc.assert(
+      fc.property(filePath, filePath, nonEmpty, nonEmpty, (from, to, suite, title) => {
+        fc.pre(normalizeFilePath(from) !== normalizeFilePath(to))
+
+        const original = candidateFor(from, suite, title, null)
+        const afterMove = candidateFor(to, suite, title, null)
+        const resolution = resolveIdentity(afterMove, [asExisting(original)])
+
+        expect(resolution.kind).toBe('moved')
+        if (resolution.kind === 'moved') expect(resolution.identityId).toBe('identity-1')
+      }),
+    )
+  })
+
+  it('treats a different title as a different identity even at the same path', () => {
+    fc.assert(
+      fc.property(filePath, nonEmpty, nonEmpty, nonEmpty, (path, suite, titleA, titleB) => {
+        fc.pre(titleA.trim() !== titleB.trim())
+
+        const original = candidateFor(path, suite, titleA, null)
+        const renamed = candidateFor(path, suite, titleB, null)
+
+        expect(renamed.fingerprint).not.toBe(original.fingerprint)
+        expect(resolveIdentity(renamed, [asExisting(original)]).kind).toBe('new')
+      }),
+    )
+  })
+
+  it('resolves exactly when nothing changed', () => {
+    fc.assert(
+      fc.property(filePath, nonEmpty, nonEmpty, (path, suite, title) => {
+        const candidate = candidateFor(path, suite, title, null)
+        expect(resolveIdentity(candidate, [asExisting(candidate)]).kind).toBe('exact')
+      }),
+    )
+  })
+
+  it('resolves through an alias recorded by an earlier move', () => {
+    fc.assert(
+      fc.property(filePath, filePath, nonEmpty, nonEmpty, (from, to, suite, title) => {
+        fc.pre(normalizeFilePath(from) !== normalizeFilePath(to))
+
+        const original = candidateFor(from, suite, title, null)
+        const afterMove = candidateFor(to, suite, title, null)
+        const stitched: ExistingIdentity = {
+          ...asExisting(original),
+          aliases: [afterMove.fingerprint],
+        }
+
+        expect(resolveIdentity(afterMove, [stitched]).kind).toBe('exact')
+      }),
+    )
+  })
+
+  it('normalizes paths idempotently', () => {
+    fc.assert(
+      fc.property(filePath, (path) => {
+        const once = normalizeFilePath(path)
+        expect(normalizeFilePath(once)).toBe(once)
+      }),
+    )
+  })
+
+  it('hashes params independently of key order', () => {
+    fc.assert(
+      fc.property(fc.dictionary(nonEmpty, fc.integer(), { maxKeys: 6 }), (params) => {
+        const reversed = Object.fromEntries(Object.entries(params).reverse())
+        expect(hashParams(params)).toBe(hashParams(reversed))
+      }),
+    )
+  })
+
+  it('keeps parameterized variants distinct', () => {
+    fc.assert(
+      fc.property(
+        filePath,
+        nonEmpty,
+        nonEmpty,
+        fc.integer(),
+        fc.integer(),
+        (path, suite, title, a, b) => {
+          fc.pre(a !== b)
+
+          const first = candidateFor(path, suite, title, hashParams({ case: a }))
+          const second = candidateFor(path, suite, title, hashParams({ case: b }))
+
+          expect(second.fingerprint).not.toBe(first.fingerprint)
+          expect(resolveIdentity(second, [asExisting(first)]).kind).toBe('new')
+        },
+      ),
+    )
+  })
+})