ch22-analytics-engine

對應 22. Analytics Engine：自己蓋一個分析後台

在 GitHub 上檢視·4 個檔案·8.5 KB

取得並執行

這個範例可以獨立 clone 執行，不依賴其他章節。

npx degit mrsuner/cloudflare-serverless-platform-manuals/examples/ch22-analytics-engine
cd ch22-analytics-engine
npm install

可用指令

npm run dev	# wrangler dev
npm run deploy	# wrangler deploy
npm run types	# wrangler types --env-interface CloudflareBindings

說明

ch22 — Workers Analytics Engine

Probe project for chapter 22.

The headline result: the local Analytics Engine binding validates nothing and stores nothing. Every route below exists to demonstrate that, and to give you a validation harness you can run in its place.

Run

npm install
npx wrangler types --env-interface CloudflareBindings
npx wrangler dev --port 9024

Routes

Route	What it shows
/shape	Binding prototype, constructor name (LocalAnalyticsEngineDataset), and the return value of writeDataPoint (undefined, not a Promise).
/write?country=&slug=	A well-formed data point.
/limits	23 probes: no args, 21 blobs, 21 doubles, 2 indexes, 97-byte index, 16 KB + 1 blob, numbers as blobs, null, Uint8Array, NaN, Infinity, unknown keys. All 23 return undefined and throw nothing locally.
/burst?n=300	300 data points in one invocation. Documented limit is 250; locally there is no error.
/no-dataset	The AE_NO_DATASET binding has no dataset field — it defaults to the binding name.
/after	writeDataPoint from inside ctx.waitUntil.
/sql?q=	Raw SQL API passthrough. Needs credentials (below).
/tables	SHOW TABLES.
/report	The sampling-aware aggregation from the chapter: sum(_sample_interval), weighted average, quantileExactWeighted, and count() as a confidence signal.

SQL routes

The SQL API is not simulated locally and there is no local query path. To use /sql, /tables and /report you need real credentials:

wrangler.jsonc

"vars": { "CF_ACCOUNT_ID": "<your account id>" }

npx wrangler secret put CF_API_TOKEN   # custom token with Account Analytics:Read

Then deploy — writes from wrangler dev do not reach the production dataset.

Reproducing the key findings

# writeDataPoint is synchronous and returns undefined
curl -s localhost:9024/shape | jq '{ctorName, writeReturn}'

# Nothing validates. Every probe is identical.
curl -s localhost:9024/limits | jq 'to_entries | map(.value.ok) | unique'
# => ["undefined"]

# 300 > the documented 250-per-invocation limit, no error
curl -s localhost:9024/burst?n=300 | jq

# No local persistence: there is no analytics-engine state directory
ls .wrangler/state/v3/

Config schema facts

From node_modules/wrangler/config-schema.json:

{
  "properties": { "binding": {"type":"string"}, "dataset": {"type":"string"} },
  "required": ["binding"],
  "additionalProperties": false
}

Two consequences: dataset is optional (defaults to the binding name), and there is no remote support — adding "remote": true produces Unexpected fields found in analytics_engine_datasets[0] field: "remote", a warning only, so a typo here is silently ignored at deploy time.

Column mapping is positional and permanent

There are no column names. blobs[0] is blob1, doubles[1] is double2, indexes[0] is index1. Once data is written you cannot rename or reorder, and there is no ALTER. Declare the mapping as a constant next to the writer and treat it as an append-only schema.

blob1=slug blob2=country blob3=referrer blob4=ua blob5=tenantId
double1=count double2=latencyMs
index1=tenantId

Always aggregate with _sample_interval

count() is not the event count — it is the number of rows the query actually read, i.e. a confidence signal. The event count is sum(_sample_interval). See /report for the full pattern.

原始碼

• wrangler.jsonc
• package.json
• src/index.ts
• tsconfig.json

wrangler.jsonc

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "ch22-analytics-engine",
  "main": "src/index.ts",
  "compatibility_date": "2026-07-24",
  "observability": { "enabled": true },
  "analytics_engine_datasets": [
    { "binding": "AE", "dataset": "ch22_clicks" },
    // `dataset` is optional -- it defaults to the binding name.
    { "binding": "AE_NO_DATASET" }
  ]
}

package.json

