From bbd78a4e88046a08a4f0e99bce19a6a15a387884 Mon Sep 17 00:00:00 2001 From: Yaacov Date: Mon, 20 Jul 2026 19:02:25 +0300 Subject: [PATCH] Add first-party analytics gateway --- README.md | 23 +- functions/_lib/events.ts | 45 +-- functions/api/download/[asset].test.ts | 39 +-- functions/api/events.test.ts | 30 +- migrations/0002_analytics_gateway.sql | 25 ++ package.json | 6 +- scripts/analytics-report.mjs | 64 ++++- src/pages/privacy.astro | 3 + tsconfig.json | 2 +- workers/events/src/index.test.ts | 169 +++++++++++ workers/events/src/index.ts | 344 +++++++++++++++++++++++ workers/events/tsconfig.json | 9 + workers/events/worker-configuration.d.ts | 12 + workers/events/wrangler.jsonc | 27 ++ 14 files changed, 743 insertions(+), 55 deletions(-) create mode 100644 migrations/0002_analytics_gateway.sql create mode 100644 workers/events/src/index.test.ts create mode 100644 workers/events/src/index.ts create mode 100644 workers/events/tsconfig.json create mode 100644 workers/events/worker-configuration.d.ts create mode 100644 workers/events/wrangler.jsonc diff --git a/README.md b/README.md index daa56b7..7d8a224 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for the contribution workflow. The cross- ## First-party event measurement -Cloudflare D1 stores four anonymous event types: +Cloudflare D1 stores four anonymous website event types: - `page_viewed` - `download_clicked` @@ -43,7 +43,7 @@ Cloudflare D1 stores four anonymous event types: The implementation intentionally stores no IP address, cookie, fingerprint, referrer, or persistent visitor identifier. Counts therefore represent events rather than unique visitors. `download_failed` is limited to a failure in ScientFactory's redirect service; the website cannot observe a transfer failure after GitHub begins serving an installer. -The production binding is `DOWNLOAD_DB`, backed by the `scientfactory-downloads` D1 database. Apply new migrations before deploying code that depends on them: +The production binding is `DOWNLOAD_DB`, backed by the `scientfactory-downloads` D1 database. New events use the shared `analytics_events` table; the earlier `site_events` table remains as read-only historical data. Apply new migrations before deploying code that depends on them: ```sh bun run db:migrate @@ -56,3 +56,22 @@ bun run analytics:report ``` Local and Cloudflare preview hosts do not write events, which keeps production counts free of development traffic. + +## Analytics gateway + +The Worker under `workers/events` is ScientFactory's first-party telemetry gateway. Desktop clients submit bounded event batches to `https://events.scientfactory.com/v1/events`; the Worker stores them in D1 first and can then forward anonymous copies to the ScientFactory EU PostHog project. PostHog is an optional analysis layer rather than the primary event store. + +Generate binding types and validate the Worker with: + +```sh +bun run events:types +bun run events:typecheck +``` + +Deploy the Worker only from an approved production change: + +```sh +bun run events:deploy +``` + +`POSTHOG_PROJECT_TOKEN` is a Cloudflare Worker secret and must never be committed. If it is absent, ingestion continues and events remain queued in D1 for later delivery. diff --git a/functions/_lib/events.ts b/functions/_lib/events.ts index b2e7013..745f4e4 100644 --- a/functions/_lib/events.ts +++ b/functions/_lib/events.ts @@ -30,17 +30,15 @@ export interface SiteEventContext { } const INSERT_EVENT = ` - INSERT INTO site_events ( + INSERT OR IGNORE INTO analytics_events ( + event_id, event_name, - page_path, - asset_key, - release_tag, - asset_name, - destination_host, - destination_path, - failure_stage, - failure_reason - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + source, + privacy_level, + occurred_at, + distinct_id, + properties_json + ) VALUES (?, ?, 'website', ?, ?, ?, ?) `; const PRODUCTION_HOSTS = new Set(["scientfactory.com", "www.scientfactory.com"]); @@ -82,18 +80,29 @@ export function shouldPersistSiteEvents(request: Request): boolean { } export async function insertSiteEvent(db: D1Database, event: SiteEvent): Promise { + const eventId = crypto.randomUUID(); + const properties = Object.fromEntries( + Object.entries({ + page_path: cleanPath(event.pagePath), + asset_key: limited(event.assetKey, 64), + release_tag: limited(event.releaseTag, 80), + asset_name: limited(event.assetName, 255), + destination_host: limited(event.destinationHost, 253), + destination_path: cleanPath(event.destinationPath), + failure_stage: limited(event.failureStage, 80), + failure_reason: limited(event.failureReason, 120), + }).filter((entry): entry is [string, string] => entry[1] !== null), + ); + await db .prepare(INSERT_EVENT) .bind( + eventId, event.eventName, - cleanPath(event.pagePath), - limited(event.assetKey, 64), - limited(event.releaseTag, 80), - limited(event.assetName, 255), - limited(event.destinationHost, 253), - cleanPath(event.destinationPath), - limited(event.failureStage, 80), - limited(event.failureReason, 120), + event.eventName === "download_failed" ? "diagnostic" : "product", + new Date().toISOString(), + `web-event:${eventId}`, + JSON.stringify(properties), ) .run(); } diff --git a/functions/api/download/[asset].test.ts b/functions/api/download/[asset].test.ts index 68de73a..8bb319b 100644 --- a/functions/api/download/[asset].test.ts +++ b/functions/api/download/[asset].test.ts @@ -59,15 +59,20 @@ describe("tracked download redirect", () => { expect(context.waitUntil).toHaveBeenCalledTimes(1); await context.waitUntil.mock.calls[0]?.[0]; expect(db.bind).toHaveBeenCalledWith( + expect.any(String), "download_clicked", - "/download", - "macArm64", - "v0.5.7", - "Scient-0.5.7-arm64.dmg", - "github.com", - "/ScientFactory/scient-desktop/releases/download/v0.5.7/Scient-0.5.7-arm64.dmg", - null, - null, + "product", + expect.any(String), + expect.stringMatching(/^web-event:/), + JSON.stringify({ + page_path: "/download", + asset_key: "macArm64", + release_tag: "v0.5.7", + asset_name: "Scient-0.5.7-arm64.dmg", + destination_host: "github.com", + destination_path: + "/ScientFactory/scient-desktop/releases/download/v0.5.7/Scient-0.5.7-arm64.dmg", + }), ); }); @@ -97,15 +102,17 @@ describe("tracked download redirect", () => { expect(response.status).toBe(503); await context.waitUntil.mock.calls[0]?.[0]; expect(db.bind).toHaveBeenCalledWith( + expect.any(String), "download_failed", - "/download", - "macArm64", - null, - null, - null, - null, - "release_fetch", - "upstream_unavailable", + "diagnostic", + expect.any(String), + expect.stringMatching(/^web-event:/), + JSON.stringify({ + page_path: "/download", + asset_key: "macArm64", + failure_stage: "release_fetch", + failure_reason: "upstream_unavailable", + }), ); }); diff --git a/functions/api/events.test.ts b/functions/api/events.test.ts index 1feed11..525af15 100644 --- a/functions/api/events.test.ts +++ b/functions/api/events.test.ts @@ -49,15 +49,12 @@ describe("browser event endpoint", () => { expect(response.status).toBe(204); await context.waitUntil.mock.calls[0]?.[0]; expect(database.bind).toHaveBeenCalledWith( + expect.any(String), "page_viewed", - "/docs", - null, - null, - null, - null, - null, - null, - null, + "product", + expect.any(String), + expect.stringMatching(/^web-event:/), + JSON.stringify({ page_path: "/docs" }), ); }); @@ -73,15 +70,16 @@ describe("browser event endpoint", () => { expect(response.status).toBe(204); await context.waitUntil.mock.calls[0]?.[0]; expect(database.bind).toHaveBeenCalledWith( + expect.any(String), "outbound_link_clicked", - "/about", - null, - null, - null, - "github.com", - "/ScientFactory/scient-desktop", - null, - null, + "product", + expect.any(String), + expect.stringMatching(/^web-event:/), + JSON.stringify({ + page_path: "/about", + destination_host: "github.com", + destination_path: "/ScientFactory/scient-desktop", + }), ); }); diff --git a/migrations/0002_analytics_gateway.sql b/migrations/0002_analytics_gateway.sql new file mode 100644 index 0000000..bc7326a --- /dev/null +++ b/migrations/0002_analytics_gateway.sql @@ -0,0 +1,25 @@ +CREATE TABLE analytics_events ( + event_id TEXT PRIMARY KEY, + event_name TEXT NOT NULL, + source TEXT NOT NULL CHECK (source IN ('website', 'desktop')), + privacy_level TEXT NOT NULL CHECK ( + privacy_level IN ('essential', 'product', 'diagnostic', 'contribution') + ), + occurred_at TEXT NOT NULL, + received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + distinct_id TEXT NOT NULL, + properties_json TEXT NOT NULL, + posthog_state TEXT NOT NULL DEFAULT 'pending' CHECK (posthog_state IN ('pending', 'sent')), + posthog_attempts INTEGER NOT NULL DEFAULT 0, + posthog_last_error TEXT, + posthog_sent_at TEXT +); + +CREATE INDEX analytics_events_name_occurred_at + ON analytics_events (event_name, occurred_at); + +CREATE INDEX analytics_events_source_occurred_at + ON analytics_events (source, occurred_at); + +CREATE INDEX analytics_events_posthog_queue + ON analytics_events (posthog_state, received_at); diff --git a/package.json b/package.json index 15d556f..9a6be55 100644 --- a/package.json +++ b/package.json @@ -13,9 +13,13 @@ "test": "vitest run", "db:migrate": "wrangler d1 migrations apply scientfactory-downloads --remote", "analytics:report": "node scripts/analytics-report.mjs", + "events:types": "wrangler types workers/events/worker-configuration.d.ts --config workers/events/wrangler.jsonc --env-interface AnalyticsWorkerBindings --include-runtime false", + "events:types:check": "wrangler types workers/events/worker-configuration.d.ts --config workers/events/wrangler.jsonc --env-interface AnalyticsWorkerBindings --include-runtime false --check", + "events:typecheck": "tsc --noEmit --project workers/events/tsconfig.json", + "events:deploy": "wrangler deploy --config workers/events/wrangler.jsonc", "format": "oxfmt", "format:check": "oxfmt --check", - "check": "bun run format:check && bun run typecheck && bun run test && bun run build && bun run build:edge" + "check": "bun run format:check && bun run typecheck && bun run events:types:check && bun run events:typecheck && bun run test && bun run build && bun run build:edge" }, "dependencies": { "@fontsource-variable/dm-sans": "5.3.0", diff --git a/scripts/analytics-report.mjs b/scripts/analytics-report.mjs index a6247b2..918098d 100644 --- a/scripts/analytics-report.mjs +++ b/scripts/analytics-report.mjs @@ -2,7 +2,7 @@ import { spawnSync } from "node:child_process"; const query = ` SELECT - 'all_time_event' AS report_section, + 'legacy_all_time_event' AS report_section, event_name AS item, COUNT(*) AS event_count, MAX(occurred_at) AS latest_event @@ -11,8 +11,30 @@ const query = ` UNION ALL + SELECT + 'all_time_event' AS report_section, + source || ':' || event_name AS item, + COUNT(*) AS event_count, + MAX(occurred_at) AS latest_event + FROM analytics_events + GROUP BY source, event_name + + UNION ALL + SELECT '30_day_download' AS report_section, + COALESCE(json_extract(properties_json, '$.asset_key'), 'unknown') AS item, + COUNT(*) AS event_count, + MAX(occurred_at) AS latest_event + FROM analytics_events + WHERE event_name = 'download_clicked' + AND occurred_at >= datetime('now', '-30 days') + GROUP BY json_extract(properties_json, '$.asset_key') + + UNION ALL + + SELECT + 'legacy_30_day_download' AS report_section, COALESCE(asset_key, 'unknown') AS item, COUNT(*) AS event_count, MAX(occurred_at) AS latest_event @@ -25,6 +47,21 @@ const query = ` SELECT '30_day_outbound' AS report_section, + COALESCE(json_extract(properties_json, '$.destination_host'), 'unknown') || + COALESCE(json_extract(properties_json, '$.destination_path'), '/') AS item, + COUNT(*) AS event_count, + MAX(occurred_at) AS latest_event + FROM analytics_events + WHERE event_name = 'outbound_link_clicked' + AND occurred_at >= datetime('now', '-30 days') + GROUP BY + json_extract(properties_json, '$.destination_host'), + json_extract(properties_json, '$.destination_path') + + UNION ALL + + SELECT + 'legacy_30_day_outbound' AS report_section, COALESCE(destination_host, 'unknown') || COALESCE(destination_path, '/') AS item, COUNT(*) AS event_count, MAX(occurred_at) AS latest_event @@ -37,6 +74,21 @@ const query = ` SELECT '30_day_download_failure' AS report_section, + COALESCE(json_extract(properties_json, '$.failure_stage'), 'unknown') || ':' || + COALESCE(json_extract(properties_json, '$.failure_reason'), 'unknown') AS item, + COUNT(*) AS event_count, + MAX(occurred_at) AS latest_event + FROM analytics_events + WHERE event_name = 'download_failed' + AND occurred_at >= datetime('now', '-30 days') + GROUP BY + json_extract(properties_json, '$.failure_stage'), + json_extract(properties_json, '$.failure_reason') + + UNION ALL + + SELECT + 'legacy_30_day_download_failure' AS report_section, COALESCE(failure_stage, 'unknown') || ':' || COALESCE(failure_reason, 'unknown') AS item, COUNT(*) AS event_count, MAX(occurred_at) AS latest_event @@ -45,6 +97,16 @@ const query = ` AND occurred_at >= datetime('now', '-30 days') GROUP BY failure_stage, failure_reason + UNION ALL + + SELECT + 'posthog_delivery' AS report_section, + posthog_state AS item, + COUNT(*) AS event_count, + MAX(received_at) AS latest_event + FROM analytics_events + GROUP BY posthog_state + ORDER BY report_section, event_count DESC, item `; diff --git a/src/pages/privacy.astro b/src/pages/privacy.astro index 1a83e19..09b5c3f 100644 --- a/src/pages/privacy.astro +++ b/src/pages/privacy.astro @@ -24,6 +24,9 @@ import { REPO_URL } from "../lib/releases";

