From d19e43b42b36afe66a49721ff12019e12c8f1bd3 Mon Sep 17 00:00:00 2001 From: t8y2 <1156263951@qq.com> Date: Tue, 15 Sep 2026 20:58:00 +0800 Subject: [PATCH] feat(plugins): move marketplace counters to analytics engine --- deploy/plugin-stats-worker/README.md | 73 ++++++--- deploy/plugin-stats-worker/worker.ts | 183 +++++++++++++++++------ deploy/plugin-stats-worker/wrangler.json | 30 +++- 3 files changed, 217 insertions(+), 69 deletions(-) diff --git a/deploy/plugin-stats-worker/README.md b/deploy/plugin-stats-worker/README.md index 3259b4b3ff..782fa272bf 100644 --- a/deploy/plugin-stats-worker/README.md +++ b/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": "", "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 "" | 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: " +``` + +## 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//analytics_engine/sql" \ + -H "Authorization: Bearer " \ + --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. diff --git a/deploy/plugin-stats-worker/worker.ts b/deploy/plugin-stats-worker/worker.ts index 8bdf3a95df..a54a12b420 100644 --- a/deploy/plugin-stats-worker/worker.ts +++ b/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; put(key: string, value: string): Promise; - 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 = { "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 { +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> { - const listing = await env.PLUGIN_STATS.list({ prefix }); - const counters: Record = {}; - 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> { + if (typeof payload !== "object" || payload === null) return []; + const { meta, data, result } = payload as Record; + 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; + return {}; + }); + } + return Array.isArray(result) ? (result as Array>) : []; +} + +async function fetchWindowCounts(env: Env, from: Date, to: Date): Promise { + 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>; + + 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 { 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): void }): Promise { + async fetch(request: Request, env: Env): Promise { 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): void }): Promise { + ctx.waitUntil( + runAggregation(env).catch((error) => { + console.error("plugin-stats scheduled aggregation failed", error); + }), + ); + }, }; diff --git a/deploy/plugin-stats-worker/wrangler.json b/deploy/plugin-stats-worker/wrangler.json index bc74f3b8a7..e5dbe4a9f5 100644 --- a/deploy/plugin-stats-worker/wrangler.json +++ b/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 * * *" + ] + } }