feat(plugins): move marketplace counters to analytics engine

deploy/plugin-stats-worker/README.md

@@ -1,34 +1,69 @@
 # plugin-stats-worker
 
 Cloudflare Worker that counts plugin marketplace traffic. Deployed on the same
-Cloudflare account as the dbxio.com site (`pnpm dlx wrangler deploy` from this
+Cloudflare account as the dbxio.com site (`npx wrangler deploy` from this
 directory; OAuth login required).
 
 ## Routes
 
-- `dl.dbxio.com/plugins/*` — counts every `GET` of a marketplace artifact, then
-  passes the request through to the R2 custom domain. Artifact bytes, headers,
-  and Range semantics are untouched; the route sits in front of the origin only.
-- `dbxio.com/api/plugins/*` —
-  `POST /api/plugins/install` is the fire-and-forget beacon the desktop app
-  sends after a successful marketplace install;
-  `GET /api/plugins/stats` returns the current counters
-  (`{ downloads: {key: n}, installs: {key: n} }`) and is how counters should be
-  read (the wrangler CLI can show a stale/split view of a fresh namespace).
+- `dl.dbxio.com/plugins/*` — counts every `GET` of a real `.dbxp` artifact
+  (icons and other static assets under the same prefix are excluded: they are
+  fetched on every marketplace page/app view, which both burned KV quota and
+  inflated the numbers), then passes the request through to the R2 custom
+  domain. Artifact bytes, headers, and Range semantics are untouched.
+- `dbxio.com/api/plugins/install` — `POST` fire-and-forget beacon the desktop
+  app can send after a successful marketplace install
+  (`{"id": "<plugin id>", "version": "<version>"}`, 204 on accept). Decorative
+  statistics only: no auth, no PII.
 
-## Storage
+## Storage — Workers Analytics Engine
 
-Single KV namespace `plugin_stats` (binding `PLUGIN_STATS`):
+One datapoint per event (binding `PLUGIN_STATS`):
 
-- `dl:{pluginId}:{version}` — passive artifact download count
-- `inst:{pluginId}:{version}` — client-reported install count
+- `blobs = [kind, pluginId, version]` where kind is `dl` (artifact GET) or
+  `inst` (install beacon); `doubles = [1]`; `indexes = [pluginId]`.
 
-Counters are best-effort (KV read-modify-write can drop counts under same-key
-concurrency; acceptable for decorative stats). Inspect values with:
+Chosen over KV counters: KV read-modify-write costs 1 write per request, and
+the free tier's 1k writes/day is below what a 30k-user base generates on a hot
+plugin release day (2026-09-15: icon traffic blew the quota within hours).
+Analytics Engine writes are effectively free at this scale.
+
+## Daily aggregation (permanent archive)
+
+AE retains only three months, so a cron trigger (`30 0 * * *` UTC) aggregates the
+trailing window into the `plugin_stats_archive` KV namespace — the permanent
+layer future UI reads:
+
+- `total:{kind}:{pluginId}:{version}` — all-time cumulative events
+- `summary` — one JSON blob `{dl: {pluginId: n}, inst: {...}}` for single-read display
+- `meta:last-success` — window marker; only advanced after all writes land, so a
+  failed run retries the same window on the next cron
+
+Required secret (SQL API access; OAuth login does not carry analytics scope):
 
 ```sh
-pnpm dlx wrangler kv key list --namespace-id 483dcc36cd3c4da5af5cdbb4532166f2 --prefix "dl:io.dbx.ssh"
-pnpm dlx wrangler kv key get --namespace-id 483dcc36cd3c4da5af5cdbb4532166f2 "dl:io.dbx.ssh:0.4.73"
+echo "<token>" | npx wrangler secret put ANALYTICS_TOKEN
 ```
 
-No display in the app yet; counts accrue until the marketplace UI surfaces them.
+Create the token in the dashboard with **Account → Analytics → Read**. The same
+token authorizes the manual trigger (verification / backfill):
+
+```sh
+curl -s -X POST https://dbxio.com/api/plugins/archive -H "x-archive-token: <token>"
+```
+
+## Reading counters
+
+No HTTP stats endpoint (there is no consumer yet; add one when the marketplace
+UI ships). Query via the SQL REST API with an API token that has
+Account → Analytics → Read:
+
+```sh
+curl -s "https://api.cloudflare.com/client/v4/accounts/<account_id>/analytics_engine/sql" \
+  -H "Authorization: Bearer <token>" \
+  --data-urlencode "query=SELECT blob1 AS kind, blob2 AS plugin, blob3 AS version, SUM(_sample_interval) AS events FROM PLUGIN_STATS WHERE timestamp > NOW() - INTERVAL '7' DAY GROUP BY 1,2,3 ORDER BY events DESC"
+```
+
+At current volume sampling is 1:1, so `SUM(_sample_interval)` equals the event
+count. Counters start from the Analytics Engine migration; the polluted KV-era
+numbers (`plugin_stats` namespace) were discarded.

deploy/plugin-stats-worker/worker.ts

@@ -1,80 +1,163 @@
 // Counts plugin marketplace traffic on Cloudflare without touching artifact bytes.
 //
 // Routes (see wrangler.json):
-// - dl.dbxio.com/plugins/*: counts each GET of a marketplace artifact, then passes
-//   the request through to the R2 custom domain origin. Same-zone subrequests do not
-//   re-enter Workers, so the pass-through cannot loop.
-// - dbxio.com/api/plugins/install: beacon endpoint the desktop app calls after a
-//   successful marketplace install. Decorative statistics only: no auth, no PII,
-//   failures are silently ignored by clients.
-// - dbxio.com/api/plugins/stats: read view of the counters, for later display in
-//   the marketplace UI and for ops checks.
+// - dl.dbxio.com/plugins/*: counts each GET of a real `.dbxp` artifact (icons and
+//   other static assets are excluded — they are fetched on every marketplace
+//   page/app view and would both burn quota and pollute the numbers), then
+//   passes the request through to the R2 custom domain. Same-zone subrequests do
+//   not re-enter Workers, so the pass-through cannot loop.
+// - dbxio.com/api/plugins/install: fire-and-forget beacon the desktop app can
+//   send after a successful marketplace install. Decorative statistics only.
+// - dbxio.com/api/plugins/archive: token-gated manual trigger for the daily
+//   aggregation (same handler the cron runs), for verification and backfills.
 //
-// Counters live in KV under `dl:{pluginId}:{version}` (passive downloads) and
-// `inst:{pluginId}:{version}` (client beacons). KV read-modify-write can drop
-// counts under same-key concurrency; acceptable for decorative stats. NOTE: the
-// wrangler CLI/API view of a freshly created namespace can lag or split from the
-// runtime view for a long time — always read counters through the stats endpoint,
-// not `wrangler kv key get`.
+// Storage:
+// - Workers Analytics Engine (binding PLUGIN_STATS) — one datapoint per event.
+//   AE only retains three months, so a daily cron aggregates the trailing
+//   window into the `plugin_stats_archive` KV namespace, which holds the
+//   permanent totals. KV is fine here: ~1 write per active key per day.
 
-type KvNamespaceBinding = {
+type AnalyticsEngineBinding = {
+  writeDataPoint(event: { blobs?: string[]; doubles?: number[]; indexes?: string[] }): void;
+};
+
+type KvBinding = {
   get(key: string): Promise<string | null>;
   put(key: string, value: string): Promise<void>;
-  list(options?: { prefix?: string }): Promise<{ keys: { name: string }[] }>;
 };
 
 type Env = {
-  PLUGIN_STATS: KvNamespaceBinding;
+  PLUGIN_STATS: AnalyticsEngineBinding;
+  PLUGIN_ARCHIVE: KvBinding;
+  ANALYTICS_TOKEN: string;
+  CF_ACCOUNT_ID: string;
 };
 
 const DOWNLOAD_PATTERN = /^\/plugins\/([A-Za-z0-9._-]{1,64})\/([0-9A-Za-z.+-]{1,32})\//;
+const ARTIFACT_SUFFIX_PATTERN = /\.dbxp$/;
 const PLUGIN_ID_PATTERN = /^[A-Za-z0-9._-]{1,64}$/;
 const VERSION_PATTERN = /^[0-9A-Za-z.+-]{1,32}$/;
 const INSTALL_BODY_LIMIT_BYTES = 512;
+const COUNTED_KINDS = new Set(["dl", "inst"]);
+const ARCHIVE_META_KEY = "meta:last-success";
+const ARCHIVE_SUMMARY_KEY = "summary";
+const ARCHIVE_TRIGGER_HEADER = "x-archive-token";
+const SQL_ENDPOINT = "https://api.cloudflare.com/client/v4/accounts";
 const CORS_HEADERS: Record<string, string> = {
   "Access-Control-Allow-Origin": "*",
-  "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
+  "Access-Control-Allow-Methods": "POST, OPTIONS",
   "Access-Control-Allow-Headers": "Content-Type",
   "Access-Control-Max-Age": "86400",
 };
 
-async function increment(env: Env, key: string): Promise<void> {
+function recordEvent(env: Env, kind: "dl" | "inst", pluginId: string, version: string): void {
   try {
-    const current = await env.PLUGIN_STATS.get(key);
-    const next = Number.parseInt(current ?? "0", 10) || 0;
-    await env.PLUGIN_STATS.put(key, String(next + 1));
+    env.PLUGIN_STATS.writeDataPoint({
+      // index on pluginId keeps per-plugin queries well-sharded
+      indexes: [pluginId],
+      blobs: [kind, pluginId, version],
+      doubles: [1],
+    });
   } catch (error) {
-    console.error(`plugin-stats increment failed for ${key}`, error);
+    console.error(`plugin-stats writeDataPoint failed for ${kind}:${pluginId}:${version}`, error);
   }
 }
 
-async function readCounters(env: Env, prefix: string): Promise<Record<string, number>> {
-  const listing = await env.PLUGIN_STATS.list({ prefix });
-  const counters: Record<string, number> = {};
-  await Promise.all(
-    listing.keys.map(async (entry) => {
-      try {
-        counters[entry.name] = Number.parseInt((await env.PLUGIN_STATS.get(entry.name)) ?? "0", 10) || 0;
-      } catch {
-        counters[entry.name] = 0;
-      }
-    }),
-  );
-  return counters;
-}
-
 function emptyResponse(status: number): Response {
   return new Response(null, { status, headers: CORS_HEADERS });
 }
 
+function sqlString(value: string): string {
+  return `'${value.replaceAll("'", "''")}'`;
+}
+
+function sqlTimestamp(date: Date): string {
+  return date.toISOString().slice(0, 19).replace("T", " ");
+}
+
+type WindowCount = { kind: string; plugin: string; version: string; events: number };
+
+// The SQL API answers with ClickHouse-style {meta, data}; rows may arrive as
+// objects or as arrays to zip against meta. Accept a plain record array too so
+// a format drift degrades instead of breaking.
+function parseSqlRows(payload: unknown): Array<Record<string, unknown>> {
+  if (typeof payload !== "object" || payload === null) return [];
+  const { meta, data, result } = payload as Record<string, unknown>;
+  if (Array.isArray(data)) {
+    const columns = Array.isArray(meta) ? (meta as Array<{ name?: string }>).map((entry) => entry?.name ?? "") : [];
+    return (data as unknown[]).map((row) => {
+      if (Array.isArray(row)) return Object.fromEntries(columns.map((column, index) => [column, row[index]]));
+      if (typeof row === "object" && row !== null) return row as Record<string, unknown>;
+      return {};
+    });
+  }
+  return Array.isArray(result) ? (result as Array<Record<string, unknown>>) : [];
+}
+
+async function fetchWindowCounts(env: Env, from: Date, to: Date): Promise<WindowCount[]> {
+  const query =
+    "SELECT blob1 AS kind, blob2 AS plugin, blob3 AS version, SUM(_sample_interval) AS events " +
+    `FROM PLUGIN_STATS WHERE timestamp > toDateTime(${sqlString(sqlTimestamp(from))}) ` +
+    `AND timestamp <= toDateTime(${sqlString(sqlTimestamp(to))}) GROUP BY blob1, blob2, blob3`;
+  // The SQL API takes the statement as a URL parameter, not a form body (a
+  // form-encoded body is parsed as the raw SQL and rejected with a 422).
+  const response = await fetch(`${SQL_ENDPOINT}/${env.CF_ACCOUNT_ID}/analytics_engine/sql?query=${encodeURIComponent(query)}`, {
+    method: "GET",
+    headers: { Authorization: `Bearer ${env.ANALYTICS_TOKEN}` },
+  });
+  if (!response.ok) {
+    throw new Error(`Analytics SQL API ${response.status}: ${(await response.text()).slice(0, 300)}`);
+  }
+  const payload = (await response.json()) as unknown;
+  return parseSqlRows(payload)
+    .map((row) => ({
+      kind: String(row.kind ?? ""),
+      plugin: String(row.plugin ?? ""),
+      version: String(row.version ?? ""),
+      events: Number(row.events ?? 0) || 0,
+    }))
+    .filter((row) => COUNTED_KINDS.has(row.kind) && PLUGIN_ID_PATTERN.test(row.plugin) && VERSION_PATTERN.test(row.version));
+}
+
+// Aggregates events in (last-success, now] into permanent KV totals. The window
+// marker is only advanced after every write lands, so a failed run retries the
+// same window on the next cron; a crash mid-write can double-count a window,
+// which is acceptable for decorative stats.
+async function runAggregation(env: Env): Promise<{ from: string; to: string; activeKeys: number }> {
+  const to = new Date();
+  const lastSuccess = await env.PLUGIN_ARCHIVE.get(ARCHIVE_META_KEY);
+  const from = lastSuccess ? new Date(`${lastSuccess.replace(" ", "T")}Z`) : new Date(0);
+  if (Number.isNaN(from.getTime())) throw new Error(`Corrupted ${ARCHIVE_META_KEY}: ${lastSuccess}`);
+
+  const counts = await fetchWindowCounts(env, from, to);
+  const summary = JSON.parse((await env.PLUGIN_ARCHIVE.get(ARCHIVE_SUMMARY_KEY)) ?? "{}") as Record<string, Record<string, number>>;
+
+  for (const count of counts) {
+    const totalKey = `total:${count.kind}:${count.plugin}:${count.version}`;
+    const current = Number.parseInt((await env.PLUGIN_ARCHIVE.get(totalKey)) ?? "0", 10) || 0;
+    await env.PLUGIN_ARCHIVE.put(totalKey, String(current + count.events));
+    const byKind = (summary[count.kind] ??= {});
+    byKind[count.plugin] = (byKind[count.plugin] ?? 0) + count.events;
+  }
+
+  await env.PLUGIN_ARCHIVE.put(ARCHIVE_SUMMARY_KEY, JSON.stringify(summary));
+  await env.PLUGIN_ARCHIVE.put(ARCHIVE_META_KEY, sqlTimestamp(to));
+  return { from: sqlTimestamp(from), to: sqlTimestamp(to), activeKeys: counts.length };
+}
+
 async function handleInstallBeacon(request: Request, env: Env): Promise<Response> {
   if (request.method === "OPTIONS") return emptyResponse(204);
 
   const url = new URL(request.url);
-  if (url.pathname === "/api/plugins/stats") {
-    if (request.method !== "GET") return emptyResponse(405);
-    const [downloads, installs] = await Promise.all([readCounters(env, "dl:"), readCounters(env, "inst:")]);
-    return Response.json({ downloads, installs }, { headers: { ...CORS_HEADERS, "Cache-Control": "no-store" } });
+  if (url.pathname === "/api/plugins/archive") {
+    if (request.method !== "POST") return emptyResponse(405);
+    if (request.headers.get(ARCHIVE_TRIGGER_HEADER) !== env.ANALYTICS_TOKEN) return emptyResponse(401);
+    try {
+      return Response.json(await runAggregation(env), { headers: { "Cache-Control": "no-store" } });
+    } catch (error) {
+      console.error("plugin-stats aggregation failed", error);
+      return Response.json({ error: String(error) }, { status: 500, headers: { "Cache-Control": "no-store" } });
+    }
   }
 
   if (request.method !== "POST") return emptyResponse(405);
@@ -91,20 +174,30 @@ async function handleInstallBeacon(request: Request, env: Env): Promise<Response
   if (typeof id !== "string" || !PLUGIN_ID_PATTERN.test(id)) return emptyResponse(400);
   if (typeof version !== "string" || !VERSION_PATTERN.test(version)) return emptyResponse(400);
 
-  await increment(env, `inst:${id}:${version}`);
+  recordEvent(env, "inst", id, version);
   return emptyResponse(204);
 }
 
 export default {
-  async fetch(request: Request, env: Env, ctx: { waitUntil(promise: Promise<unknown>): void }): Promise<Response> {
+  async fetch(request: Request, env: Env): Promise<Response> {
     const url = new URL(request.url);
     if (url.hostname === "dl.dbxio.com") {
       if (request.method === "GET") {
         const download = url.pathname.match(DOWNLOAD_PATTERN);
-        if (download) ctx.waitUntil(increment(env, `dl:${download[1]}:${download[2]}`));
+        if (download && ARTIFACT_SUFFIX_PATTERN.test(url.pathname)) {
+          recordEvent(env, "dl", download[1], download[2]);
+        }
       }
       return fetch(request);
     }
     return handleInstallBeacon(request, env);
   },
+
+  async scheduled(_controller: unknown, env: Env, ctx: { waitUntil(promise: Promise<unknown>): void }): Promise<void> {
+    ctx.waitUntil(
+      runAggregation(env).catch((error) => {
+        console.error("plugin-stats scheduled aggregation failed", error);
+      }),
+    );
+  },
 };

deploy/plugin-stats-worker/wrangler.json

@@ -3,13 +3,33 @@
   "compatibility_date": "2026-05-07",
   "main": "worker.ts",
   "routes": [
-    { "pattern": "dl.dbxio.com/plugins/*", "zone_name": "dbxio.com" },
-    { "pattern": "dbxio.com/api/plugins/*", "zone_name": "dbxio.com" }
+    {
+      "pattern": "dl.dbxio.com/plugins/*",
+      "zone_name": "dbxio.com"
+    },
+    {
+      "pattern": "dbxio.com/api/plugins/*",
+      "zone_name": "dbxio.com"
+    }
+  ],
+  "analytics_engine_datasets": [
+    {
+      "binding": "PLUGIN_STATS",
+      "dataset": "DBX_PLUGIN_STATS"
+    }
   ],
   "kv_namespaces": [
     {
-      "binding": "PLUGIN_STATS",
-      "id": "483dcc36cd3c4da5af5cdbb4532166f2"
+      "binding": "PLUGIN_ARCHIVE",
+      "id": "0cc9e49a117f4edeb931464fa0256834"
     }
-  ]
+  ],
+  "vars": {
+    "CF_ACCOUNT_ID": "7ad3349ede62c17968a90108d39fd4d0"
+  },
+  "triggers": {
+    "crons": [
+      "30 0 * * *"
+    ]
+  }
 }