feat: add anonymous opt-out telemetry

.changeset/anonymous-telemetry.md

@@ -0,0 +1,6 @@
+---
+'@pgflow/core': patch
+'@pgflow/edge-worker': patch
+---
+
+Add anonymous, opt-out telemetry. A daily `pg_cron` job aggregates yesterday's `pgflow.*` activity into coarse, identifier-free buckets (versions in use, run and worker counts, flow shapes, feature adoption, durations) and sends one small payload through `pg_net` to a Cloudflare Worker backed by Workers Analytics Engine. Workers stamp their package version at registration. Nothing identifying is ever sent; every payload is stored locally in `pgflow_telemetry.sent_reports` for audit; `pgflow_telemetry.preview()` shows any day's payload without sending; `pgflow_telemetry.disable()` opts out permanently (the cron job row is the switch) and local development databases never report.

.changeset/config.json

@@ -15,5 +15,5 @@
   "access": "public",
   "baseBranch": "main",
   "updateInternalDependencies": "patch",
-  "ignore": ["@pgflow/demo", "@pgflow/website", "@pgflow/example-flows"]
+  "ignore": ["@pgflow/demo", "@pgflow/website", "@pgflow/example-flows", "@pgflow/telemetry-worker"]
 }

apps/telemetry-worker/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "@pgflow/telemetry-worker",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "typecheck": "tsc --noEmit",
+    "test": "vitest run",
+    "deploy": "wrangler deploy"
+  },
+  "devDependencies": {
+    "@cloudflare/workers-types": "^4.20250901.0",
+    "typescript": "^5.5.0",
+    "vitest": "^3.0.0",
+    "wrangler": "^4.0.0"
+  }
+}

apps/telemetry-worker/src/index.test.ts

@@ -0,0 +1,125 @@
+import { describe, expect, it } from 'vitest';
+import worker, { type Env } from './index';
+
+function makeEnv() {
+  const points: Array<{ indexes: string[]; blobs: string[]; doubles: number[] }> = [];
+  const env = {
+    PGFLOW_TELEMETRY: {
+      writeDataPoint: (p: { indexes: string[]; blobs: string[]; doubles: number[] }) =>
+        points.push(p),
+    },
+  } as unknown as Env;
+  return { env, points };
+}
+
+function post(body: unknown, env: Env, raw = false): Promise<Response> {
+  return worker.fetch(
+    new Request('https://pgflow-telemetry.workers.dev/', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: raw ? (body as string) : JSON.stringify(body),
+    }),
+    env,
+  );
+}
+
+const valid = {
+  schema: 1,
+  contributions: [
+    { metric: 'runs_started_day', bucket: '2-3' },
+    { metric: 'workers_by_version', bucket: '0.17.0', count: '1' },
+    { metric: 'uses_map', bucket: 'yes' },
+  ],
+};
+
+describe('telemetry ingest', () => {
+  it('accepts a valid payload, writes one point per contribution, returns 204', async () => {
+    const { env, points } = makeEnv();
+    const res = await post(valid, env);
+    expect(res.status).toBe(204);
+    expect(points).toHaveLength(3);
+    expect(points[0]).toEqual({
+      indexes: ['runs_started_day'],
+      blobs: ['2-3'],
+      doubles: [1],
+    });
+  });
+
+  it('rejects GET with 405', async () => {
+    const { env } = makeEnv();
+    const res = await worker.fetch(
+      new Request('https://pgflow-telemetry.workers.dev/', { method: 'GET' }),
+      env,
+    );
+    expect(res.status).toBe(405);
+  });
+
+  it('rejects malformed JSON with 400', async () => {
+    const { env } = makeEnv();
+    const res = await post('{not json', env, true);
+    expect(res.status).toBe(400);
+  });
+
+  it('rejects wrong schema version with 400', async () => {
+    const { env, points } = makeEnv();
+    const res = await post({ schema: 2, contributions: valid.contributions }, env);
+    expect(res.status).toBe(400);
+    expect(points).toHaveLength(0);
+  });
+
+  it('rejects unknown metric with 400', async () => {
+    const { env } = makeEnv();
+    const res = await post({
+      schema: 1,
+      contributions: [{ metric: 'flow_slug', bucket: 'yes' }],
+    }, env);
+    expect(res.status).toBe(400);
+  });
+
+  it('rejects arbitrary bucket strings with 400', async () => {
+    const { env } = makeEnv();
+    const res = await post({
+      schema: 1,
+      contributions: [{ metric: 'runs_started_day', bucket: 'banana' }],
+    }, env);
+    expect(res.status).toBe(400);
+  });
+
+  it('rejects non-semantic-version buckets with 400', async () => {
+    const { env } = makeEnv();
+    const res = await post({
+      schema: 1,
+      contributions: [{ metric: 'workers_by_version', bucket: 'my-flow-name' }],
+    }, env);
+    expect(res.status).toBe(400);
+  });
+
+  it('rejects oversized bodies with 413', async () => {
+    const { env } = makeEnv();
+    const big = 'x'.repeat(2049);
+    const res = await post(big, env, true);
+    expect(res.status).toBe(413);
+  });
+
+  it('rejects more than 64 contributions', async () => {
+    const { env, points } = makeEnv();
+    const res = await post({
+      schema: 1,
+      contributions: Array.from({ length: 65 }, () => ({
+        metric: 'uses_map',
+        bucket: 'yes',
+      })),
+    }, env);
+    // 65 valid contributions always exceed the 2 KB body cap, so the size
+    // guard (413) fires before the count guard. MAX_CONTRIBUTIONS stays as
+    // defense-in-depth below the byte cap.
+    expect(res.status).toBe(413);
+    expect(points).toHaveLength(0);
+  });
+
+  it('sets no cookies on success', async () => {
+    const { env } = makeEnv();
+    const res = await post(valid, env);
+    expect(res.headers.getSetCookie()).toHaveLength(0);
+  });
+});