{
  "name": "ch22-analytics-engine",
  "private": true,
  "scripts": {
    "dev": "wrangler dev",
    "deploy": "wrangler deploy",
    "types": "wrangler types --env-interface CloudflareBindings"
  },
  "devDependencies": { "wrangler": "^4.118.0", "typescript": "^5.9.2" }
}

src/index.ts

// Probe project for chapter 22 -- Workers Analytics Engine.
// Every route exists to surface a runtime behaviour the docs omit.

const t = (fn: () => unknown): unknown => {
  try {
    const v = fn();
    return { ok: v === undefined ? "undefined" : v, isPromise: v instanceof Promise };
  } catch (e) {
    return { threwName: (e as Error).name, threw: String(e).slice(0, 240) };
  }
};

type SqlBody = { meta?: unknown; data?: unknown; rows?: number };

async function sql(env: CloudflareBindings, query: string): Promise<unknown> {
  const account = (env as unknown as { CF_ACCOUNT_ID?: string }).CF_ACCOUNT_ID;
  const token = (env as unknown as { CF_API_TOKEN?: string }).CF_API_TOKEN;
  if (!account || !token) {
    return { skipped: "set CF_ACCOUNT_ID (vars) and CF_API_TOKEN (secret) to run SQL" };
  }
  const res = await fetch(
    `https://api.cloudflare.com/client/v4/accounts/${account}/analytics_engine/sql`,
    { method: "POST", headers: { authorization: `Bearer ${token}` }, body: query },
  );
  const text = await res.text();
  let parsed: SqlBody | string = text;
  try {
    parsed = JSON.parse(text) as SqlBody;
  } catch {
    /* FORMAT TabSeparated etc. */
  }
  return { status: res.status, contentType: res.headers.get("content-type"), body: parsed };
}

