feat(traffic): scaffold Cloudflare Worker traffic source (foundation)

packages/api-routes/test/db-dto-coverage.test.ts

@@ -332,6 +332,8 @@ const COVERAGE: Record<string, CoverageEntry> = {
     internal: {
       lastEventIds: 'Bounded ring buffer of recent event IDs; internal dedup state, not part of the source DTO.',
       configJson: 'Exposed on the DTO as `config`; the DB column keeps the `Json` suffix for grep-ability.',
+      ingestTokenHash: 'sha256 of the per-source bearer token issued to push-receive adapters (Cloudflare Worker). Auth-only — never returned via the DTO.',
+      lastWorkerVersion: 'Semver reported by the most recent forwarded event from a push-receive Worker. Surfaced through the doctor check, not the source DTO.',
     },
   },
 

packages/contracts/src/traffic.ts

@@ -169,6 +169,91 @@ export const trafficConnectVercelRequestSchema = z.object({
 })
 export type TrafficConnectVercelRequest = z.infer<typeof trafficConnectVercelRequestSchema>
 
+/**
+ * Persisted in `traffic_sources.configJson` for `sourceType = 'cloudflare'`
+ * when the source is a Worker push (the only Cloudflare delivery shape this
+ * release supports). The per-source bearer token + HMAC secret never live
+ * here — they go to `~/.canonry/config.yaml` under
+ * `cloudflareTraffic.connections.<sourceId>`. The DB only carries the
+ * sha256 hash of the bearer for verification.
+ */
+export const cloudflareWorkerSourceConfigSchema = z.object({
+  schemaVersion: z.literal(1),
+  /** Semver of the Worker script bundle that was generated at connect/rotate time. */
+  workerVersion: z.string().min(1),
+  /** Identifier of the bot/referer keyword set baked into the deployed Worker. */
+  expectedBotListVersion: z.string().min(1),
+  /** Operator-supplied Cloudflare zone id for the deployed Worker. Optional in Phase 1. */
+  zoneId: z.string().nullable(),
+  /** Operator-supplied Cloudflare account id. Optional in Phase 1; required for Phase 2 auto-deploy. */
+  accountId: z.string().nullable(),
+})
+export type CloudflareWorkerSourceConfig = z.infer<typeof cloudflareWorkerSourceConfigSchema>
+
+export const trafficConnectCloudflareRequestSchema = z.object({
+  displayName: z.string().min(1).optional(),
+  /** Cloudflare zone id of the deployed Worker (informational; not validated against Cloudflare). */
+  zoneId: z.string().min(1).optional(),
+  /** Cloudflare account id (informational; required when Phase 2 auto-deploy lands). */
+  accountId: z.string().min(1).optional(),
+})
+export type TrafficConnectCloudflareRequest = z.infer<typeof trafficConnectCloudflareRequestSchema>
+
+/**
+ * Returned by `POST /traffic/connect/cloudflare`. The operator deploys the
+ * generated Worker script to their Cloudflare zone; the embedded bearer +
+ * HMAC secret authenticate every subsequent ingest request.
+ */
+export const trafficConnectCloudflareResponseSchema = z.object({
+  sourceId: z.string().min(1),
+  workerScript: z.string().min(1),
+  wranglerToml: z.string().min(1),
+  workerVersion: z.string().min(1),
+  instructions: z.string().min(1),
+})
+export type TrafficConnectCloudflareResponse = z.infer<typeof trafficConnectCloudflareResponseSchema>
+
+/**
+ * One event row inside a `cloudflareWorkerIngestRequest`. Field shape mirrors
+ * what a Cloudflare Worker can pull off a `Request` (`request.url`,
+ * `request.headers`, `request.cf`). Every non-mandatory field is nullable —
+ * `cf.*` properties depend on the customer's plan tier and are absent on
+ * free/Pro plans without Bot Management.
+ */
+export const cloudflareWorkerEventSchema = z.object({
+  /** Cloudflare `cf-ray` request id — globally unique per request. */
+  eventId: z.string().min(1),
+  observedAt: z.string().min(1),
+  method: z.string().nullable(),
+  host: z.string().nullable(),
+  path: z.string().min(1),
+  queryString: z.string().nullable(),
+  status: z.number().int().nullable(),
+  userAgent: z.string().nullable(),
+  remoteIp: z.string().nullable(),
+  referer: z.string().nullable(),
+  cf: z.object({
+    verifiedBot: z.boolean().nullable(),
+    botScore: z.number().int().nullable(),
+    country: z.string().nullable(),
+    asn: z.number().int().nullable(),
+    asOrganization: z.string().nullable(),
+  }).nullable(),
+})
+export type CloudflareWorkerEvent = z.infer<typeof cloudflareWorkerEventSchema>
+
+/**
+ * Body of `POST /api/v1/projects/:name/traffic/cloudflare/ingest`. The
+ * Worker forwards one event per request in this release; the array shape
+ * keeps the door open for a future Logpush sibling adapter that batches.
+ */
+export const cloudflareWorkerIngestRequestSchema = z.object({
+  schemaVersion: z.literal(1),
+  workerVersion: z.string().min(1),
+  events: z.array(cloudflareWorkerEventSchema).min(1).max(100),
+})
+export type CloudflareWorkerIngestRequest = z.infer<typeof cloudflareWorkerIngestRequestSchema>
+
 export const trafficSyncResponseSchema = z.object({
   sourceId: z.string(),
   runId: z.string(),

packages/contracts/test/traffic.test.ts

@@ -3,7 +3,12 @@ import {
   TrafficEventConfidences,
   TrafficEvidenceKinds,
   TrafficSourceTypes,
+  cloudflareWorkerEventSchema,
+  cloudflareWorkerIngestRequestSchema,
+  cloudflareWorkerSourceConfigSchema,
   normalizedTrafficRequestSchema,
+  trafficConnectCloudflareRequestSchema,
+  trafficConnectCloudflareResponseSchema,
   trafficConnectVercelRequestSchema,
   trafficConnectWordpressRequestSchema,
   vercelTrafficSourceConfigSchema,
@@ -169,3 +174,250 @@ describe('trafficConnectVercelRequestSchema', () => {
     })).toThrow()
   })
 })