apps/telemetry-worker/src/index.ts

@@ -0,0 +1,133 @@
+// pgflow anonymous telemetry ingest. Accepts exactly one closed schema,
+// rejects everything else, writes one Analytics Engine data point per
+// contribution. No cookies, no body logging, IP only as transport.
+
+export interface Env {
+  PGFLOW_TELEMETRY: AnalyticsEngineDataset;
+}
+
+const SCHEMA_VERSION = 1;
+const MAX_BODY_BYTES = 2048;
+const MAX_CONTRIBUTIONS = 64;
+
+const COUNT_BUCKETS = new Set([
+  '0', '1', '2-3', '4-7', '8-15', '16-31', '32-63', '64-127', '128-255', '256+',
+]);
+const STEPS_BUCKETS = new Set(['1', '2-3', '4-7', '8-15', '16-31', '32+']);
+const DURATION_BUCKETS = new Set([
+  '<100ms', '100-999ms', '1-9.9s', '10-59s', '1-4.9m', '5-29m',
+  '30m-1.9h', '2-23h', '1-6d', '7d+',
+]);
+const OUTCOME_BUCKETS = new Set(['completed', 'failed', 'started']);
+const MODE_BUCKETS = new Set(['http', 'process']);
+const QUEUE_MODE_BUCKETS = new Set(['flow', 'step']);
+const SEMVER_RE = /^\d+\.\d+\.\d+(-[0-9A-Za-z.-]+)?$/;
+
+type BucketKind =
+  | 'count' | 'steps' | 'duration' | 'outcome' | 'mode' | 'queue_mode'
+  | 'yes' | 'semver';
+
+const METRICS: Record<string, BucketKind> = {
+  active_db_day: 'yes',
+  workers_by_version: 'semver',
+  version_changed: 'yes',
+  worker_functions: 'count',
+  worker_functions_by_mode: 'mode',
+  worker_functions_disabled: 'count',
+  workers_started_day: 'count',
+  workers_stopped_day: 'count',
+  workers_deprecated_day: 'count',
+  worker_starts_per_function: 'count',
+  steps_per_flow: 'steps',
+  uses_map: 'yes',
+  uses_root_map: 'yes',
+  uses_condition: 'yes',
+  uses_graceful_failure: 'yes',
+  uses_skip_cascade: 'yes',
+  uses_retry_override: 'yes',
+  uses_timeout_override: 'yes',
+  uses_start_delay: 'yes',
+  queue_mode: 'queue_mode',
+  runs_started_day: 'count',
+  run_outcomes: 'outcome',
+  run_wall_duration: 'duration',
+  step_wall_duration: 'duration',
+  task_attempts_day: 'count',
+  map_task_count: 'steps',
+};
+
+function validBucket(kind: BucketKind, bucket: unknown): boolean {
+  if (typeof bucket !== 'string' || bucket.length > 32) return false;
+  switch (kind) {
+    case 'count': return COUNT_BUCKETS.has(bucket);
+    case 'steps': return STEPS_BUCKETS.has(bucket);
+    case 'duration': return DURATION_BUCKETS.has(bucket);
+    case 'outcome': return OUTCOME_BUCKETS.has(bucket);
+    case 'mode': return MODE_BUCKETS.has(bucket);
+    case 'queue_mode': return QUEUE_MODE_BUCKETS.has(bucket);
+    case 'yes': return bucket === 'yes';
+    case 'semver': return SEMVER_RE.test(bucket);
+  }
+}
+
+interface Contribution {
+  metric?: unknown;
+  bucket?: unknown;
+  count?: unknown;
+}
+
+const reject = (status: number) => new Response(null, {
+  status,
+  headers: { 'cache-control': 'no-store' },
+});
+
+export default {
+  async fetch(request: Request, env: Env): Promise<Response> {
+    if (request.method !== 'POST') return reject(405);
+
+    const contentType = request.headers.get('content-type') ?? '';
+    if (!contentType.includes('application/json')) return reject(415);
+
+    const raw = await request.text();
+    if (raw.length > MAX_BODY_BYTES) return reject(413);
+
+    let body: { schema?: unknown; contributions?: unknown };
+    try {
+      body = JSON.parse(raw);
+    } catch {
+      return reject(400);
+    }
+
+    if (body.schema !== SCHEMA_VERSION) return reject(400);
+    if (!Array.isArray(body.contributions)) return reject(400);
+    const contributions = body.contributions as Contribution[];
+    if (contributions.length === 0 || contributions.length > MAX_CONTRIBUTIONS) {
+      return reject(400);
+    }
+
+    for (const c of contributions) {
+      if (c === null || typeof c !== 'object') return reject(400);
+      const kind = METRICS[c.metric as string];
+      if (!kind) return reject(400);
+      if (!validBucket(kind, c.bucket)) return reject(400);
+      if (c.count !== undefined) {
+        if (typeof c.count !== 'string' || !COUNT_BUCKETS.has(c.count)) {
+          return reject(400);
+        }
+      }
+    }
+
+    for (const c of contributions) {
+      env.PGFLOW_TELEMETRY.writeDataPoint({
+        indexes: [c.metric as string],
+        blobs: [c.bucket as string],
+        doubles: [1],
+      });
+    }
+
+    return new Response(null, {
+      status: 204,
+      headers: { 'cache-control': 'no-store' },
+    });
+  },
+};