export default {
  async fetch(request: Request, env: CloudflareBindings, ctx: ExecutionContext): Promise<Response> {
    const url = new URL(request.url);
    const q = url.searchParams;
    const ae = env.AE;

    switch (url.pathname) {
      // What does the binding actually look like, and what does writeDataPoint return?
      case "/shape": {
        const proto = Object.getPrototypeOf(ae);
        return Response.json({
          typeofBinding: typeof ae,
          ownKeys: Object.getOwnPropertyNames(ae),
          protoKeys: Object.getOwnPropertyNames(proto),
          ctorName: proto?.constructor?.name ?? null,
          writeReturn: t(() => ae.writeDataPoint({ blobs: ["shape"], doubles: [1], indexes: ["k"] })),
        });
      }

      // A normal, well-formed write.
      case "/write": {
        const country = q.get("country") ?? "TW";
        const slug = q.get("slug") ?? "abc";
        return Response.json({
          wrote: t(() =>
            ae.writeDataPoint({
              blobs: [slug, country, request.headers.get("user-agent") ?? "", "referrer"],
              doubles: [1, Date.now() % 1000],
              indexes: [slug],
            }),
          ),
        });
      }

      // Every documented and undocumented limit, probed one at a time.
      case "/limits": {
        const big = (n: number) => "x".repeat(n);
        const out: Record<string, unknown> = {};

        out.noArgs = t(() => (ae as unknown as { writeDataPoint(): void }).writeDataPoint());
        out.emptyObject = t(() => ae.writeDataPoint({}));
        out.blobsOnly = t(() => ae.writeDataPoint({ blobs: ["a"] }));
        out.doublesOnly = t(() => ae.writeDataPoint({ doubles: [1] }));

        // documented: 20 blobs max
        out.blobs20 = t(() => ae.writeDataPoint({ blobs: Array(20).fill("a") }));
        out.blobs21 = t(() => ae.writeDataPoint({ blobs: Array(21).fill("a") }));

        // documented: 20 doubles max
        out.doubles20 = t(() => ae.writeDataPoint({ doubles: Array(20).fill(1) }));
        out.doubles21 = t(() => ae.writeDataPoint({ doubles: Array(21).fill(1) }));

        // documented: exactly 1 index; more => "your data point will not be recorded"
        out.indexes0 = t(() => ae.writeDataPoint({ indexes: [], blobs: ["i0"] }));
        out.indexes1 = t(() => ae.writeDataPoint({ indexes: ["a"], blobs: ["i1"] }));
        out.indexes2 = t(() => ae.writeDataPoint({ indexes: ["a", "b"], blobs: ["i2"] }));

        // documented: index <= 96 bytes
        out.index96 = t(() => ae.writeDataPoint({ indexes: [big(96)] }));
        out.index97 = t(() => ae.writeDataPoint({ indexes: [big(97)] }));

        // documented: total blobs <= 16 KB per data point
        out.blobs16k = t(() => ae.writeDataPoint({ blobs: [big(16 * 1024)] }));
        out.blobs16kPlus1 = t(() => ae.writeDataPoint({ blobs: [big(16 * 1024 + 1)] }));

        // type contract: blobs are strings, doubles are numbers
        out.blobNumber = t(() =>
          ae.writeDataPoint({ blobs: [123 as unknown as string] }),
        );
        out.blobNull = t(() => ae.writeDataPoint({ blobs: [null as unknown as string] }));
        out.blobBuffer = t(() =>
          ae.writeDataPoint({ blobs: [new Uint8Array([1, 2, 3]) as unknown as string] }),
        );
        out.doubleString = t(() =>
          ae.writeDataPoint({ doubles: ["7" as unknown as number] }),
        );
        out.doubleNaN = t(() => ae.writeDataPoint({ doubles: [NaN] }));
        out.doubleInfinity = t(() => ae.writeDataPoint({ doubles: [Infinity] }));
        out.indexNonString = t(() =>
          ae.writeDataPoint({ indexes: [42 as unknown as string] }),
        );

        // unknown key
        out.unknownKey = t(() =>
          ae.writeDataPoint({ blobs: ["u"], tags: ["nope"] } as unknown as AnalyticsEngineDataPoint),
        );

        return Response.json(out);
      }

      // documented: 250 data points per invocation
      case "/burst": {
        const n = Number(q.get("n") ?? 260);
        const errors: unknown[] = [];
        for (let i = 0; i < n; i++) {
          const r = t(() => ae.writeDataPoint({ blobs: [`burst-${i}`], indexes: ["burst"] })) as {
            threw?: string;
          };
          if (r.threw) errors.push({ i, ...r });
        }
        return Response.json({ attempted: n, firstErrors: errors.slice(0, 3), errorCount: errors.length });
      }

      // Does a binding without an explicit `dataset` work?
      case "/no-dataset":
        return Response.json({
          wrote: t(() => env.AE_NO_DATASET.writeDataPoint({ blobs: ["default-dataset"] })),
        });

      // Is writeDataPoint allowed from a waitUntil / after the response?
      case "/after": {
        ctx.waitUntil(
          (async () => {
            await new Promise((r) => setTimeout(r, 50));
            ae.writeDataPoint({ blobs: ["from-waitUntil"], indexes: ["late"] });
          })(),
        );
        return Response.json({ scheduled: true });
      }

      // SQL API. Requires CF_ACCOUNT_ID (var) + CF_API_TOKEN (secret).
      case "/sql":
        return Response.json(
          await sql(env, q.get("q") ?? "SELECT 1 AS one FORMAT JSON"),
        );

      case "/tables":
        return Response.json(await sql(env, "SHOW TABLES"));

      // The correct, sampling-aware aggregation for LinkForge clicks.
      case "/report":
        return Response.json(
          await sql(
            env,
            `SELECT
               blob2                                   AS country,
               sum(_sample_interval)                   AS clicks,
               sum(double1 * _sample_interval)         AS weight,
               sum(double1 * _sample_interval) / sum(_sample_interval) AS avg_weight,
               quantileExactWeighted(0.95)(double2, _sample_interval)  AS p95,
               count()                                 AS rows_read
             FROM ch22_clicks
             WHERE timestamp > NOW() - INTERVAL '1' DAY
             GROUP BY country
             ORDER BY clicks DESC
             LIMIT 20
             FORMAT JSON`,
          ),
        );

      default:
        return new Response(
          "/shape /write?country=&slug= /limits /burst?n= /no-dataset /after\n" +
            "/sql?q=  /tables  /report   (SQL routes need CF_ACCOUNT_ID + CF_API_TOKEN)\n",
          { status: 404 },
        );
    }
  },
} satisfies ExportedHandler<CloudflareBindings>;

tsconfig.json

{
  "compilerOptions": {
    "target": "esnext",
    "lib": ["esnext"],
    "module": "esnext",
    "moduleResolution": "bundler",
    "types": ["./worker-configuration.d.ts"],
    "strict": true,
    "skipLibCheck": true,
    "noEmit": true,
    "isolatedModules": true
  },
  "include": ["src/**/*.ts", "worker-configuration.d.ts"]
}