The event database stores the time, page path, and only the details needed for the event—for example, the installer type or the destination website and path. Query strings are discarded. We do not add an IP address, cookie, browser fingerprint, advertising identifier, referrer, or persistent visitor ID to these event records. This means the counts describe events, not unique people or a person's journey through the site.

+

+ ScientFactory stores these events first in its Cloudflare database. Anonymous copies may also be processed in ScientFactory's EU-hosted PostHog project so we can analyze product usage. PostHog receives the same limited event details and a new event-specific identifier—not a persistent website visitor profile. +

ScientFactory does not use advertising trackers, behavioral session recording, or cross-site tracking on this website.

diff --git a/tsconfig.json b/tsconfig.json index e0f5375..3a0cf94 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -1,4 +1,4 @@ { "extends": "astro/tsconfigs/strict", - "exclude": ["dist", "functions", "node_modules"] + "exclude": ["dist", "functions", "node_modules", "workers"] } diff --git a/workers/events/src/index.test.ts b/workers/events/src/index.test.ts new file mode 100644 index 0000000..da66734 --- /dev/null +++ b/workers/events/src/index.test.ts @@ -0,0 +1,169 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import worker, { flushPendingEvents, validateIngestionPayload } from "./index"; + +function validPayload() { + return { + schema_version: 1, + source: "desktop", + events: [ + { + id: "8e0ee7d5-2c4b-48b6-8209-08f1e536f665", + name: "provider.turn.sent", + distinct_id: "installation:16ace444-e7c3-4b26-893f-98713188ae52", + occurred_at: new Date().toISOString(), + privacy_level: "product", + properties: { provider: "codex", clientType: "desktop-app" }, + }, + ], + }; +} + +function createDatabase() { + const run = vi.fn().mockResolvedValue({ success: true }); + const all = vi.fn().mockResolvedValue({ results: [] }); + const bind = vi.fn(() => ({ run, all })); + const prepare = vi.fn(() => ({ bind })); + const batch = vi.fn().mockResolvedValue([]); + return { database: { prepare, batch } as unknown as D1Database, prepare, bind, batch }; +} + +function incoming(request: Request): Request { + return request as Request; +} + +afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); +}); + +describe("event gateway validation", () => { + it("accepts a bounded desktop event batch", () => { + expect(validateIngestionPayload(validPayload())).toHaveLength(1); + }); + + it("rejects an unsupported source", () => { + expect(() => validateIngestionPayload({ ...validPayload(), source: "unknown" })).toThrow( + "Unsupported event payload", + ); + }); + + it("rejects oversized batches", () => { + const event = validPayload().events[0]; + expect(() => + validateIngestionPayload({ + ...validPayload(), + events: Array.from({ length: 51 }, () => event), + }), + ).toThrow("At most 50 events are accepted"); + }); +}); + +describe("event gateway routes", () => { + it("stores a valid desktop event before accepting it", async () => { + const database = createDatabase(); + const waitUntil = vi.fn(); + const response = await worker.fetch!( + incoming( + new Request("https://events.scientfactory.com/v1/events", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(validPayload()), + }), + ), + { ANALYTICS_DB: database.database } as Cloudflare.Env, + { waitUntil } as unknown as ExecutionContext, + ); + + expect(response.status).toBe(202); + expect(await response.json()).toEqual({ accepted: 1 }); + expect(database.batch).toHaveBeenCalledOnce(); + expect(database.bind).toHaveBeenCalledWith( + "8e0ee7d5-2c4b-48b6-8209-08f1e536f665", + "provider.turn.sent", + "product", + expect.any(String), + "installation:16ace444-e7c3-4b26-893f-98713188ae52", + JSON.stringify({ provider: "codex", clientType: "desktop-app" }), + ); + expect(waitUntil).toHaveBeenCalledOnce(); + }); + + it("rejects browser submissions from unknown origins", async () => { + const database = createDatabase(); + const response = await worker.fetch!( + incoming( + new Request("https://events.scientfactory.com/v1/events", { + method: "POST", + headers: { "Content-Type": "application/json", Origin: "https://attacker.example" }, + body: JSON.stringify(validPayload()), + }), + ), + { ANALYTICS_DB: database.database } as Cloudflare.Env, + { waitUntil: vi.fn() } as unknown as ExecutionContext, + ); + + expect(response.status).toBe(403); + expect(database.batch).not.toHaveBeenCalled(); + }); + + it("reports whether optional PostHog forwarding is configured", async () => { + const database = createDatabase(); + const response = await worker.fetch!( + incoming(new Request("https://events.scientfactory.com/health")), + { ANALYTICS_DB: database.database } as Cloudflare.Env, + { waitUntil: vi.fn() } as unknown as ExecutionContext, + ); + + expect(response.status).toBe(200); + expect(await response.json()).toMatchObject({ + status: "ready", + posthog_forwarding: "pending_configuration", + }); + }); +}); + +describe("PostHog forwarding", () => { + it("forwards stored events without creating PostHog person profiles", async () => { + const row = { + event_id: "8e0ee7d5-2c4b-48b6-8209-08f1e536f665", + event_name: "provider.turn.sent", + source: "desktop", + privacy_level: "product", + occurred_at: new Date().toISOString(), + distinct_id: "installation:16ace444-e7c3-4b26-893f-98713188ae52", + properties_json: JSON.stringify({ provider: "codex" }), + }; + const run = vi.fn().mockResolvedValue({ success: true }); + const all = vi.fn().mockResolvedValue({ results: [row] }); + const bind = vi.fn(() => ({ run, all })); + const prepare = vi.fn(() => ({ bind })); + const batch = vi.fn().mockResolvedValue([]); + const database = { prepare, batch } as unknown as D1Database; + const fetchMock = vi.fn().mockResolvedValue(new Response(null, { status: 200 })); + vi.stubGlobal("fetch", fetchMock); + + const forwarded = await flushPendingEvents({ + ANALYTICS_DB: database, + POSTHOG_PROJECT_TOKEN: "phc_scientfactory_test", + }); + + expect(forwarded).toBe(1); + expect(fetchMock).toHaveBeenCalledOnce(); + const [url, request] = fetchMock.mock.calls[0] ?? []; + expect(url).toBe("https://eu.i.posthog.com/batch"); + const payload = JSON.parse(String(request?.body)) as { + api_key: string; + batch: ReadonlyArray<{ properties: Record }>; + }; + expect(payload.api_key).toBe("phc_scientfactory_test"); + expect(payload.batch[0]?.properties).toMatchObject({ + provider: "codex", + event_id: row.event_id, + source: "desktop", + privacy_level: "product", + $process_person_profile: false, + }); + expect(batch).toHaveBeenCalledOnce(); + }); +}); diff --git a/workers/events/src/index.ts b/workers/events/src/index.ts new file mode 100644 index 0000000..5d56681 --- /dev/null +++ b/workers/events/src/index.ts @@ -0,0 +1,344 @@ +const ALLOWED_WEB_ORIGINS = new Set(["https://scientfactory.com", "https://www.scientfactory.com"]); +const EVENT_NAME_PATTERN = /^[a-z][a-z0-9_.-]{0,79}$/; +const IDENTIFIER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; +const PRIVACY_LEVELS = new Set(["essential", "product", "diagnostic", "contribution"]); +const MAX_REQUEST_BYTES = 128 * 1024; +const MAX_BATCH_SIZE = 50; +const MAX_PROPERTIES_BYTES = 16 * 1024; +const POSTHOG_BATCH_SIZE = 100; +const POSTHOG_HOST = "https://eu.i.posthog.com"; + +type AnalyticsEnv = AnalyticsWorkerBindings & { + readonly POSTHOG_PROJECT_TOKEN?: string; +}; + +interface AcceptedEvent { + readonly id: string; + readonly name: string; + readonly distinctId: string; + readonly occurredAt: string; + readonly privacyLevel: string; + readonly properties: Record; +} + +interface PendingEventRow { + readonly event_id: string; + readonly event_name: string; + readonly source: string; + readonly privacy_level: string; + readonly occurred_at: string; + readonly distinct_id: string; + readonly properties_json: string; +} + +class RequestValidationError extends Error {} + +function jsonResponse(body: unknown, status = 200, origin?: string | null): Response { + const headers = new Headers({ + "Cache-Control": "no-store", + "Content-Type": "application/json; charset=utf-8", + }); + if (origin && ALLOWED_WEB_ORIGINS.has(origin)) { + headers.set("Access-Control-Allow-Origin", origin); + headers.set("Vary", "Origin"); + } + return new Response(JSON.stringify(body), { status, headers }); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function requireString(value: unknown, field: string, pattern: RegExp): string { + if (typeof value !== "string" || !pattern.test(value)) { + throw new RequestValidationError(`Invalid ${field}`); + } + return value; +} + +function parseEvent(value: unknown): AcceptedEvent { + if (!isRecord(value)) throw new RequestValidationError("Each event must be an object"); + + const properties = value.properties; + if (!isRecord(properties)) throw new RequestValidationError("Invalid event properties"); + if (Object.keys(properties).length > 64) { + throw new RequestValidationError("Too many event properties"); + } + if (new TextEncoder().encode(JSON.stringify(properties)).byteLength > MAX_PROPERTIES_BYTES) { + throw new RequestValidationError("Event properties are too large"); + } + + const privacyLevel = requireString(value.privacy_level, "privacy_level", /^[a-z]+$/); + if (!PRIVACY_LEVELS.has(privacyLevel)) { + throw new RequestValidationError("Invalid privacy_level"); + } + + const occurredAtInput = requireString(value.occurred_at, "occurred_at", /^.{10,40}$/); + const occurredAtDate = new Date(occurredAtInput); + if (Number.isNaN(occurredAtDate.valueOf())) { + throw new RequestValidationError("Invalid occurred_at"); + } + const now = Date.now(); + if (occurredAtDate.valueOf() > now + 24 * 60 * 60 * 1000) { + throw new RequestValidationError("occurred_at is too far in the future"); + } + if (occurredAtDate.valueOf() < now - 180 * 24 * 60 * 60 * 1000) { + throw new RequestValidationError("occurred_at is too old"); + } + + return { + id: requireString(value.id, "event id", IDENTIFIER_PATTERN), + name: requireString(value.name, "event name", EVENT_NAME_PATTERN), + distinctId: requireString(value.distinct_id, "distinct_id", IDENTIFIER_PATTERN), + occurredAt: occurredAtDate.toISOString(), + privacyLevel, + properties, + }; +} + +export function validateIngestionPayload(value: unknown): ReadonlyArray { + if (!isRecord(value) || value.schema_version !== 1 || value.source !== "desktop") { + throw new RequestValidationError("Unsupported event payload"); + } + if (!Array.isArray(value.events) || value.events.length < 1) { + throw new RequestValidationError("At least one event is required"); + } + if (value.events.length > MAX_BATCH_SIZE) { + throw new RequestValidationError(`At most ${MAX_BATCH_SIZE} events are accepted`); + } + return value.events.map(parseEvent); +} + +async function readJsonBody(request: Request): Promise { + const contentLength = Number(request.headers.get("Content-Length") ?? "0"); + if (Number.isFinite(contentLength) && contentLength > MAX_REQUEST_BYTES) { + throw new RequestValidationError("Request body is too large"); + } + const body = await request.text(); + if (new TextEncoder().encode(body).byteLength > MAX_REQUEST_BYTES) { + throw new RequestValidationError("Request body is too large"); + } + try { + return JSON.parse(body) as unknown; + } catch { + throw new RequestValidationError("Request body must be valid JSON"); + } +} + +async function persistEvents( + database: D1Database, + events: ReadonlyArray, +): Promise { + const insert = ` + INSERT OR IGNORE INTO analytics_events ( + event_id, + event_name, + source, + privacy_level, + occurred_at, + distinct_id, + properties_json + ) VALUES (?, ?, 'desktop', ?, ?, ?, ?) + `; + await database.batch( + events.map((event) => + database + .prepare(insert) + .bind( + event.id, + event.name, + event.privacyLevel, + event.occurredAt, + event.distinctId, + JSON.stringify(event.properties), + ), + ), + ); +} + +function posthogEvent(row: PendingEventRow): Record { + let properties: Record = {}; + try { + const parsed = JSON.parse(row.properties_json) as unknown; + if (isRecord(parsed)) properties = parsed; + } catch { + // The gateway writes valid JSON; retaining an empty object makes a malformed legacy row retryable. + } + return { + event: row.event_name, + distinct_id: row.distinct_id, + timestamp: row.occurred_at, + properties: { + ...properties, + event_id: row.event_id, + source: row.source, + privacy_level: row.privacy_level, + $process_person_profile: false, + }, + }; +} + +async function markPosthogFailure( + database: D1Database, + rows: ReadonlyArray, + error: string, +): Promise { + if (rows.length === 0) return; + await database.batch( + rows.map((row) => + database + .prepare( + `UPDATE analytics_events + SET posthog_attempts = posthog_attempts + 1, + posthog_last_error = ? + WHERE event_id = ? AND posthog_state = 'pending'`, + ) + .bind(error.slice(0, 500), row.event_id), + ), + ); +} + +export async function flushPendingEvents(env: AnalyticsEnv): Promise { + if (!env.POSTHOG_PROJECT_TOKEN) return 0; + + const result = await env.ANALYTICS_DB.prepare( + `SELECT + event_id, + event_name, + source, + privacy_level, + occurred_at, + distinct_id, + properties_json + FROM analytics_events + WHERE posthog_state = 'pending' + ORDER BY received_at, event_id + LIMIT ?`, + ) + .bind(POSTHOG_BATCH_SIZE) + .all(); + const rows = result.results; + if (rows.length === 0) return 0; + + let response: Response; + try { + response = await fetch(`${POSTHOG_HOST}/batch`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + api_key: env.POSTHOG_PROJECT_TOKEN, + batch: rows.map(posthogEvent), + }), + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await markPosthogFailure(env.ANALYTICS_DB, rows, message); + throw error; + } + + if (!response.ok) { + const message = `PostHog returned ${response.status}`; + await markPosthogFailure(env.ANALYTICS_DB, rows, message); + throw new Error(message); + } + + await env.ANALYTICS_DB.batch( + rows.map((row) => + env.ANALYTICS_DB.prepare( + `UPDATE analytics_events + SET posthog_state = 'sent', + posthog_attempts = posthog_attempts + 1, + posthog_last_error = NULL, + posthog_sent_at = CURRENT_TIMESTAMP + WHERE event_id = ? AND posthog_state = 'pending'`, + ).bind(row.event_id), + ), + ); + return rows.length; +} + +async function handleIngestion( + request: Request, + env: AnalyticsEnv, + context: ExecutionContext, +): Promise { + const origin = request.headers.get("Origin"); + if (origin && !ALLOWED_WEB_ORIGINS.has(origin)) { + return jsonResponse({ error: "Origin is not allowed" }, 403); + } + if (!request.headers.get("Content-Type")?.toLowerCase().startsWith("application/json")) { + return jsonResponse({ error: "Content-Type must be application/json" }, 415, origin); + } + + try { + const events = validateIngestionPayload(await readJsonBody(request)); + await persistEvents(env.ANALYTICS_DB, events); + context.waitUntil( + flushPendingEvents(env).catch((error: unknown) => { + console.error( + JSON.stringify({ + message: "PostHog forwarding failed", + error: error instanceof Error ? error.message : String(error), + }), + ); + }), + ); + return jsonResponse({ accepted: events.length }, 202, origin); + } catch (error) { + if (error instanceof RequestValidationError) { + return jsonResponse({ error: error.message }, 400, origin); + } + console.error( + JSON.stringify({ + message: "Analytics ingestion failed", + error: error instanceof Error ? error.message : String(error), + }), + ); + return jsonResponse({ error: "Analytics ingestion failed" }, 500, origin); + } +} + +const worker: ExportedHandler = { + async fetch(request, env, context) { + const url = new URL(request.url); + if (request.method === "GET" && url.pathname === "/health") { + return jsonResponse({ + status: "ready", + storage: "configured", + posthog_forwarding: env.POSTHOG_PROJECT_TOKEN ? "configured" : "pending_configuration", + }); + } + if (request.method === "OPTIONS" && url.pathname === "/v1/events") { + const origin = request.headers.get("Origin"); + if (!origin || !ALLOWED_WEB_ORIGINS.has(origin)) { + return jsonResponse({ error: "Origin is not allowed" }, 403); + } + const response = new Response(null, { status: 204 }); + response.headers.set("Access-Control-Allow-Origin", origin); + response.headers.set("Access-Control-Allow-Headers", "Content-Type"); + response.headers.set("Access-Control-Allow-Methods", "POST, OPTIONS"); + response.headers.set("Access-Control-Max-Age", "86400"); + response.headers.set("Vary", "Origin"); + return response; + } + if (request.method === "POST" && url.pathname === "/v1/events") { + return handleIngestion(request, env, context); + } + return jsonResponse({ error: "Not found" }, 404); + }, + + async scheduled(_controller, env, context) { + context.waitUntil( + flushPendingEvents(env).catch((error: unknown) => { + console.error( + JSON.stringify({ + message: "Scheduled PostHog forwarding failed", + error: error instanceof Error ? error.message : String(error), + }), + ); + }), + ); + }, +}; + +export default worker; diff --git a/workers/events/tsconfig.json b/workers/events/tsconfig.json new file mode 100644 index 0000000..55a4378 --- /dev/null +++ b/workers/events/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2023", "WebWorker"], + "types": ["@cloudflare/workers-types"], + "noEmit": true + }, + "include": ["./src/**/*.ts", "./worker-configuration.d.ts"] +} diff --git a/workers/events/worker-configuration.d.ts b/workers/events/worker-configuration.d.ts new file mode 100644 index 0000000..6a29728 --- /dev/null +++ b/workers/events/worker-configuration.d.ts @@ -0,0 +1,12 @@ +/* eslint-disable */ +// Generated by Wrangler by running `wrangler types --config=workers/events/wrangler.jsonc --include-runtime=false --env-interface=AnalyticsWorkerBindings workers/events/worker-configuration.d.ts` (hash: 24722b03744e2288284d05eef119d9b4) +interface __BaseEnv_AnalyticsWorkerBindings { + ANALYTICS_DB: D1Database; +} +declare namespace Cloudflare { + interface GlobalProps { + mainModule: typeof import("./src/index"); + } + interface Env extends __BaseEnv_AnalyticsWorkerBindings {} +} +interface AnalyticsWorkerBindings extends __BaseEnv_AnalyticsWorkerBindings {} diff --git a/workers/events/wrangler.jsonc b/workers/events/wrangler.jsonc new file mode 100644 index 0000000..113a6e9 --- /dev/null +++ b/workers/events/wrangler.jsonc @@ -0,0 +1,27 @@ +{ + "$schema": "../../node_modules/wrangler/config-schema.json", + "name": "scientfactory-events", + "main": "src/index.ts", + "compatibility_date": "2026-07-20", + "compatibility_flags": ["nodejs_compat"], + "routes": [ + { + "pattern": "events.scientfactory.com", + "custom_domain": true, + }, + ], + "triggers": { + "crons": ["*/5 * * * *"], + }, + "d1_databases": [ + { + "binding": "ANALYTICS_DB", + "database_name": "scientfactory-downloads", + "database_id": "e106f11e-18e3-4d9d-b61e-7003ac4dd994", + "migrations_dir": "../../migrations", + }, + ], + "observability": { + "enabled": true, + }, +}