+
+describe('cloudflareWorkerSourceConfigSchema', () => {
+  it('accepts a valid Cloudflare Worker source config', () => {
+    const parsed = cloudflareWorkerSourceConfigSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      expectedBotListVersion: '2026-05-27',
+      zoneId: 'zone_abc123',
+      accountId: 'acct_xyz789',
+    })
+    expect(parsed.workerVersion).toBe('1.0.0')
+    expect(parsed.zoneId).toBe('zone_abc123')
+  })
+
+  it('allows zoneId and accountId to be null', () => {
+    const parsed = cloudflareWorkerSourceConfigSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      expectedBotListVersion: '2026-05-27',
+      zoneId: null,
+      accountId: null,
+    })
+    expect(parsed.zoneId).toBeNull()
+    expect(parsed.accountId).toBeNull()
+  })
+
+  it('rejects a schemaVersion other than 1', () => {
+    expect(() => cloudflareWorkerSourceConfigSchema.parse({
+      schemaVersion: 2,
+      workerVersion: '1.0.0',
+      expectedBotListVersion: '2026-05-27',
+      zoneId: null,
+      accountId: null,
+    })).toThrow()
+  })
+
+  it('rejects an empty workerVersion or expectedBotListVersion', () => {
+    expect(() => cloudflareWorkerSourceConfigSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '',
+      expectedBotListVersion: '2026-05-27',
+      zoneId: null,
+      accountId: null,
+    })).toThrow()
+    expect(() => cloudflareWorkerSourceConfigSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      expectedBotListVersion: '',
+      zoneId: null,
+      accountId: null,
+    })).toThrow()
+  })
+})
+
+describe('trafficConnectCloudflareRequestSchema', () => {
+  it('accepts an empty body (all fields optional)', () => {
+    const parsed = trafficConnectCloudflareRequestSchema.parse({})
+    expect(parsed.displayName).toBeUndefined()
+    expect(parsed.zoneId).toBeUndefined()
+    expect(parsed.accountId).toBeUndefined()
+  })
+
+  it('accepts a connect request with every optional field', () => {
+    const parsed = trafficConnectCloudflareRequestSchema.parse({
+      displayName: 'Example zone',
+      zoneId: 'zone_abc123',
+      accountId: 'acct_xyz789',
+    })
+    expect(parsed.displayName).toBe('Example zone')
+    expect(parsed.zoneId).toBe('zone_abc123')
+    expect(parsed.accountId).toBe('acct_xyz789')
+  })
+
+  it('rejects an empty string for any provided field', () => {
+    expect(() => trafficConnectCloudflareRequestSchema.parse({ displayName: '' })).toThrow()
+    expect(() => trafficConnectCloudflareRequestSchema.parse({ zoneId: '' })).toThrow()
+    expect(() => trafficConnectCloudflareRequestSchema.parse({ accountId: '' })).toThrow()
+  })
+})
+
+describe('trafficConnectCloudflareResponseSchema', () => {
+  it('accepts a populated response', () => {
+    const parsed = trafficConnectCloudflareResponseSchema.parse({
+      sourceId: 'src_abc123',
+      workerScript: 'addEventListener("fetch", () => {})',
+      wranglerToml: 'name = "canonry-worker"',
+      workerVersion: '1.0.0',
+      instructions: 'Deploy to your zone',
+    })
+    expect(parsed.sourceId).toBe('src_abc123')
+    expect(parsed.workerScript).toContain('fetch')
+  })
+
+  it('rejects empty string for any required field', () => {
+    expect(() => trafficConnectCloudflareResponseSchema.parse({
+      sourceId: '',
+      workerScript: 'x',
+      wranglerToml: 'x',
+      workerVersion: 'x',
+      instructions: 'x',
+    })).toThrow()
+  })
+})
+
+describe('cloudflareWorkerEventSchema', () => {
+  it('accepts a full event with cf properties populated', () => {
+    const parsed = cloudflareWorkerEventSchema.parse({
+      eventId: '8a3d2b0c-cf-ray',
+      observedAt: '2026-05-27T15:30:00.123Z',
+      method: 'GET',
+      host: 'example.com',
+      path: '/blog/post',
+      queryString: 'utm_source=chatgpt',
+      status: 200,
+      userAgent: 'GPTBot/1.2',
+      remoteIp: '20.171.207.34',
+      referer: 'https://chat.openai.com/',
+      cf: {
+        verifiedBot: true,
+        botScore: 30,
+        country: 'US',
+        asn: 8075,
+        asOrganization: 'Microsoft Corporation',
+      },
+    })
+    expect(parsed.eventId).toBe('8a3d2b0c-cf-ray')
+    expect(parsed.cf?.verifiedBot).toBe(true)
+  })
+
+  it('accepts a minimal event with cf=null and most fields null', () => {
+    const parsed = cloudflareWorkerEventSchema.parse({
+      eventId: 'ray-id',
+      observedAt: '2026-05-27T15:30:00.123Z',
+      method: null,
+      host: null,
+      path: '/',
+      queryString: null,
+      status: null,
+      userAgent: null,
+      remoteIp: null,
+      referer: null,
+      cf: null,
+    })
+    expect(parsed.cf).toBeNull()
+    expect(parsed.path).toBe('/')
+  })
+
+  it('rejects an empty path', () => {
+    expect(() => cloudflareWorkerEventSchema.parse({
+      eventId: 'ray-id',
+      observedAt: '2026-05-27T15:30:00.123Z',
+      method: null,
+      host: null,
+      path: '',
+      queryString: null,
+      status: null,
+      userAgent: null,
+      remoteIp: null,
+      referer: null,
+      cf: null,
+    })).toThrow()
+  })
+
+  it('rejects an empty eventId', () => {
+    expect(() => cloudflareWorkerEventSchema.parse({
+      eventId: '',
+      observedAt: '2026-05-27T15:30:00.123Z',
+      method: null,
+      host: null,
+      path: '/',
+      queryString: null,
+      status: null,
+      userAgent: null,
+      remoteIp: null,
+      referer: null,
+      cf: null,
+    })).toThrow()
+  })
+})
+
+describe('cloudflareWorkerIngestRequestSchema', () => {
+  const validEvent = {
+    eventId: 'ray-id',
+    observedAt: '2026-05-27T15:30:00.123Z',
+    method: 'GET',
+    host: 'example.com',
+    path: '/',
+    queryString: null,
+    status: 200,
+    userAgent: 'GPTBot/1.2',
+    remoteIp: '20.171.207.34',
+    referer: null,
+    cf: null,
+  }
+
+  it('accepts a single-event ingest request', () => {
+    const parsed = cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      events: [validEvent],
+    })
+    expect(parsed.events).toHaveLength(1)
+  })
+
+  it('accepts an array of up to 100 events', () => {
+    const events = Array.from({ length: 100 }, (_, i) => ({ ...validEvent, eventId: `ray-${i}` }))
+    const parsed = cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      events,
+    })
+    expect(parsed.events).toHaveLength(100)
+  })
+
+  it('rejects an empty events array', () => {
+    expect(() => cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      events: [],
+    })).toThrow()
+  })
+
+  it('rejects more than 100 events', () => {
+    const events = Array.from({ length: 101 }, (_, i) => ({ ...validEvent, eventId: `ray-${i}` }))
+    expect(() => cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '1.0.0',
+      events,
+    })).toThrow()
+  })
+
+  it('rejects a non-1 schemaVersion', () => {
+    expect(() => cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 2,
+      workerVersion: '1.0.0',
+      events: [validEvent],
+    })).toThrow()
+  })
+
+  it('rejects an empty workerVersion', () => {
+    expect(() => cloudflareWorkerIngestRequestSchema.parse({
+      schemaVersion: 1,
+      workerVersion: '',
+      events: [validEvent],
+    })).toThrow()
+  })
+})

packages/db/src/migrate.ts

@@ -1431,6 +1431,24 @@ export const MIGRATION_VERSIONS: ReadonlyArray<MigrationVersion> = [
       }
     },
   },
+  {
+    version: 67,
+    name: 'traffic-sources-cloudflare-worker-columns',
+    // Push-receive traffic sources (currently only `cloudflare`) need a
+    // per-source bearer for the Worker to authenticate against canonry's
+    // ingest endpoint, plus a place to remember the deployed Worker
+    // version so `cloudflare.worker.version-stale` can compare. The
+    // cleartext bearer + HMAC secret stay in ~/.canonry/config.yaml under
+    // `cloudflareTraffic.connections.<sourceId>`; only the sha256 of the
+    // bearer is stored here.
+    //
+    // Idempotent: `ALTER TABLE ADD COLUMN` errors with "duplicate column
+    // name" on retry, which the runner already swallows.
+    statements: [
+      `ALTER TABLE traffic_sources ADD COLUMN ingest_token_hash TEXT`,
+      `ALTER TABLE traffic_sources ADD COLUMN last_worker_version TEXT`,
+    ],
+  },
 ]
 
 /**