apps/telemetry-worker/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "lib": ["ES2022"],
+    "module": "ES2022",
+    "moduleResolution": "bundler",
+    "strict": true,
+    "noEmit": true,
+    "types": ["@cloudflare/workers-types"]
+  },
+  "include": ["src/index.ts"]
+}

apps/telemetry-worker/vite.config.ts

@@ -0,0 +1,7 @@
+import { defineConfig } from 'vitest/config';
+
+export default defineConfig({
+  test: {
+    include: ['src/**/*.test.ts'],
+  },
+});

apps/telemetry-worker/wrangler.toml

@@ -0,0 +1,11 @@
+name = "pgflow-telemetry"
+main = "src/index.ts"
+compatibility_date = "2026-09-01"
+
+[[analytics_engine_datasets]]
+binding = "PGFLOW_TELEMETRY"
+dataset = "pgflow_telemetry"
+
+# No request logging: bodies and transport metadata are never stored.
+[observability]
+enabled = false

pkgs/core/schemas/0055_tables_workers.sql

@@ -7,8 +7,14 @@ create table if not exists pgflow.workers (
   started_at timestamptz not null default now(),
   deprecated_at timestamptz,
   stopped_at timestamptz,
-  last_heartbeat_at timestamptz not null default now()
+  last_heartbeat_at timestamptz not null default now(),
+  -- Version of the pgflow package the worker was built from, stamped once at
+  -- registration. Nullable: rows written before telemetry exist. Telemetry
+  -- reads a per-day distribution; MAX() is the newest version ever run.
+  pgflow_version text
 );
 
 create index if not exists idx_workers_queue_name on pgflow.workers (queue_name);
 create index if not exists idx_workers_heartbeat on pgflow.workers (last_heartbeat_at);
+create index if not exists idx_workers_started_at on pgflow.workers (started_at);
+create index if not exists idx_workers_stopped_at on pgflow.workers (stopped_at);

pkgs/core/schemas/0060_tables_runtime.sql

@@ -19,6 +19,7 @@ create table pgflow.runs (
 
 create index if not exists idx_runs_flow_slug on pgflow.runs (flow_slug);
 create index if not exists idx_runs_status on pgflow.runs (status);
+create index if not exists idx_runs_started_at on pgflow.runs (started_at);
 
 -- Step states table - tracks the state of individual steps within a run
 create table pgflow.step_states (

pkgs/core/schemas/0130_schema_pgflow_telemetry.sql

@@ -0,0 +1,5 @@
+-- Self-contained telemetry schema: droppable without touching pgflow engine
+-- objects. The cron.job row is the enabled switch; no settings table.
+create schema if not exists pgflow_telemetry;
+
+revoke all on schema pgflow_telemetry from public;

pkgs/core/schemas/0131_table_sent_reports.sql

@@ -0,0 +1,12 @@
+-- One row per reported UTC day. The unique day doubles as the dedup marker
+-- (no separate state row) and the payload is the exact bytes that left the
+-- database, so users can audit every report.
+create table pgflow_telemetry.sent_reports (
+  day date primary key,
+  payload jsonb not null,
+  request_id bigint,
+  sent_at timestamptz not null default now()
+);
+
+comment on table pgflow_telemetry.sent_reports is
+'Audit log of every telemetry payload sent; unique day is the dedup marker';