From 7a3f46e4149694711d4e5504aa64071b873c97b4 Mon Sep 17 00:00:00 2001 From: Anthony Ronning <101225832+AnthonyRonning@users.noreply.github.com> Date: Tue, 27 Oct 2026 03:12:58 +0000 Subject: [PATCH] Record private unsampled download events in D1 --- .github/workflows/agent-updates-tests.yml | 2 + .github/workflows/updates-tests.yml | 2 + scripts/ci/agent-updates.sh | 1 + scripts/ci/change_detection.py | 2 +- scripts/ci/hook_change_detection.py | 3 + scripts/ci/opensecret_change_detection.py | 2 +- scripts/ci/test_hook_change_detection.py | 3 + services/agent-updates/README.md | 3 + .../migrations/0001_download_events.sql | 22 + services/agent-updates/package.json | 4 +- services/agent-updates/src/dry-run.ts | 12 + .../src/tracking-runtime.test.ts | 275 +++++++++ services/agent-updates/src/tracking.test.ts | 528 ++++++++++++++++++ services/agent-updates/src/worker.ts | 30 +- services/agent-updates/wrangler.jsonc | 30 + services/download-tracking/README.md | 172 ++++++ services/download-tracking/query.py | 247 ++++++++ services/download-tracking/test_query.py | 180 ++++++ services/download-tracking/tracking.ts | 491 ++++++++++++++++ services/updates/README.md | 6 +- services/updates/package.json | 4 +- services/updates/src/worker.ts | 17 +- services/updates/wrangler.jsonc | 12 + 23 files changed, 2037 insertions(+), 11 deletions(-) create mode 100644 services/agent-updates/migrations/0001_download_events.sql create mode 100644 services/agent-updates/src/tracking-runtime.test.ts create mode 100644 services/agent-updates/src/tracking.test.ts create mode 100644 services/download-tracking/README.md create mode 100644 services/download-tracking/query.py create mode 100644 services/download-tracking/test_query.py create mode 100644 services/download-tracking/tracking.ts diff --git a/.github/workflows/agent-updates-tests.yml b/.github/workflows/agent-updates-tests.yml index d92e7e286..2578f846f 100644 --- a/.github/workflows/agent-updates-tests.yml +++ b/.github/workflows/agent-updates-tests.yml @@ -9,6 +9,7 @@ on: paths: - ".github/workflows/agent-updates-*.yml" - "services/agent-updates/**" + - "services/download-tracking/**" - "scripts/ci/agent-updates.sh" - "scripts/ci/test_agent_updates_workflows.py" - "scripts/ci/_common.sh" @@ -20,6 +21,7 @@ on: paths: - ".github/workflows/agent-updates-*.yml" - "services/agent-updates/**" + - "services/download-tracking/**" - "scripts/ci/agent-updates.sh" - "scripts/ci/test_agent_updates_workflows.py" - "scripts/ci/_common.sh" diff --git a/.github/workflows/updates-tests.yml b/.github/workflows/updates-tests.yml index ad0c0168e..cb617b94d 100644 --- a/.github/workflows/updates-tests.yml +++ b/.github/workflows/updates-tests.yml @@ -10,6 +10,7 @@ on: - ".github/workflows/updates-tests.yml" - ".github/workflows/updates-publish.yml" - "services/updates/**" + - "services/download-tracking/**" - "scripts/ci/_common.sh" - "scripts/ci/updates.sh" - "flake.nix" @@ -20,6 +21,7 @@ on: - ".github/workflows/updates-tests.yml" - ".github/workflows/updates-publish.yml" - "services/updates/**" + - "services/download-tracking/**" - "scripts/ci/_common.sh" - "scripts/ci/updates.sh" - "flake.nix" diff --git a/scripts/ci/agent-updates.sh b/scripts/ci/agent-updates.sh index b4c9de1c1..e7029fc8d 100755 --- a/scripts/ci/agent-updates.sh +++ b/scripts/ci/agent-updates.sh @@ -8,6 +8,7 @@ print_source_provenance export WRANGLER_SEND_METRICS=false cd "${REPO_ROOT}" python3 scripts/ci/test_agent_updates_workflows.py +python3 -I -m unittest discover -s services/download-tracking -p 'test_*.py' python3 -I -m unittest discover -s services/agent-updates/publisher -p 'test_*.py' cd services/agent-updates bun install --frozen-lockfile --ignore-scripts diff --git a/scripts/ci/change_detection.py b/scripts/ci/change_detection.py index 236dd5d2c..8378ededc 100644 --- a/scripts/ci/change_detection.py +++ b/scripts/ci/change_detection.py @@ -31,7 +31,7 @@ "apps/maple-research/zapstore.yaml", } ) -INERT_PREFIXES = (".agents/", ".githooks/", "docs/", "apps/maple-research/docs/", "apps/maple-research/.githooks/", "services/updates/", "services/opensecret/", "apps/maple-agent/", "apps/maple-auth/") +INERT_PREFIXES = (".agents/", ".githooks/", "docs/", "apps/maple-research/docs/", "apps/maple-research/.githooks/", "services/updates/", "services/download-tracking/", "services/opensecret/", "apps/maple-agent/", "apps/maple-auth/") PURE_FRONTEND_PREFIXES = ("apps/maple-research/frontend/public/", "apps/maple-research/frontend/src/") PURE_FRONTEND_FILES = frozenset({"apps/maple-research/frontend/icon.svg", "apps/maple-research/frontend/index.html"}) RUST_SDK_PREFIX = "sdk/rust/" diff --git a/scripts/ci/hook_change_detection.py b/scripts/ci/hook_change_detection.py index 59a525ee6..d4fee25ac 100644 --- a/scripts/ci/hook_change_detection.py +++ b/scripts/ci/hook_change_detection.py @@ -165,6 +165,9 @@ def classify_path(path: str) -> frozenset[str]: return frozenset() return frozenset({"opensecret"}) + if path.startswith("services/download-tracking/"): + return frozenset({"updates", "agent_updates"}) + if path.startswith(UPDATES_PREFIX + ".githooks/"): return frozenset() if path.startswith(UPDATES_PREFIX): diff --git a/scripts/ci/opensecret_change_detection.py b/scripts/ci/opensecret_change_detection.py index 7a0de0ede..534848d78 100644 --- a/scripts/ci/opensecret_change_detection.py +++ b/scripts/ci/opensecret_change_detection.py @@ -44,7 +44,7 @@ ".agents/skills/validate-opensecret/scripts/disposable_db_tests.sh", }) KNOWN_INDEPENDENT_PREFIXES = ( - "apps/", "proxy/", "services/updates/", ".agents/", ".github/", + "apps/", "proxy/", "services/updates/", "services/download-tracking/", ".agents/", ".github/", ".githooks/", "docs/", "scripts/", ) KNOWN_INDEPENDENT_FILES = frozenset({ diff --git a/scripts/ci/test_hook_change_detection.py b/scripts/ci/test_hook_change_detection.py index 06c8d7b1b..d6027772f 100644 --- a/scripts/ci/test_hook_change_detection.py +++ b/scripts/ci/test_hook_change_detection.py @@ -102,6 +102,9 @@ def test_proxy_opensecret_and_updates(self) -> None: self.assert_selects(["services/opensecret/nix/eif.nix"]) self.assert_selects(["services/updates/src/index.ts"], "updates") + def test_download_tracking_selects_both_workers(self) -> None: + self.assert_selects(["services/download-tracking/tracking.ts"], "updates", "agent_updates") + def test_agent_updates_component_is_isolated_from_research(self) -> None: for path in ( "services/agent-updates/src/worker.ts", diff --git a/services/agent-updates/README.md b/services/agent-updates/README.md index a8f64be38..b499ab43c 100644 --- a/services/agent-updates/README.md +++ b/services/agent-updates/README.md @@ -8,6 +8,9 @@ See [publisher tooling](publisher/README.md) for exact commands and retry behavi The checked-in binding names the owner-provisioned private `maple-agent-updates` bucket. This is configuration, not evidence of Worker deployment or activated delivery. There is no deploy script or automatic deployment workflow. +Private download metrics use the shared Queue and D1 ledger described in +[download tracking](../download-tracking/README.md). The existing Agent Worker +consumes events from both update Workers; it has no public metrics endpoint. `workers_dev` and preview URLs are off; routine configuration must never contain `route`, `routes` or a Custom Domain. Local checks use a separate fixture bucket. diff --git a/services/agent-updates/migrations/0001_download_events.sql b/services/agent-updates/migrations/0001_download_events.sql new file mode 100644 index 000000000..dad593538 --- /dev/null +++ b/services/agent-updates/migrations/0001_download_events.sql @@ -0,0 +1,22 @@ +-- Response starts, not unique users, completed transfers, or installations. +-- No path, query, IP, raw User-Agent, cookies, or client identifiers are stored. +CREATE TABLE download_events ( + event_id TEXT PRIMARY KEY NOT NULL, + observed_at TEXT NOT NULL, + product TEXT NOT NULL CHECK (product IN ('agent', 'research')), + host TEXT NOT NULL CHECK (host IN ('updates.maple.ai', 'updates.trymaple.ai')), + version TEXT, + platform TEXT NOT NULL CHECK (platform IN ('macos', 'windows', 'linux', 'android', 'all')), + artifact TEXT NOT NULL CHECK (artifact IN ('macos-dmg', 'macos-updater', 'windows-installer', 'linux-bundle', 'linux-appimage', 'linux-deb', 'linux-rpm', 'android-apk', 'latest', 'manifest')), + kind TEXT NOT NULL CHECK (kind IN ('artifact', 'alias', 'metadata')), + method TEXT NOT NULL CHECK (method IN ('GET', 'HEAD')), + status INTEGER NOT NULL CHECK (status BETWEEN 200 AND 599), + range TEXT NOT NULL CHECK (range IN ('none', 'requested', 'partial', 'unsatisfiable')), + result TEXT NOT NULL CHECK (result IN ('success', 'redirect', 'not_modified', 'client_error', 'server_error')), + response_length INTEGER CHECK (response_length BETWEEN 0 AND 4294967296), + client_kind TEXT NOT NULL CHECK (client_kind IN ('browser', 'maple', 'cli', 'other')), + is_probe INTEGER NOT NULL CHECK (is_probe IN (0, 1)) +); + +CREATE INDEX download_events_observed_at ON download_events(observed_at); +CREATE INDEX download_events_trends ON download_events(product, kind, observed_at); diff --git a/services/agent-updates/package.json b/services/agent-updates/package.json index b11bac800..577c2fccf 100644 --- a/services/agent-updates/package.json +++ b/services/agent-updates/package.json @@ -6,8 +6,8 @@ "scripts": { "check": "bun run format:check && bun run typecheck && bun run test && bun run build:dry-run", "build:dry-run": "bun src/dry-run.ts", - "format": "prettier --write \"src/**/*.ts\" README.md package.json tsconfig.json wrangler.jsonc", - "format:check": "prettier --check \"src/**/*.ts\" README.md package.json tsconfig.json wrangler.jsonc", + "format": "prettier --write \"src/**/*.ts\" ../download-tracking/tracking.ts ../download-tracking/README.md README.md package.json tsconfig.json wrangler.jsonc", + "format:check": "prettier --check \"src/**/*.ts\" ../download-tracking/tracking.ts ../download-tracking/README.md README.md package.json tsconfig.json wrangler.jsonc", "test": "bun --no-env-file test", "typecheck": "tsc --noEmit" }, diff --git a/services/agent-updates/src/dry-run.ts b/services/agent-updates/src/dry-run.ts index d5f7b3f4f..0cb7ebf79 100644 --- a/services/agent-updates/src/dry-run.ts +++ b/services/agent-updates/src/dry-run.ts @@ -38,6 +38,18 @@ try { bucket_name: "agent-updates-local-dry-run", }, ], + d1_databases: [ + { + binding: "DOWNLOAD_DB", + database_name: "downloads-local-dry-run", + database_id: "00000000-0000-4000-8000-000000000001", + }, + ], + queues: { + producers: [ + { binding: "DOWNLOAD_EVENTS", queue: "downloads-local-dry-run" }, + ], + }, }), ); const result = Bun.spawn( diff --git a/services/agent-updates/src/tracking-runtime.test.ts b/services/agent-updates/src/tracking-runtime.test.ts new file mode 100644 index 000000000..c4d4207fc --- /dev/null +++ b/services/agent-updates/src/tracking-runtime.test.ts @@ -0,0 +1,275 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, test } from "bun:test"; +import { prepareInstallerCatalog } from "../../updates/src/prepare-installers"; +import { fixture as researchFixture } from "../../updates/src/test-fixtures"; +import { fixture, PAYLOAD } from "./fixtures"; + +// Use Wrangler's locked Miniflare/workerd dependency. All storage, queues and +// fixture-only control routes below live in one disposable local runtime. +const server = ` + import { Miniflare, convertV4MiniflareOptions } from 'miniflare'; + import { readFileSync } from 'node:fs'; + const mf = new Miniflare(convertV4MiniflareOptions({ + modules: true, name: 'download-tracking-test', + script: readFileSync(process.env.MAPLE_TEST_WORKER, 'utf8'), + compatibilityDate: '2026-08-24', port: 0, inspectorPort: 0, + r2Buckets: ['AGENT_RELEASES'], d1Databases: ['DOWNLOAD_DB'], + queueProducers: { DOWNLOAD_EVENTS: 'download-events-test' }, + queueConsumers: { 'download-events-test': { + maxBatchSize: 1, maxBatchTimeout: 0, retryDelay: 1, maxRetries: 5, + } }, + })); + process.on('SIGTERM', async () => { await mf.dispose(); process.exit(0); }); + process.send({ origin: (await mf.ready).origin }); +`; + +async function eventually( + read: () => Promise, + accept: (value: T) => boolean, +): Promise { + const deadline = Date.now() + 15000; + let last: T; + do { + last = await read(); + if (accept(last)) return last; + await Bun.sleep(100); + } while (Date.now() < deadline); + throw new Error( + `tracking runtime condition timed out: ${JSON.stringify(last)}`, + ); +} + +test("real workerd queues persist Agent and Research request events once and retry a D1 outage", async () => { + const directory = await mkdtemp(join(tmpdir(), "download-tracking-workerd-")); + let child: ReturnType | undefined; + let timer: ReturnType | undefined; + try { + const wrapper = join(directory, "entry.ts"); + await Bun.write( + wrapper, + ` + import agent from ${JSON.stringify(join(import.meta.dir, "index.ts"))}; + import research from ${JSON.stringify(join(import.meta.dir, "../../updates/src/index.ts"))}; + export default { + async queue(batch, env, ctx) { + await agent.queue(batch, env, ctx); + await env.DOWNLOAD_DB.prepare('INSERT INTO test_deliveries VALUES (?)').bind(batch.messages.length).run(); + }, + async fetch(request, env, ctx) { + const path = new URL(request.url).pathname; + if (path === '/__seed' && request.method === 'POST') { + const { metadata, payload, latest, catalog, migration } = await request.json(); + for (const statement of migration.split(';')) { + if (statement.trim()) await env.DOWNLOAD_DB.prepare(statement).run(); + } + await env.DOWNLOAD_DB.prepare('CREATE TABLE test_deliveries (message_count INTEGER NOT NULL)').run(); + await env.AGENT_RELEASES.put('stable/latest.json', Uint8Array.from(metadata)); + await env.AGENT_RELEASES.put('stable/releases/0.0.2/manifest.json', Uint8Array.from(metadata)); + await env.AGENT_RELEASES.put('stable/releases/0.0.2/Maple-Agent.app.tar.gz', Uint8Array.from(payload)); + await env.AGENT_RELEASES.put('stable/releases/0.0.2/Maple-Agent.dmg', Uint8Array.from(payload)); + await env.AGENT_RELEASES.put('test/latest.json', Uint8Array.from(latest)); + await env.AGENT_RELEASES.put('test/installers.json', JSON.stringify(catalog)); + return new Response('seeded'); + } + if (path === '/__rows') { + const rows = await env.DOWNLOAD_DB.prepare('SELECT * FROM download_events ORDER BY observed_at, event_id').all(); + const delivered = await env.DOWNLOAD_DB.prepare('SELECT COALESCE(SUM(message_count), 0) AS count FROM test_deliveries').first(); + return Response.json({ rows: rows.results, deliveries: delivered.count }); + } + if (path === '/__replay' && request.method === 'POST') { + const event = await env.DOWNLOAD_DB.prepare('SELECT * FROM download_events LIMIT 1').first(); + await env.DOWNLOAD_EVENTS.send({ schema_version: 1, ...event }); + await env.DOWNLOAD_EVENTS.send({ schema_version: 1, ...event }); + return new Response('replayed'); + } + if (path === '/__outage' && request.method === 'POST') { + await env.DOWNLOAD_DB.prepare('ALTER TABLE download_events RENAME TO test_unavailable_events').run(); + return new Response('unavailable'); + } + if (path === '/__attempts') { + const row = await env.DOWNLOAD_DB.prepare('SELECT COALESCE(SUM(message_count), 0) AS count FROM test_deliveries').first(); + return Response.json(row); + } + if (path === '/__recover' && request.method === 'POST') { + await env.DOWNLOAD_DB.prepare('ALTER TABLE test_unavailable_events RENAME TO download_events').run(); + return new Response('recovered'); + } + const canonical = new Request('https://updates.maple.ai' + path + new URL(request.url).search, request); + if (path.startsWith('/agent/') || path.startsWith('/download/agent/')) { + return agent.fetch(canonical, env, ctx); + } + return research.fetch(canonical, { + DOWNLOAD_EVENTS: env.DOWNLOAD_EVENTS, + ASSETS: { async fetch(assetRequest) { + const object = await env.AGENT_RELEASES.get('test' + new URL(assetRequest.url).pathname); + return object ? new Response(object.body) : new Response(null, { status: 404 }); + } }, + }, ctx); + }, + }; + `, + ); + const build = await Bun.build({ + entrypoints: [wrapper], + outdir: directory, + target: "browser", + format: "esm", + }); + if (!build.success) throw new Error("tracking fixture bundle failed"); + const origin = await new Promise((resolve, reject) => { + timer = setTimeout( + () => reject(new Error("tracking workerd startup timeout")), + 20000, + ); + child = Bun.spawn(["node", "--input-type=module", "-e", server], { + env: { + ...process.env, + MAPLE_TEST_WORKER: join(directory, "entry.js"), + }, + stdout: "ignore", + stderr: "inherit", + serialization: "json", + ipc(value: unknown) { + if ( + value && + typeof value === "object" && + "origin" in value && + typeof value.origin === "string" + ) + resolve(value.origin); + }, + onExit(_child, code) { + reject(new Error(`tracking workerd exited ${code}`)); + }, + }); + }); + clearTimeout(timer); + const { release, latestBytes } = researchFixture(); + expect( + ( + await fetch(`${origin}/__seed`, { + method: "POST", + body: JSON.stringify({ + migration: await Bun.file( + join(import.meta.dir, "../migrations/0001_download_events.sql"), + ).text(), + metadata: Array.from(fixture().bytes), + payload: Array.from(PAYLOAD), + latest: Array.from(latestBytes), + catalog: prepareInstallerCatalog(release, latestBytes), + }), + }) + ).status, + ).toBe(200); + const artifact = `${origin}/agent/stable/releases/0.0.2/Maple-Agent.app.tar.gz`; + const full = await fetch(artifact, { + headers: { + authorization: "Bearer private-must-not-be-stored", + "x-forwarded-for": "203.0.113.10", + "user-agent": "private-ua-must-not-be-stored", + }, + }); + expect(full.status).toBe(200); + expect(new Uint8Array(await full.arrayBuffer())).toEqual(PAYLOAD); + const partial = await fetch(artifact, { + headers: { range: "bytes=2-7", "x-maple-download-probe": "1" }, + }); + expect(partial.status).toBe(206); + expect(new Uint8Array(await partial.arrayBuffer())).toEqual( + PAYLOAD.slice(2, 8), + ); + expect((await fetch(artifact, { method: "HEAD" })).status).toBe(200); + expect( + ( + await fetch(`${origin}/download/agent/stable/macos`, { + redirect: "manual", + }) + ).status, + ).toBe(302); + expect((await fetch(`${origin}/agent/stable/latest.json`)).status).toBe( + 200, + ); + expect( + ( + await fetch( + `${origin}/download/research/stable/macos?utm_source=private-query-must-not-be-stored`, + { + redirect: "manual", + }, + ) + ).status, + ).toBe(302); + expect((await fetch(`${origin}/latest.json`)).status).toBe(200); + type Snapshot = { + rows: Array>; + deliveries: number; + }; + const snapshot = () => + fetch(`${origin}/__rows`).then( + (response) => response.json() as Promise, + ); + const first = await eventually( + snapshot, + (value) => value.rows.length === 7 && value.deliveries >= 7, + ); + expect(new Set(first.rows.map((row) => row.event_id)).size).toBe(7); + expect(first.rows.filter((row) => row.product === "research")).toHaveLength( + 2, + ); + expect(first.rows.filter((row) => row.kind === "artifact")).toHaveLength(3); + expect(first.rows.filter((row) => row.is_probe === 1)).toHaveLength(1); + expect(JSON.stringify(first.rows)).not.toContain("private-"); + expect(JSON.stringify(first.rows)).not.toContain("203.0.113.10"); + expect( + first.rows.every( + (row) => + typeof row.observed_at === "string" && + Number.isFinite(Date.parse(row.observed_at)), + ), + ).toBe(true); + + expect((await fetch(`${origin}/__replay`, { method: "POST" })).status).toBe( + 200, + ); + const replayed = await eventually( + snapshot, + (value) => value.deliveries >= first.deliveries + 2, + ); + expect(replayed.rows).toEqual(first.rows); + + expect((await fetch(`${origin}/__outage`, { method: "POST" })).status).toBe( + 200, + ); + const duringOutage = await fetch(artifact); + expect(duringOutage.status).toBe(200); + expect(new Uint8Array(await duringOutage.arrayBuffer())).toEqual(PAYLOAD); + await eventually( + () => + fetch(`${origin}/__attempts`).then( + (response) => response.json() as Promise<{ count: number }>, + ), + (value) => value.count > replayed.deliveries, + ); + expect( + (await fetch(`${origin}/__recover`, { method: "POST" })).status, + ).toBe(200); + const recovered = await eventually( + snapshot, + (value) => value.rows.length === 8, + ); + expect( + recovered.rows.filter((row) => row.kind === "artifact"), + ).toHaveLength(4); + } finally { + clearTimeout(timer); + child?.kill("SIGTERM"); + if (child) { + const forceStop = setTimeout(() => child?.kill("SIGKILL"), 3000); + await child.exited; + clearTimeout(forceStop); + } + await rm(directory, { recursive: true, force: true }); + } +}, 60000); diff --git a/services/agent-updates/src/tracking.test.ts b/services/agent-updates/src/tracking.test.ts new file mode 100644 index 000000000..792b0ea98 --- /dev/null +++ b/services/agent-updates/src/tracking.test.ts @@ -0,0 +1,528 @@ +import { afterAll, afterEach, expect, spyOn, test } from "bun:test"; +import { Database } from "bun:sqlite"; +import { + consumeEvents, + isDownloadEvent, + trackResponse, + type DownloadEvent, + type EventDatabase, + type EventMessage, + type Product, +} from "../../download-tracking/tracking"; + +const ORIGIN = "https://updates.maple.ai"; +const ARTIFACT = + "/agent/stable/releases/0.0.9/maple-agent-prod-0.0.9-d784804-macos-aarch64.dmg"; +const logs = spyOn(console, "error").mockImplementation(() => {}); +afterEach(() => logs.mockClear()); +afterAll(() => logs.mockRestore()); + +function capture( + path = ARTIFACT, + response = new Response("payload", { headers: { "content-length": "7" } }), + init?: RequestInit, + product: Product = "agent", +) { + const events: DownloadEvent[] = []; + const pending: Promise[] = []; + trackResponse( + new Request(new URL(path, ORIGIN), init), + response, + { + DOWNLOAD_EVENTS: { + async send(event) { + events.push(event); + }, + }, + }, + { + waitUntil(promise) { + pending.push(promise); + }, + }, + product, + ); + return { events, pending, response, done: () => Promise.all(pending) }; +} + +async function event(): Promise { + const observation = capture(); + await observation.done(); + return observation.events[0]; +} + +function message( + body: unknown, +): EventMessage & { acks: number; retries: number } { + return { + body, + acks: 0, + retries: 0, + ack() { + this.acks++; + }, + retry() { + this.retries++; + }, + }; +} + +async function database() { + const sqlite = new Database(":memory:"); + sqlite.exec( + await Bun.file( + new URL("../migrations/0001_download_events.sql", import.meta.url), + ).text(), + ); + const db: EventDatabase = { + prepare(sql) { + return { + bind(...values) { + return { + async run() { + sqlite.prepare(sql).run(...values); + return { success: true }; + }, + }; + }, + }; + }, + }; + return { sqlite, db }; +} + +test("captures original UTC timestamp and random observation ID without reading/replacing bytes or headers", async () => { + const bytes = new Uint8Array([0, 255, 1, 128]); + const response = new Response(bytes, { + headers: { + "content-length": "4", + etag: '"exact"', + "content-type": "application/octet-stream", + "cache-control": "public, max-age=31536000, immutable", + }, + }); + const body = response.body; + const headers = [...response.headers]; + const before = Date.now(); + const observation = capture(ARTIFACT, response); + expect(observation.events).toHaveLength(0); + expect(response.body).toBe(body); + expect(response.bodyUsed).toBe(false); + expect([...response.headers]).toEqual(headers); + await observation.done(); + const [captured] = observation.events; + expect(isDownloadEvent(captured)).toBe(true); + expect(Date.parse(captured.observed_at)).toBeGreaterThanOrEqual(before); + expect(captured).toMatchObject({ + product: "agent", + host: "updates.maple.ai", + version: "0.0.9", + platform: "macos", + artifact: "macos-dmg", + kind: "artifact", + method: "GET", + status: 200, + range: "none", + result: "success", + response_length: 4, + is_probe: 0, + }); + expect(response.bodyUsed).toBe(false); + expect(new Uint8Array(await response.arrayBuffer())).toEqual(bytes); + const second = await event(); + expect(second.event_id).not.toBe(captured.event_id); +}); + +test("nonblocking queue outage retries with the same ID and only sanitized errors", async () => { + const pending: Promise[] = []; + const sent: DownloadEvent[] = []; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const response = new Response("still delivered", { + headers: { "content-length": "15" }, + }); + trackResponse( + new Request(ORIGIN + ARTIFACT), + response, + { + DOWNLOAD_EVENTS: { + async send(item) { + sent.push(item); + await gate; + throw new Error("private SQL URL token"); + }, + }, + }, + { + waitUntil(promise) { + pending.push(promise); + }, + }, + "agent", + ); + expect(await response.text()).toBe("still delivered"); + release(); + await Promise.all(pending); + expect(sent).toHaveLength(2); + expect(sent[0].event_id).toBe(sent[1].event_id); + expect(sent[0].observed_at).toBe(sent[1].observed_at); + expect(logs.mock.calls).toEqual( + Array(2).fill([ + '{"service":"download-tracking","code":"queue_send_failed"}', + ]), + ); +}); + +test("missing binding or failed scheduling never affects availability", async () => { + const response = new Response("same"); + const request = new Request(ORIGIN + ARTIFACT); + const pending: Promise[] = []; + expect(() => + trackResponse( + request, + response, + {}, + { + waitUntil() { + throw new Error("unused"); + }, + }, + "agent", + ), + ).not.toThrow(); + expect(() => + trackResponse( + request, + response, + { DOWNLOAD_EVENTS: { async send() {} } }, + { + waitUntil(promise) { + pending.push(promise); + throw new Error("private scheduler error"); + }, + }, + "agent", + ), + ).not.toThrow(); + await Promise.all(pending); + expect(await response.text()).toBe("same"); + expect(logs.mock.calls).toEqual([ + ['{"service":"download-tracking","code":"queue_binding_missing"}'], + ['{"service":"download-tracking","code":"capture_failed"}'], + ]); +}); + +test("missing binding logs only for classified production routes", () => { + const context = { + waitUntil() { + throw new Error("must not schedule"); + }, + }; + for (const url of [ + ORIGIN + "/private", + "https://attacker.example" + ARTIFACT, + ]) + trackResponse(new Request(url), new Response(null), {}, context, "agent"); + expect(logs.mock.calls).toHaveLength(0); +}); + +test("alias redirect intent and follow-up payload are separate kinds", async () => { + const alias = capture( + "/download/agent/stable/macos", + new Response(null, { + status: 302, + headers: { location: ARTIFACT }, + }), + ); + await alias.done(); + expect(alias.events[0]).toMatchObject({ + kind: "alias", + version: "0.0.9", + artifact: "macos-dmg", + result: "redirect", + }); + expect(isDownloadEvent(alias.events[0])).toBe(true); + expect((await event()).kind).toBe("artifact"); +}); + +test("Research aliases record only validated GitHub redirect versions", async () => { + const cases = [ + ["macos", "Maple_3.4.5_universal.dmg", "macos-dmg", "macos"], + ["windows", "Maple_3.4.5_x64-setup.exe", "windows-installer", "windows"], + ["linux-appimage", "Maple_3.4.5_amd64.AppImage", "linux-appimage", "linux"], + ["linux-deb", "Maple_3.4.5_amd64.deb", "linux-deb", "linux"], + ["linux-rpm", "Maple-3.4.5-1.x86_64.rpm", "linux-rpm", "linux"], + ["android", "app-universal-release.apk", "android-apk", "android"], + ]; + for (const [key, file, artifact, platform] of cases) { + const observation = capture( + `/download/research/stable/${key}`, + new Response(null, { + status: 302, + headers: { + location: `https://github.com/MaplePrivacyLabs/Maple/releases/download/v3.4.5/${file}`, + }, + }), + undefined, + "research", + ); + await observation.done(); + expect(observation.events[0]).toMatchObject({ + kind: "alias", + version: "3.4.5", + product: "research", + artifact, + platform, + }); + expect(isDownloadEvent(observation.events[0])).toBe(true); + } + const untrusted = capture( + "/download/research/stable/macos", + new Response(null, { + status: 302, + headers: { location: "https://untrusted.example/private?token=secret" }, + }), + undefined, + "research", + ); + await untrusted.done(); + expect(untrusted.events[0].version).toBeNull(); + expect(JSON.stringify(untrusted.events)).not.toContain("secret"); +}); + +test("Research marketing query requests are counted without retaining query values", async () => { + for (const path of ["/latest.json", "/download/research/stable/macos"]) { + const observation = capture( + path + "?utm_source=private-query&token=not-for-storage", + new Response(null), + undefined, + "research", + ); + await observation.done(); + expect(observation.events).toHaveLength(1); + expect(isDownloadEvent(observation.events[0])).toBe(true); + const serialized = JSON.stringify(observation.events[0]); + expect(serialized).not.toContain("private-query"); + expect(serialized).not.toContain("not-for-storage"); + } +}); + +test("HEAD, ranges, validators, errors and self-reported probes remain distinct", async () => { + for (const [status, range, method, expectedRange, result] of [ + [200, null, "HEAD", "none", "success"], + [206, "bytes=0-1", "GET", "partial", "success"], + [416, "bytes=999-", "GET", "unsatisfiable", "client_error"], + [200, "bytes=0-1", "GET", "requested", "success"], + [304, null, "GET", "none", "not_modified"], + [404, null, "GET", "none", "client_error"], + [503, null, "GET", "none", "server_error"], + ] as const) { + const observation = capture(ARTIFACT, new Response(null, { status }), { + method, + headers: { + ...(range ? { range } : {}), + "x-maple-download-probe": "1", + "user-agent": "Maple-Agent-Updater/1", + }, + }); + await observation.done(); + expect(observation.events[0]).toMatchObject({ + method, + status, + range: expectedRange, + result, + is_probe: 1, + client_kind: "maple", + }); + expect(isDownloadEvent(observation.events[0])).toBe(true); + } +}); + +test("metadata capture never parses or rewrites bodies and needs no version assertion", async () => { + for (const [path, product, artifact, version, platform] of [ + ["/latest.json", "research", "latest", null, "all"], + ["/agent/stable/latest.json", "agent", "latest", null, "macos"], + ["/agent/stable/linux/latest.json", "agent", "latest", null, "linux"], + [ + "/agent/stable/releases/0.0.9/manifest.json", + "agent", + "manifest", + "0.0.9", + "macos", + ], + [ + "/agent/stable/releases/0.0.9/linux-manifest.json", + "agent", + "manifest", + "0.0.9", + "linux", + ], + ] as const) { + const observation = capture( + path, + new Response("unparsed original bytes"), + undefined, + product, + ); + await observation.done(); + expect(observation.events[0]).toMatchObject({ + kind: "metadata", + artifact, + version, + platform, + }); + expect(isDownloadEvent(observation.events[0])).toBe(true); + expect(observation.response.bodyUsed).toBe(false); + expect(await observation.response.text()).toBe("unparsed original bytes"); + } +}); + +test("privacy allowlist omits arbitrary routes, query strings, methods, hosts and private request fields", async () => { + for (const path of [ + "/private?token=secret", + ARTIFACT + "?token=secret", + "/installers.json", + "/download/agent/stable/constructor", + "/download/agent/stable/__proto__", + "/agent/stable/releases/0.0.9/private.txt", + "/agent/stable/releases/not-a-version/app.dmg", + "https://attacker.example" + ARTIFACT, + "http://updates.maple.ai" + ARTIFACT, + "/agent/stable/releases/0.0.9/app%2Fdmg", + "/download/agent/stable/macos/extra", + ]) { + const observation = capture(path); + await observation.done(); + expect(observation.events).toHaveLength(0); + } + const post = capture(ARTIFACT, new Response(null, { status: 405 }), { + method: "POST", + }); + await post.done(); + expect(post.events).toHaveLength(0); + const observation = capture(ARTIFACT, new Response(null), { + headers: { + authorization: "Bearer private", + cookie: "private", + "cf-connecting-ip": "192.0.2.1", + "user-agent": "curl/8 private custom identifier", + "x-maple-download-probe": "false", + }, + }); + await observation.done(); + expect(observation.events[0]).toMatchObject({ + client_kind: "cli", + is_probe: 0, + }); + const serialized = JSON.stringify(observation.events[0]); + for (const text of ["private", "192.0.2.1", ARTIFACT, "curl/8"]) + expect(serialized).not.toContain(text); +}); + +test("content length is bounded expected response size, never transferred bytes", async () => { + for (const length of [ + "-1", + "1.5", + "9007199254740992", + "4294967297", + "private", + ]) { + const observation = capture( + ARTIFACT, + new Response(null, { headers: { "content-length": length } }), + ); + await observation.done(); + expect(observation.events[0].response_length).toBeNull(); + } +}); + +test("D1 commits original timestamps and deduplicates ambiguous/redelivered queue writes", async () => { + const { sqlite, db } = await database(); + try { + const body = await event(); + body.observed_at = "2026-10-27T12:00:00.000Z"; + const first = message(body); + const duplicate = message(structuredClone(body)); + await consumeEvents({ messages: [first, duplicate] }, { DOWNLOAD_DB: db }); + expect(first).toMatchObject({ acks: 1, retries: 0 }); + expect(duplicate).toMatchObject({ acks: 1, retries: 0 }); + expect( + sqlite.query("SELECT COUNT(*) AS count FROM download_events").get(), + ).toEqual({ count: 1 }); + expect( + sqlite.query("SELECT observed_at, version FROM download_events").get(), + ).toEqual({ observed_at: body.observed_at, version: "0.0.9" }); + } finally { + sqlite.close(); + } +}); + +test("SQL outage and unsuccessful D1 result retry without acknowledging or leaking exceptions", async () => { + for (const run of [ + async () => { + throw new Error("private database secret"); + }, + async () => ({ success: false }), + ]) { + const item = message(await event()); + await consumeEvents( + { messages: [item] }, + { + DOWNLOAD_DB: { + prepare() { + return { + bind() { + return { run }; + }, + }; + }, + }, + }, + ); + expect(item).toMatchObject({ acks: 0, retries: 1 }); + } + expect(logs.mock.calls).toEqual( + Array(2).fill([ + '{"service":"download-tracking","code":"database_write_failed"}', + ]), + ); +}); + +test("invalid or poison events retry for configured DLQ, never insert or log their content", async () => { + const good = await event(); + const bad: unknown[] = [ + null, + "private", + { ...good, ip: "192.0.2.1" }, + { ...good, event_id: "private" }, + { ...good, observed_at: "2026-02-30T00:00:00.000Z" }, + { ...good, host: "private.example" }, + { ...good, version: "x".repeat(1000) }, + { ...good, product: "research" }, + { ...good, status: 503 }, + { ...good, response_length: -1 }, + { ...good, kind: "metadata" }, + { ...good, platform: "windows" }, + { ...good, is_probe: "1" }, + ]; + const items = bad.map(message); + let writes = 0; + await consumeEvents( + { messages: items }, + { + DOWNLOAD_DB: { + prepare() { + writes++; + throw new Error("must not prepare"); + }, + }, + }, + ); + expect(writes).toBe(0); + for (const item of items) expect(item).toMatchObject({ acks: 0, retries: 1 }); + expect(logs.mock.calls).toEqual( + bad.map(() => ['{"service":"download-tracking","code":"invalid_event"}']), + ); +}); diff --git a/services/agent-updates/src/worker.ts b/services/agent-updates/src/worker.ts index c9401c0d8..8ddca65dd 100644 --- a/services/agent-updates/src/worker.ts +++ b/services/agent-updates/src/worker.ts @@ -1,3 +1,11 @@ +import { + trackResponse, + consumeEvents, + type TrackingEnv, + type TrackingContext, + type EventDatabase, + type EventBatch, +} from "../../download-tracking/tracking"; import { MAX_ENVELOPE_BYTES, parseEnvelope, @@ -18,8 +26,9 @@ export interface Bucket { ): Promise; head(key: string): Promise; } -export interface Env { +export interface Env extends TrackingEnv { AGENT_RELEASES: Bucket; + DOWNLOAD_DB?: EventDatabase; } const NO_STORE = { @@ -191,7 +200,7 @@ async function artifactResponse( return new Response(full.body, { status, headers }); } -export default { +const deliveryWorker = { async fetch(request: Request, env: Env): Promise { const url = new URL(request.url); const path = url.pathname; @@ -307,3 +316,20 @@ export default { } }, }; + +export default { + async fetch( + request: Request, + env: Env, + ctx?: TrackingContext, + ): Promise { + const result = await deliveryWorker.fetch(request, env); + if (ctx) trackResponse(request, result, env, ctx, "agent"); + return result; + }, + async queue(batch: EventBatch, env: Env): Promise { + if (!env.DOWNLOAD_DB) + throw new Error("download tracking database binding missing"); + await consumeEvents(batch, { DOWNLOAD_DB: env.DOWNLOAD_DB }); + }, +}; diff --git a/services/agent-updates/wrangler.jsonc b/services/agent-updates/wrangler.jsonc index 557fe145c..df32e8b46 100644 --- a/services/agent-updates/wrangler.jsonc +++ b/services/agent-updates/wrangler.jsonc @@ -13,4 +13,34 @@ "bucket_name": "maple-agent-updates", }, ], + "d1_databases": [ + { + "binding": "DOWNLOAD_DB", + "database_name": "maple-downloads", + "database_id": "d47c0388-7e13-4474-a535-8f29dfd9138b", + "migrations_dir": "migrations", + }, + ], + "queues": { + "producers": [ + { "binding": "DOWNLOAD_EVENTS", "queue": "maple-download-events" }, + ], + "consumers": [ + { + "queue": "maple-download-events", + "max_batch_size": 25, + "max_batch_timeout": 5, + "max_retries": 10, + "retry_delay": 60, + "dead_letter_queue": "maple-download-events-dlq", + }, + ], + }, + "observability": { + "logs": { + "enabled": true, + "head_sampling_rate": 1, + "invocation_logs": false, + }, + }, } diff --git a/services/download-tracking/README.md b/services/download-tracking/README.md new file mode 100644 index 000000000..c70cc7190 --- /dev/null +++ b/services/download-tracking/README.md @@ -0,0 +1,172 @@ +# Private download tracking + +The two update Workers record unsampled response observations in one private D1 +ledger. They enqueue through `DOWNLOAD_EVENTS`; the Agent Worker consumes the +queue and writes through `DOWNLOAD_DB`. Reporting uses Cloudflare's authenticated +account API. There is no public statistics endpoint or dashboard. + +## What the counts mean + +Each eligible HTTP response gets a random event UUID and the original server UTC +timestamp. Queue redelivery and the producer's single retry preserve that UUID; +`INSERT ... ON CONFLICT(event_id) DO NOTHING` prevents duplicate ledger rows. +Separate HTTP requests get separate IDs, including client retries. + +These are exact counts of persisted observations, with no sampling or estimates. +They do not measure unique people, completed transfers, or successful installs. +Agent payload events describe an R2 response starting. The stream is returned +unchanged: tracking never tees, clones, buffers or consumes it. `response_length` +and report `expected_response_bytes` describe expected response size, not bytes +actually delivered to a client. + +Research installers remain hosted on GitHub. Their Worker records redirect intent; +it cannot observe the subsequent GitHub transfer. Research native updater payloads +also go directly to GitHub. Agent aliases and their follow-up payload responses +are separate event kinds. Never add aliases and payloads together as downloads. + +Reports keep `kind`, method, status and range separate. A useful Agent full-payload +view filters `kind=artifact`, `method=GET`, `status=200`; `range=requested` can still +be a full response after an `If-Range` mismatch. `206` is a partial response, +`416` is an unsatisfied range, and `304` transfers no payload. HEAD and metadata +polls are also recorded separately. Probe exclusion is the reporting default. + +## Availability and retry boundaries + +Capture runs under `waitUntil` after response construction and never awaits the +queue on the download response path. A failed send receives one bounded retry +with the same ID. A missing binding, scheduling failure, or final enqueue failure +fails open: the download continues and observations can be lost before the queue +accepts them. A fail-open collector cannot promise every HTTP request is counted. + +The consumer validates every bounded schema field and acknowledges only after a +successful D1 insert. SQL errors and invalid messages retry; the configured +consumer uses batches of 25, up to five seconds of batching, ten retries and a +60-second retry delay. Exhausted messages go to `maple-download-events-dlq`. +Transient SQL failure does not change downloads. Invalid events remain poison +messages until their cause is understood; they are never silently accepted. + +Both provisioned queues use a 14-day message-retention setting. Queue and DLQ +retention are finite, so inspect backlog and DLQ failures before messages expire. +Check the live queue settings during incident handling; retention is not a D1 +history policy. The D1 migration has no TTL or cleanup job: committed observations +remain until explicitly deleted, subject to the database's storage limits. Any +future pruning, backup or rollup policy must be deliberate. + +Tracking emits only fixed JSON error codes: `queue_binding_missing`, +`capture_failed`, `queue_send_failed`, `invalid_event`, and `database_write_failed`. +Worker observability enables full-rate console error capture with +`head_sampling_rate: 1` and disables automatic invocation logs. Our emitted error +message never contains exceptions, URLs, request/response bodies, credentials or +event contents. Cloudflare may attach its own operational context, including +request metadata, to a log event; this configuration is not a guarantee that all +provider logs omit URLs, queries or IP addresses. The privacy guarantees below +describe our Queue/D1 event fields and application error messages. Infrastructure +backlog/log monitoring is separate from the private report CLI. + +## Privacy and trust boundaries + +Only the two fixed update hostnames and known route shapes are classified. The +ledger contains product, version when safely known, platform, fixed artifact kind, +response kind, method, status, coarse range/result, expected length, coarse client +kind, timestamp, random per-observation UUID, and a probe flag. No IP, account or +device identity, cookies, raw User-Agent, filename, path, query string, or other +arbitrary header value is retained. The UUID identifies one observation, not a +visitor. Research marketing queries are ignored while classifying the pathname. + +Client kind is derived from a few User-Agent patterns and is self-reported. The +`x-maple-download-probe: 1` flag is also self-reported: anyone can set it. It is a +convenient filter for our verification requests, not authentication or proof that +other traffic is human. Versions on latest-metadata events are null because the +tracker does not read metadata bodies to infer them. + +## Private reports + +Use the pinned repository shell. Inject existing credentials through the owning +SecretSpec route; do not put tokens in arguments, shell history or output. The CLI +reads `CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_API_TOKEN`. A D1 Read token is sufficient +for reporting. It reads the `DOWNLOAD_DB` database ID from Agent's checked-in +`wrangler.jsonc`; `DOWNLOAD_ANALYTICS_DATABASE_ID` optionally overrides that +nonsecret ID. The script creates no resources and issues only fixed SELECTs with +bound values through the private D1 SQL API. HTTP redirects are refused. + +From the repository root, with the existing credential environment already +injected: + +```sh +python3 services/download-tracking/query.py +python3 services/download-tracking/query.py --bucket hour --product agent --kind artifact --method GET --status 200 +python3 services/download-tracking/query.py --product research --kind alias --method GET --status 302 +python3 services/download-tracking/query.py --since 2026-10-27T00:00:00Z --until 2026-10-28T00:00:00Z +python3 services/download-tracking/query.py --health --include-probes +``` + +The default window is the last seven days, with UTC day buckets. `--since` is +inclusive and `--until` is exclusive. Both require UTC timestamps ending in `Z`, +with seconds or exactly three decimal digits. `--bucket hour` uses UTC hours. +Windows must be positive and at most 31 days. Query longer histories in adjacent +bounded windows. Other optional filters are `--version`, `--platform`, +`--artifact`, `--range` and `--host`; `--help` lists fixed accepted values. + +JSON rows group by UTC bucket, product, version, platform, artifact, kind, status, +method, range and probe flag. Responses are capped at 10,000 grouped rows; an +extra row detects overflow and sets `truncated: true`. Narrow the window or +filters before using a truncated report. Empty buckets have no row. This is not +a sampling limit: stored events are retained. + +`--health` reports the latest persisted timestamp, total matching observations, +and matching observations since the start of the current UTC day, all within the +selected window. A quiet traffic period can look stale. Confirm pipeline health +with a marked probe and its subsequent ledger record, then inspect primary/DLQ +backlog and fixed failure logs. The CLI does not claim queue health or read live +queue state. + +Failures emit numeric JSON codes on stderr, without upstream bodies, exception +messages or URLs: `2` argument syntax, `10` credential/config shape, `11` time +window, `12` filter value, `20` HTTP failure (plus numeric `http_status`), `21` D1 +API rejection, `22` invalid reply shape, `23` oversized reply, `24` network failure. + +## Manual DLQ recovery + +Replay is an authenticated operational action, outside the read-only CLI. Before +replay, repair the schema/storage/consumer failure and inspect fixed failure logs +and both queue backlogs. Preserve the original message; do not fabricate counts +for a capture gap. + +1. Confirm the intended account, primary queue and DLQ, including live retention. + Use an explicitly authorized temporary pull consumer for the DLQ, without + replacing the primary queue's deployed consumer. Pull a small bounded batch + with enough visibility time for validation and republishing. +2. Decode the pull API's JSON message encoding as documented. Validate each event + with `isDownloadEvent` in the trusted module. Preserve `event_id`, + `observed_at`, and every original field. Do not log the message or credentials. + Investigate invalid messages instead of weakening validation or auto-looping + them through the primary queue. +3. Republish each valid original JSON event to `maple-download-events` using JSON + content type. Wait for the primary queue's successful enqueue acknowledgement. + Only then acknowledge that DLQ lease. On ambiguous enqueue/ack, retry with the + same event ID; the committed D1 row deduplicates it. +4. Check D1 persistence and both backlogs, then remove the temporary DLQ pull + consumer through its owning configuration. Do not purge queues, recreate the + database, change old timestamps, or generate new IDs to make replay pass. + +The query token need not have queue-mutation permission. Use the existing +separately authorized operations credential for replay; never broaden reporting +access merely to drain a DLQ. + +## Local verification + +```sh +nix develop --no-update-lock-file .#ci -c python3 -m unittest discover -s services/download-tracking -p 'test_*.py' +``` + +Focused TypeScript tests cover response preservation, route privacy, queue/SQL +outages, malformed events and real SQLite deduplication. The Agent service's +runtime test exercises actual local workerd request → Queue → D1 delivery, +duplicate events and a missing-table outage. These are separate from production +configuration, deployment and live collection verification. + +References: [D1 query API and accepted permissions](https://developers.cloudflare.com/api/resources/d1/subresources/database/methods/query/), +[Queue delivery guarantees](https://developers.cloudflare.com/queues/reference/delivery-guarantees/), +[configurable retention](https://developers.cloudflare.com/changelog/post/2025-02-14-customize-queue-retention-period/), +[dead-letter queues](https://developers.cloudflare.com/queues/configuration/dead-letter-queues/), +[authenticated pull and acknowledgement](https://developers.cloudflare.com/queues/configuration/pull-consumers/). diff --git a/services/download-tracking/query.py b/services/download-tracking/query.py new file mode 100644 index 000000000..a0e1e7c84 --- /dev/null +++ b/services/download-tracking/query.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +"""Private, read-only D1 reports. Credentials come only from the environment.""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import json +import os +from pathlib import Path +import re +import sys +from typing import Callable, Mapping +from urllib.error import HTTPError, URLError +from urllib.request import HTTPRedirectHandler, Request, build_opener + +MAX_ROWS = 10_000 +MAX_REPLY_BYTES = 8 * 1024 * 1024 +CONFIG = Path(__file__).resolve().parents[1] / "agent-updates" / "wrangler.jsonc" +UTC = timezone.utc +ARTIFACTS = ( + "macos-dmg", "macos-updater", "windows-installer", "linux-bundle", + "linux-appimage", "linux-deb", "linux-rpm", "android-apk", "latest", "manifest", +) +FILTERS = { + "product": ("agent", "research"), + "platform": ("macos", "windows", "linux", "android", "all"), + "artifact": ARTIFACTS, + "kind": ("artifact", "alias", "metadata"), + "method": ("GET", "HEAD"), + "range": ("none", "requested", "partial", "unsatisfiable"), + "host": ("updates.maple.ai", "updates.trymaple.ai"), +} +DIMENSIONS = ("product", "version", "platform", "artifact", "kind", "status", "method", "range", "is_probe") + + +class QueryError(Exception): + def __init__(self, code: int, http_status: int | None = None): + self.code = code + self.http_status = http_status + super().__init__(code) + + +class Parser(argparse.ArgumentParser): + def error(self, _message: str) -> None: + # Never echo invalid arguments, credentials, URLs, or upstream bodies. + raise QueryError(2) + + +def timestamp(value: str) -> datetime: + if not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z", value): + raise QueryError(11) + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + raise QueryError(11) from None + + +def iso(value: datetime) -> str: + return value.astimezone(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = Parser(description=__doc__) + parser.add_argument("--since", help="inclusive UTC timestamp, e.g. 2026-10-27T00:00:00Z") + parser.add_argument("--until", help="exclusive UTC timestamp; defaults to now") + parser.add_argument("--bucket", choices=("hour", "day"), default="day") + parser.add_argument("--include-probes", action="store_true", help="include self-reported probe requests") + parser.add_argument("--health", action="store_true", help="latest persisted observation and UTC-day count within the selected window") + for column, choices in FILTERS.items(): + parser.add_argument("--" + column, choices=choices) + parser.add_argument("--version", help="exact stable SemVer") + parser.add_argument("--status", type=int, help="exact HTTP status, 200 through 599") + args = parser.parse_args(argv) + if args.version is not None and ( + len(args.version) > 64 or not re.fullmatch(r"(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)", args.version) + ): + raise QueryError(12) + if args.status is not None and not 200 <= args.status <= 599: + raise QueryError(12) + return args + + +def query_spec(args: argparse.Namespace, now: datetime) -> tuple[str, list[object], dict[str, object]]: + now = now.astimezone(UTC).replace(microsecond=(now.microsecond // 1000) * 1000) + until = timestamp(args.until) if args.until else now.astimezone(UTC) + try: + since = timestamp(args.since) if args.since else until - timedelta(days=7) + except OverflowError: + raise QueryError(11) from None + if not timedelta(0) < until - since <= timedelta(days=31): + raise QueryError(11) + start, end = iso(since), iso(until) + clauses = ["observed_at >= ?", "observed_at < ?"] + params: list[object] = [start, end] + for column in (*FILTERS, "version", "status"): + value = getattr(args, column) + if value is not None: + clauses.append(f"{column} = ?") + params.append(value) + if not args.include_probes: + clauses.append("is_probe = 0") + where = " AND ".join(clauses) + summary: dict[str, object] = { + "since": start, "until": end, "bucket": args.bucket, + "include_probes": args.include_probes, + } + if args.health: + day_start = now.astimezone(UTC).replace(hour=0, minute=0, second=0, microsecond=0) + sql = ( + "SELECT MAX(observed_at) AS latest_observed_at, COUNT(*) AS persisted_events, " + "COALESCE(SUM(CASE WHEN observed_at >= ? THEN 1 ELSE 0 END), 0) AS utc_day_events " + f"FROM download_events WHERE {where}" + ) + params.insert(0, iso(day_start)) + summary["utc_day_since"] = iso(day_start) + else: + # Only this fixed choice enters SQL; every user filter is a bound value. + bucket = "substr(observed_at, 1, 13) || ':00:00.000Z'" if args.bucket == "hour" else "substr(observed_at, 1, 10) || 'T00:00:00.000Z'" + dimensions = ", ".join(DIMENSIONS) + sql = ( + f"SELECT {bucket} AS bucket, {dimensions}, COUNT(*) AS events, " + "COALESCE(SUM(response_length), 0) AS expected_response_bytes " + f"FROM download_events WHERE {where} GROUP BY bucket, {dimensions} " + f"ORDER BY bucket, {dimensions} LIMIT {MAX_ROWS + 1}" + ) + return sql, params, summary + + +def database_from_config(path: Path = CONFIG) -> str: + try: + # Wrangler JSONC: preserve strings while removing comments and trailing commas. + text = path.read_text(encoding="utf-8") + tokens = r'"(?:\\.|[^"\\])*"|//[^\n]*|/\*[\s\S]*?\*/' + text = re.sub(tokens, lambda m: m[0] if m[0].startswith('"') else "", text) + text = re.sub(r'"(?:\\.|[^"\\])*"|,\s*(?=[}\]])', lambda m: m[0] if m[0].startswith('"') else "", text) + config = json.loads(text) + matches = [item["database_id"] for item in config["d1_databases"] if item.get("binding") == "DOWNLOAD_DB"] + if len(matches) != 1: + raise QueryError(10) + return matches[0] + except (OSError, ValueError, KeyError, TypeError, AttributeError): + raise QueryError(10) from None + + +def credentials(environ: Mapping[str, str]) -> tuple[str, str, str]: + account = environ.get("CLOUDFLARE_ACCOUNT_ID", "") + database = environ.get("DOWNLOAD_ANALYTICS_DATABASE_ID") or database_from_config() + token = environ.get("CLOUDFLARE_API_TOKEN", "") + if ( + not re.fullmatch(r"[0-9a-f]{32}", account) + or not isinstance(database, str) + or not re.fullmatch(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", database) + or not 1 <= len(token) <= 4096 + or not all(33 <= ord(char) <= 126 for char in token) + ): + raise QueryError(10) + return account, database, token + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def transport(request: Request) -> bytes: + with build_opener(NoRedirect).open(request, timeout=30) as response: + data = response.read(MAX_REPLY_BYTES + 1) + if len(data) > MAX_REPLY_BYTES: + raise QueryError(23) + return data + + +def execute( + sql: str, + params: list[object], + environ: Mapping[str, str], + send: Callable[[Request], bytes] = transport, +) -> list[dict[str, object]]: + account, database, token = credentials(environ) + request = Request( + f"https://api.cloudflare.com/client/v4/accounts/{account}/d1/database/{database}/query", + data=json.dumps({"sql": sql, "params": params}, separators=(",", ":")).encode(), + headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"}, + method="POST", + ) + try: + data = send(request) + if len(data) > MAX_REPLY_BYTES: + raise QueryError(23) + payload = json.loads(data) + if not isinstance(payload, dict) or payload.get("success") is not True: + raise QueryError(21) + result = payload.get("result") + if not isinstance(result, list) or len(result) != 1 or not isinstance(result[0], dict): + raise QueryError(22) + if result[0].get("success") is not True: + raise QueryError(21) + rows = result[0].get("results") + if not isinstance(rows, list) or len(rows) > MAX_ROWS + 1 or not all(isinstance(row, dict) for row in rows): + raise QueryError(22) + return rows + except HTTPError as error: + # The upstream body/message/URL can be sensitive. Do not read or print it. + error.close() + raise QueryError(20, error.code) from None + except (URLError, TimeoutError, OSError): + raise QueryError(24) from None + except (ValueError, TypeError, UnicodeError): + raise QueryError(22) from None + + +def report( + args: argparse.Namespace, + environ: Mapping[str, str], + now: datetime, + send: Callable[[Request], bytes] = transport, +) -> dict[str, object]: + sql, params, summary = query_spec(args, now) + rows = execute(sql, params, environ, send) + expected_keys = {"latest_observed_at", "persisted_events", "utc_day_events"} if args.health else {"bucket", *DIMENSIONS, "events", "expected_response_bytes"} + for row in rows: + if set(row) != expected_keys or any( + not (value is None or (type(value) is int and value >= 0) or (type(value) is str and len(value) <= 64)) + for value in row.values() + ): + raise QueryError(22) + summary["truncated"] = len(rows) > MAX_ROWS + summary["rows"] = rows[:MAX_ROWS] + return summary + + +def main(argv: list[str] | None = None, environ: Mapping[str, str] | None = None) -> int: + try: + result = report(parse_args(argv), os.environ if environ is None else environ, datetime.now(UTC)) + print(json.dumps(result, separators=(",", ":"))) + return 0 + except QueryError as error: + failure = {"error_code": error.code} + if error.http_status is not None: + failure["http_status"] = error.http_status + print(json.dumps(failure, separators=(",", ":")), file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/download-tracking/test_query.py b/services/download-tracking/test_query.py new file mode 100644 index 000000000..c366436a5 --- /dev/null +++ b/services/download-tracking/test_query.py @@ -0,0 +1,180 @@ +"""Credential-free tests of query validation, SQL, bounds and error redaction.""" + +import contextlib +from datetime import datetime, timedelta +import io +import json +from pathlib import Path +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch +from urllib.error import HTTPError, URLError + +import query + + +NOW = datetime(2026, 10, 27, 12, 34, 56, tzinfo=query.UTC) +ENV = { + "CLOUDFLARE_ACCOUNT_ID": "a" * 32, + "DOWNLOAD_ANALYTICS_DATABASE_ID": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + "CLOUDFLARE_API_TOKEN": "fixture-private-token-never-print", +} + + +def reply(rows): + return json.dumps({"success": True, "result": [{"success": True, "results": rows}]}).encode() + + +def row(): + return {"bucket": "2026-10-27T00:00:00.000Z", "product": "agent", "version": "0.0.9", + "platform": "macos", "artifact": "macos-dmg", "kind": "artifact", "status": 200, + "method": "GET", "range": "none", "is_probe": 0, "events": 2, "expected_response_bytes": 42} + + +class QueryTests(unittest.TestCase): + def test_utc_time_validation_and_window_bounds(self): + for value in ("2026-10-27", "2026-10-27T12:00:00+00:00", "2026-10-27T12:00:00", + "2026-02-30T00:00:00Z", "2026-10-27T12:00:00.1Z", "2026-10-27T12:00:60Z"): + with self.subTest(value=value), self.assertRaises(query.QueryError): + query.timestamp(value) + self.assertEqual(query.iso(query.timestamp("2026-10-27T12:00:00Z")), "2026-10-27T12:00:00.000Z") + self.assertEqual(query.iso(query.timestamp("2026-10-27T12:00:00.123Z")), "2026-10-27T12:00:00.123Z") + for argv in ( + ["--since", "2026-10-27T12:34:56Z"], + ["--since", "2026-10-28T00:00:00Z"], + ["--since", "2026-09-25T00:00:00Z"], + ): + with self.subTest(argv=argv), self.assertRaises(query.QueryError): + query.query_spec(query.parse_args(argv), NOW) + _, params, summary = query.query_spec(query.parse_args([]), NOW) + self.assertEqual(params[:2], ["2026-10-20T12:34:56.000Z", "2026-10-27T12:34:56.000Z"]) + self.assertFalse(summary["include_probes"]) + + def test_queries_are_fixed_selects_with_only_bound_user_filters(self): + args = query.parse_args(["--product", "agent", "--version", "0.0.9", "--platform", "linux", + "--artifact", "linux-bundle", "--kind", "artifact", "--method", "GET", + "--range", "partial", "--status", "206", "--host", "updates.maple.ai"]) + sql, params, _ = query.query_spec(args, NOW) + self.assertTrue(sql.startswith("SELECT ")) + self.assertNotIn(";", sql) + self.assertIn("observed_at >= ? AND observed_at < ?", sql) + self.assertIn("is_probe = 0", sql) + self.assertIn("GROUP BY bucket, product, version, platform, artifact, kind, status, method, range, is_probe", sql) + self.assertTrue(sql.endswith("LIMIT 10001")) + self.assertEqual(sql.count("?"), len(params)) + self.assertIn("0.0.9", params) + self.assertNotIn("0.0.9", sql) + self.assertNotIn("updates.maple.ai", sql) + for bad in (["--product", "private;DELETE"], ["--version", "1.2.3';DELETE"], + ["--bucket", "day;DROP"], ["--status", "199"], ["--status", "600"], + ["--version", "01.2.3"], ["--unexpected", "private-token"]): + with self.subTest(bad=bad), self.assertRaises(query.QueryError): + query.parse_args(bad) + health_sql, _, _ = query.query_spec(query.parse_args(["--health", "--include-probes"]), NOW) + self.assertTrue(health_sql.startswith("SELECT ")) + self.assertNotIn("is_probe = 0", health_sql) + + def test_exact_readonly_sql_executes_on_migration_and_separates_kinds_probes_ranges(self): + db = sqlite3.connect(":memory:") + db.row_factory = sqlite3.Row + db.executescript((Path(__file__).parents[1] / "agent-updates/migrations/0001_download_events.sql").read_text()) + base = ["2026-10-27T12:00:00.000Z", "agent", "updates.maple.ai", "0.0.9", "macos", "macos-dmg", "artifact", "GET", 200, "none", "success", 42, "cli", 0] + for index, changes in enumerate(({}, {7: "HEAD"}, {8: 206, 9: "partial"}, + {6: "alias", 8: 302, 10: "redirect"}, {13: 1})): + values = base.copy() + for position, value in changes.items(): + values[position] = value + db.execute("INSERT INTO download_events VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", [str(index), *values]) + sql, params, _ = query.query_spec(query.parse_args([]), NOW) + rows = [dict(item) for item in db.execute(sql, params)] + self.assertEqual(sum(item["events"] for item in rows), 4) + self.assertEqual(len(rows), 4) + self.assertEqual({item["kind"] for item in rows}, {"alias", "artifact"}) + hour_sql, hour_params, _ = query.query_spec(query.parse_args(["--bucket", "hour", "--include-probes"]), NOW) + hour_rows = list(db.execute(hour_sql, hour_params)) + self.assertEqual({item["bucket"] for item in hour_rows}, {"2026-10-27T12:00:00.000Z"}) + self.assertEqual(sum(item["events"] for item in hour_rows), 5) + health_sql, health_params, _ = query.query_spec(query.parse_args(["--health"]), NOW) + health = dict(db.execute(health_sql, health_params).fetchone()) + self.assertEqual(health, {"latest_observed_at": base[0], "persisted_events": 4, "utc_day_events": 4}) + db.close() + + def test_environment_only_credentials_and_fixed_private_endpoint(self): + calls = [] + def send(request): + calls.append(request) + return reply([row()]) + result = query.report(query.parse_args([]), ENV, NOW, send) + request = calls[0] + self.assertEqual(request.get_method(), "POST") + self.assertEqual(request.full_url, f"https://api.cloudflare.com/client/v4/accounts/{'a' * 32}/d1/database/{ENV['DOWNLOAD_ANALYTICS_DATABASE_ID']}/query") + self.assertEqual(request.get_header("Authorization"), "Bearer " + ENV["CLOUDFLARE_API_TOKEN"]) + body = json.loads(request.data) + self.assertTrue(body["sql"].startswith("SELECT ")) + self.assertNotIn(ENV["CLOUDFLARE_API_TOKEN"], json.dumps(result)) + self.assertFalse(result["truncated"]) + self.assertEqual(result["rows"], [row()]) + for key in ENV: + bad = {**ENV, key: "private-invalid-value\n"} + with self.subTest(key=key), self.assertRaises(query.QueryError): + query.credentials(bad) + + def test_database_defaults_to_exact_config_binding_with_jsonc(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "config.jsonc" + path.write_text('''{ // private config comments not emitted + "$schema": "https://example.test/schema,}", + "d1_databases": [{"binding":"OTHER","database_id":"ignored"}, + {"binding":"DOWNLOAD_DB","database_id":"bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb",},], + }''') + self.assertEqual(query.database_from_config(path), ENV["DOWNLOAD_ANALYTICS_DATABASE_ID"]) + with patch.object(query, "database_from_config", return_value=ENV["DOWNLOAD_ANALYTICS_DATABASE_ID"]): + credentials = query.credentials({key: value for key, value in ENV.items() if key != "DOWNLOAD_ANALYTICS_DATABASE_ID"}) + self.assertEqual(credentials[1], ENV["DOWNLOAD_ANALYTICS_DATABASE_ID"]) + path.write_text('{"d1_databases": []}') + with self.assertRaises(query.QueryError): + query.database_from_config(path) + + def test_output_cap_is_explicit_not_silent(self): + report = query.report(query.parse_args([]), ENV, NOW, lambda _: reply([row()] * 10001)) + self.assertTrue(report["truncated"]) + self.assertEqual(len(report["rows"]), 10000) + + def test_upstream_errors_malformed_responses_and_large_bodies_are_sanitized(self): + def http_error(_): + raise HTTPError("https://private.example/secret", 403, "private token", {}, io.BytesIO(b"secret private body")) + def network_error(_): + raise URLError("private token URL") + cases = [(http_error, 20), (network_error, 24), (lambda _: b"private token", 22), + (lambda _: b"x" * (query.MAX_REPLY_BYTES + 1), 23), + (lambda _: json.dumps({"success": False, "errors": ["private token"]}).encode(), 21), + (lambda _: reply([{**row(), "private": "token"}]), 22)] + for send, expected in cases: + with self.subTest(expected=expected), self.assertRaises(query.QueryError) as caught: + query.report(query.parse_args([]), ENV, NOW, send) + self.assertEqual(caught.exception.code, expected) + self.assertNotIn("private", str(caught.exception)) + + def test_main_error_output_contains_numeric_codes_only(self): + out, err = io.StringIO(), io.StringIO() + with patch.object(query, "report", side_effect=query.QueryError(20, 403)): + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = query.main([], ENV) + self.assertEqual(code, 1) + self.assertEqual(out.getvalue(), "") + self.assertEqual(json.loads(err.getvalue()), {"error_code": 20, "http_status": 403}) + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = query.main(["--product", ENV["CLOUDFLARE_API_TOKEN"]], ENV) + self.assertEqual(code, 1) + self.assertEqual(json.loads(err.getvalue()), {"error_code": 2}) + self.assertNotIn(ENV["CLOUDFLARE_API_TOKEN"], err.getvalue()) + + def test_private_api_transport_rejects_redirects(self): + handler = query.NoRedirect() + self.assertIsNone(handler.redirect_request(None, None, 302, "private", {}, "https://private.example")) + + +if __name__ == "__main__": + unittest.main() diff --git a/services/download-tracking/tracking.ts b/services/download-tracking/tracking.ts new file mode 100644 index 000000000..d32f328e2 --- /dev/null +++ b/services/download-tracking/tracking.ts @@ -0,0 +1,491 @@ +import repositoryMetadata from "../../repo.meta.json"; + +export type Product = "agent" | "research"; +type Platform = "macos" | "windows" | "linux" | "android" | "all"; +type Artifact = + | "macos-dmg" + | "macos-updater" + | "windows-installer" + | "linux-bundle" + | "linux-appimage" + | "linux-deb" + | "linux-rpm" + | "android-apk" + | "latest" + | "manifest"; +type Kind = "artifact" | "alias" | "metadata"; +type Range = "none" | "requested" | "partial" | "unsatisfiable"; +type Result = + "success" | "redirect" | "not_modified" | "client_error" | "server_error"; +type ClientKind = "browser" | "maple" | "cli" | "other"; +const HOSTS = ["updates.maple.ai", "updates.trymaple.ai"] as const; +const VERSION = /^(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)$/; +const MAX_RESPONSE_LENGTH = 4 * 1024 ** 3; +const RESEARCH_RELEASE = `/${repositoryMetadata.github.owner}/${repositoryMetadata.github.repository}/releases/download/`; + +/** A response observation, not a user identity or proof of transfer/install. */ +export interface DownloadEvent { + schema_version: 1; + event_id: string; + observed_at: string; + product: Product; + host: (typeof HOSTS)[number]; + version: string | null; + platform: Platform; + artifact: Artifact; + kind: Kind; + method: "GET" | "HEAD"; + status: number; + range: Range; + result: Result; + response_length: number | null; + client_kind: ClientKind; + is_probe: 0 | 1; +} + +// Structural interfaces keep this module shared by both existing Workers. +export interface EventQueue { + send(event: DownloadEvent): Promise; +} +export interface TrackingEnv { + DOWNLOAD_EVENTS?: EventQueue; +} +export interface TrackingContext { + waitUntil(promise: Promise): void; +} +export interface EventMessage { + body: unknown; + ack(): void; + retry(): void; +} +export interface EventBatch { + messages: readonly EventMessage[]; +} +export interface EventDatabase { + prepare(sql: string): { + bind(...values: (string | number | null)[]): { + run(): Promise<{ success: boolean }>; + }; + }; +} + +interface Route { + kind: Kind; + version: string | null; + platform: Platform; + artifact: Artifact; +} + +function stableVersion(value: string): boolean { + return value.length <= 64 && VERSION.exec(value)?.[0] === value; +} + +function agentArtifact(path: string): Route | null { + const match = + /^\/agent\/stable\/releases\/([^/]+)\/([A-Za-z0-9][A-Za-z0-9._-]*)$/.exec( + path, + ); + if (!match || !stableVersion(match[1])) return null; + const file = match[2]; + if (file.length > 200 || file.includes("..")) return null; + const base = { version: match[1], kind: "artifact" as const }; + if (file.endsWith(".app.tar.gz")) + return { ...base, platform: "macos", artifact: "macos-updater" }; + if (file.endsWith(".dmg")) + return { ...base, platform: "macos", artifact: "macos-dmg" }; + if (file.endsWith(".AppImage")) + return { ...base, platform: "linux", artifact: "linux-appimage" }; + if (file.endsWith(".tar.gz")) + return { ...base, platform: "linux", artifact: "linux-bundle" }; + return null; +} + +const AGENT_ALIASES: Record> = { + macos: { platform: "macos", artifact: "macos-dmg" }, + linux: { platform: "linux", artifact: "linux-bundle" }, + "linux-appimage": { platform: "linux", artifact: "linux-appimage" }, +}; +const RESEARCH_ALIASES: Record> = { + macos: { platform: "macos", artifact: "macos-dmg" }, + windows: { platform: "windows", artifact: "windows-installer" }, + "linux-appimage": { platform: "linux", artifact: "linux-appimage" }, + "linux-deb": { platform: "linux", artifact: "linux-deb" }, + "linux-rpm": { platform: "linux", artifact: "linux-rpm" }, + android: { platform: "android", artifact: "android-apk" }, +}; + +function researchVersion(location: URL, artifact: Artifact): string | null { + if ( + location.origin !== "https://github.com" || + !location.pathname.startsWith(RESEARCH_RELEASE) + ) + return null; + const match = /^v([^/]+)\/([^/]+)$/.exec( + location.pathname.slice(RESEARCH_RELEASE.length), + ); + if (!match || !stableVersion(match[1])) return null; + const [version, file] = match.slice(1); + const names: Partial> = { + "macos-dmg": `Maple_${version}_universal.dmg`, + "windows-installer": `Maple_${version}_x64-setup.exe`, + "linux-appimage": `Maple_${version}_amd64.AppImage`, + "linux-deb": `Maple_${version}_amd64.deb`, + "android-apk": "app-universal-release.apk", + }; + const matches = + artifact === "linux-rpm" + ? new RegExp( + `^Maple-${version.replaceAll(".", "\\.")}-[1-9][0-9]*\\.x86_64\\.rpm$`, + ).exec(file)?.[0] === file + : names[artifact] === file; + return matches ? version : null; +} + +function routeFor( + url: URL, + response: Response, + product: Product, +): Route | null { + const path = url.pathname; + // Research accepts marketing queries; classify only its pathname and never + // retain query values. Agent delivery itself rejects queries. + if ( + (product === "agent" && url.search) || + url.hash || + path.includes("%") || + path.includes("\\") || + path.includes("//") + ) + return null; + if (product === "research" && path === "/latest.json") + return { + kind: "metadata", + version: null, + platform: "all", + artifact: "latest", + }; + if (product === "agent") { + if ( + path === "/agent/stable/latest.json" || + path === "/agent/stable/linux/latest.json" + ) + return { + kind: "metadata", + version: null, + platform: path.includes("/linux/") ? "linux" : "macos", + artifact: "latest", + }; + const manifest = + /^\/agent\/stable\/releases\/([^/]+)\/(linux-manifest|manifest)\.json$/.exec( + path, + ); + if (manifest && stableVersion(manifest[1])) + return { + kind: "metadata", + version: manifest[1], + platform: manifest[2] === "linux-manifest" ? "linux" : "macos", + artifact: "manifest", + }; + const artifact = agentArtifact(path); + if (artifact) return artifact; + } + const prefix = `/download/${product}/stable/`; + if (!path.startsWith(prefix)) return null; + const aliases = product === "agent" ? AGENT_ALIASES : RESEARCH_ALIASES; + const alias = aliases[path.slice(prefix.length)]; + if (!Object.hasOwn(aliases, path.slice(prefix.length)) || !alias) return null; + let version: string | null = null; + const target = response.headers.get("location"); + if (response.status === 302 && target && target.length <= 512) { + const location = new URL(target, url); + if ( + !location.search && + !location.hash && + !location.username && + !location.password + ) { + if (product === "agent" && location.origin === url.origin) { + const artifact = agentArtifact(location.pathname); + if (artifact?.artifact === alias.artifact) version = artifact.version; + } else if (product === "research") { + version = researchVersion(location, alias.artifact); + } + } + } + return { kind: "alias", version, ...alias }; +} + +function clientKind(request: Request): ClientKind { + const ua = request.headers.get("user-agent")?.slice(0, 512) ?? ""; + if (/maple.*updat/i.test(ua)) return "maple"; + if (/Mozilla\/5\.0/i.test(ua)) return "browser"; + if (/curl|wget|python-urllib|Go-http-client/i.test(ua)) return "cli"; + return "other"; +} + +function responseResult(status: number): Result { + if (status === 304) return "not_modified"; + if (status >= 500) return "server_error"; + if (status >= 400) return "client_error"; + if (status >= 300) return "redirect"; + return "success"; +} + +function logFailure(code: string): void { + // Deliberately never log the exception, request, URL, headers or event body. + console.error(JSON.stringify({ service: "download-tracking", code })); +} + +async function enqueue(queue: EventQueue, event: DownloadEvent): Promise { + // An ambiguous send can have succeeded: keep the ID for D1 deduplication. + for (let attempt = 0; attempt < 2; attempt++) { + try { + await queue.send(event); + return; + } catch { + logFailure("queue_send_failed"); + } + } +} + +/** Best-effort, unsampled response-start capture. Never waits for/reads a body. */ +export function trackResponse( + request: Request, + response: Response, + env: TrackingEnv, + ctx: TrackingContext, + product: Product, +): void { + try { + if (!["GET", "HEAD"].includes(request.method)) return; + const url = new URL(request.url); + if ( + url.protocol !== "https:" || + !HOSTS.includes(url.host as (typeof HOSTS)[number]) || + url.username || + url.password || + response.status < 200 || + response.status > 599 + ) + return; + const route = routeFor(url, response, product); + if (!route) return; + const queue = env.DOWNLOAD_EVENTS; + if (!queue) { + logFailure("queue_binding_missing"); + return; + } + const length = response.headers.get("content-length"); + const parsedLength = + length && /^(0|[1-9][0-9]*)$/.test(length) ? Number(length) : null; + const event: DownloadEvent = { + schema_version: 1, + event_id: crypto.randomUUID(), + observed_at: new Date().toISOString(), + product, + host: url.host as DownloadEvent["host"], + ...route, + method: request.method as "GET" | "HEAD", + status: response.status, + range: + route.kind !== "artifact" + ? "none" + : response.status === 206 + ? "partial" + : response.status === 416 + ? "unsatisfiable" + : request.headers.has("range") + ? "requested" + : "none", + result: responseResult(response.status), + response_length: + parsedLength !== null && + Number.isSafeInteger(parsedLength) && + parsedLength <= MAX_RESPONSE_LENGTH + ? parsedLength + : null, + client_kind: clientKind(request), + is_probe: request.headers.get("x-maple-download-probe") === "1" ? 1 : 0, + }; + // Deferring send also catches synchronous binding failures without affecting + // delivery. waitUntil is bounded by the runtime; fail-open can leave gaps. + ctx.waitUntil(Promise.resolve().then(() => enqueue(queue, event))); + } catch { + logFailure("capture_failed"); + } +} + +const EVENT_KEYS = [ + "schema_version", + "event_id", + "observed_at", + "product", + "host", + "version", + "platform", + "artifact", + "kind", + "method", + "status", + "range", + "result", + "response_length", + "client_kind", + "is_probe", +]; + +/** Queue content is a separate trust boundary, even with a private producer. */ +export function isDownloadEvent(value: unknown): value is DownloadEvent { + if (!value || typeof value !== "object" || Array.isArray(value)) return false; + const event = value as Record; + if ( + Object.keys(event).length !== EVENT_KEYS.length || + !EVENT_KEYS.every((key) => Object.hasOwn(event, key)) || + event.schema_version !== 1 || + typeof event.event_id !== "string" || + !/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test( + event.event_id, + ) || + typeof event.observed_at !== "string" || + !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/.test(event.observed_at) || + !Number.isFinite(Date.parse(event.observed_at)) || + new Date(event.observed_at).toISOString() !== event.observed_at || + !["agent", "research"].includes(event.product as string) || + !HOSTS.includes(event.host as (typeof HOSTS)[number]) || + !( + event.version === null || + (typeof event.version === "string" && stableVersion(event.version)) + ) || + !["macos", "windows", "linux", "android", "all"].includes( + event.platform as string, + ) || + ![ + "macos-dmg", + "macos-updater", + "windows-installer", + "linux-bundle", + "linux-appimage", + "linux-deb", + "linux-rpm", + "android-apk", + "latest", + "manifest", + ].includes(event.artifact as string) || + !["artifact", "alias", "metadata"].includes(event.kind as string) || + !["GET", "HEAD"].includes(event.method as string) || + typeof event.status !== "number" || + !Number.isInteger(event.status) || + event.status < 200 || + event.status > 599 || + !["none", "requested", "partial", "unsatisfiable"].includes( + event.range as string, + ) || + event.result !== responseResult(event.status) || + !( + event.response_length === null || + (typeof event.response_length === "number" && + Number.isSafeInteger(event.response_length) && + event.response_length >= 0 && + event.response_length <= MAX_RESPONSE_LENGTH) + ) || + !["browser", "maple", "cli", "other"].includes( + event.client_kind as string, + ) || + ![0, 1].includes(event.is_probe as number) + ) + return false; + // The ledger accepts only combinations produced by our route classifier. + const artifactPlatform: Record = { + "macos-dmg": ["macos"], + "macos-updater": ["macos"], + "windows-installer": ["windows"], + "linux-bundle": ["linux"], + "linux-appimage": ["linux"], + "linux-deb": ["linux"], + "linux-rpm": ["linux"], + "android-apk": ["android"], + latest: ["all", "macos", "linux"], + manifest: ["macos", "linux"], + }; + if ( + !artifactPlatform[event.artifact as Artifact].includes( + event.platform as Platform, + ) + ) + return false; + if (event.kind === "metadata") { + if ( + !["latest", "manifest"].includes(event.artifact as string) || + event.range !== "none" + ) + return false; + if (event.artifact === "latest" && event.version !== null) return false; + if (event.artifact === "manifest" && event.version === null) return false; + return event.product === "research" + ? event.artifact === "latest" && event.platform === "all" + : event.platform !== "all"; + } + if (["latest", "manifest"].includes(event.artifact as string)) return false; + if (event.kind === "alias" && event.range !== "none") return false; + if (event.product === "research") + return ( + event.kind === "alias" && + !["macos-updater", "linux-bundle"].includes(event.artifact as string) + ); + if ( + ["windows-installer", "linux-deb", "linux-rpm", "android-apk"].includes( + event.artifact as string, + ) + ) + return false; + return event.kind === "artifact" + ? event.version !== null + : event.artifact !== "macos-updater"; +} + +const INSERT = `INSERT INTO download_events + (event_id, observed_at, product, host, version, platform, artifact, kind, + method, status, range, result, response_length, client_kind, is_probe) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(event_id) DO NOTHING`; + +/** Retries/DLQ are configured on the Queue; never acknowledge lost writes. */ +export async function consumeEvents( + batch: EventBatch, + env: { DOWNLOAD_DB: EventDatabase }, +): Promise { + for (const message of batch.messages) { + if (!isDownloadEvent(message.body)) { + logFailure("invalid_event"); + message.retry(); + continue; + } + const event = message.body; + try { + const result = await env.DOWNLOAD_DB.prepare(INSERT) + .bind( + event.event_id, + event.observed_at, + event.product, + event.host, + event.version, + event.platform, + event.artifact, + event.kind, + event.method, + event.status, + event.range, + event.result, + event.response_length, + event.client_kind, + event.is_probe, + ) + .run(); + if (!result.success) throw new Error("write failed"); + message.ack(); + } catch { + logFailure("database_write_failed"); + message.retry(); + } + } +} diff --git a/services/updates/README.md b/services/updates/README.md index c87a71ee6..3b27b4c0b 100644 --- a/services/updates/README.md +++ b/services/updates/README.md @@ -17,7 +17,11 @@ Both route families accept `GET` and `HEAD`; other methods return `405`. Unknown paths, including the private `/installers.json` bundle file, return `404`. The installer routes issue temporary `302` redirects to exact, versioned GitHub release assets with browser and CDN `no-store` headers. They require no browser -JavaScript, API lookup, CORS configuration, R2, database, or visitor credentials. +JavaScript, API lookup, CORS configuration, or R2. +Optional private request metrics use the shared Queue binding described in +[download tracking](../download-tracking/README.md). The database is not part of +serving a response and visitors need no credentials. + GitHub still hosts the installer bytes; these redirects do not remove that availability dependency. diff --git a/services/updates/package.json b/services/updates/package.json index b18f69fed..bcda10e10 100644 --- a/services/updates/package.json +++ b/services/updates/package.json @@ -8,8 +8,8 @@ "build:dry-run": "wrangler deploy --dry-run --outdir .wrangler/dry-run", "deploy": "wrangler deploy", "dev": "wrangler dev", - "format": "prettier --write \"src/**/*.ts\" README.md package.json tsconfig.json wrangler.jsonc", - "format:check": "prettier --check \"src/**/*.ts\" README.md package.json tsconfig.json wrangler.jsonc", + "format": "prettier --write \"src/**/*.ts\" ../download-tracking/tracking.ts ../download-tracking/README.md README.md package.json tsconfig.json wrangler.jsonc", + "format:check": "prettier --check \"src/**/*.ts\" ../download-tracking/tracking.ts ../download-tracking/README.md README.md package.json tsconfig.json wrangler.jsonc", "test": "bun --no-env-file test", "typecheck": "tsc --noEmit", "validate:latest": "bun src/validate-latest.ts", diff --git a/services/updates/src/worker.ts b/services/updates/src/worker.ts index 64c6b2e97..8c6dd09e2 100644 --- a/services/updates/src/worker.ts +++ b/services/updates/src/worker.ts @@ -1,3 +1,8 @@ +import { + trackResponse, + type TrackingEnv, + type TrackingContext, +} from "../../download-tracking/tracking"; import repositoryMetadata from "../../../repo.meta.json"; import { INSTALLER_KEYS, @@ -27,7 +32,7 @@ interface AssetFetcher { fetch(request: Request): Promise; } -export interface Env { +export interface Env extends TrackingEnv { ASSETS: AssetFetcher; } @@ -296,5 +301,13 @@ export async function handleRequest( } export default { - fetch: handleRequest, + async fetch( + request: Request, + env: Env, + ctx?: TrackingContext, + ): Promise { + const result = await handleRequest(request, env); + if (ctx) trackResponse(request, result, env, ctx, "research"); + return result; + }, }; diff --git a/services/updates/wrangler.jsonc b/services/updates/wrangler.jsonc index 51c6d2505..004659234 100644 --- a/services/updates/wrangler.jsonc +++ b/services/updates/wrangler.jsonc @@ -10,4 +10,16 @@ "binding": "ASSETS", "run_worker_first": true, }, + "queues": { + "producers": [ + { "binding": "DOWNLOAD_EVENTS", "queue": "maple-download-events" }, + ], + }, + "observability": { + "logs": { + "enabled": true, + "head_sampling_rate": 1, + "invocation_logs": false, + }, + }, }