feat(sdk): 批量传输（批量/退避/信标/丢包计数）

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-15 16:07:18 +08:00
parent 18537c5dba
commit 6be163a329
3 changed files with 293 additions and 0 deletions
@@ -0,0 +1,12 @@
+export interface CtkConfig {
+  siteId: string;
+  endpoint: string;
+  batchSize?: number;
+  flushMs?: number;
+  maxRetry?: number;
+  queueMax?: number;
+}
+
+export function mergeConfig(opts: CtkConfig): Required<CtkConfig> {
+  return { batchSize: 10, flushMs: 15000, maxRetry: 3, queueMax: 200, ...opts };
+}
@@ -0,0 +1,86 @@
+import { uuid } from "./util";
+import { getVisitorId, getSessionId } from "./identity";
+import type { CtkConfig } from "./config";
+
+export type TransportConfig = Required<CtkConfig>;
+
+// 批量传输：满 batchSize 或 flushMs 到期即发；4xx（除 429）不重试；
+// 指数退避 + 抖动重试；重试耗尽按本批事件数计 dropped；queueMax 溢出丢最旧。
+// 红线：任何异常不得冒泡到宿主页（定时器/事件回调内的异常无人捕获）。
+export class Transport {
+  private q: object[] = [];
+  private timer: ReturnType<typeof setTimeout> | null = null;
+  private retry = 0;
+  dropped = 0;
+  constructor(public cfg: TransportConfig) {}
+
+  push(ev: object) {
+    if (this.q.length >= this.cfg.queueMax) {
+      this.q.shift();
+      this.dropped++;
+    }
+    this.q.push(ev);
+    if (this.q.length >= this.cfg.batchSize) this.flush();
+    else if (!this.timer)
+      this.timer = setTimeout(() => this.flush(), this.cfg.flushMs);
+  }
+
+  flush(beacon = false) {
+    if (this.timer) {
+      clearTimeout(this.timer);
+      this.timer = null;
+    }
+    if (!this.q.length) return;
+    try {
+      const events = this.q;
+      this.q = [];
+      const body = JSON.stringify({
+        batch_id: uuid(),
+        site_id: this.cfg.siteId,
+        visitor_id: getVisitorId(),
+        session_id: getSessionId(),
+        sent_at: new Date().toISOString(),
+        // 由 index.ts（Task 15）注入；未注入时 undefined——JSON 序列化自然缺键，照发
+        context: (globalThis as { __ctk_ctx?: unknown }).__ctk_ctx,
+        events,
+      });
+      if (beacon && navigator.sendBeacon) {
+        navigator.sendBeacon(
+          this.cfg.endpoint,
+          new Blob([body], { type: "text/plain" })
+        );
+        return;
+      }
+      this.send(body, events.length);
+    } catch {
+      // 定时器/事件回调内的任何异常不得冒泡到宿主页
+    }
+  }
+
+  private async send(body: string, count: number) {
+    try {
+      const res = await fetch(this.cfg.endpoint, {
+        method: "POST",
+        keepalive: true,
+        headers: { "Content-Type": "text/plain" },
+        body,
+      });
+      if (res.status >= 400 && res.status < 500 && res.status !== 429) {
+        this.retry = 0;
+        return; // 4xx 参数/权限错不重试（403 重试只会掩盖配置错误）
+      }
+      if (!res.ok) throw new Error(String(res.status));
+      this.retry = 0;
+    } catch {
+      if (this.retry++ < this.cfg.maxRetry) {
+        const delay = 500 * 2 ** this.retry * (0.8 + Math.random() * 0.4);
+        setTimeout(() => {
+          this.send(body, count).catch(() => {});
+        }, delay); // 重放同一 body 字符串——同 event_id 重放，服务端幂等去重
+      } else {
+        this.retry = 0;
+        this.dropped += count; // 按事件数计丢弃
+      }
+    }
+  }
+}
@@ -0,0 +1,195 @@
+import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
+import { Transport } from "../src/transport";
+import { mergeConfig } from "../src/config";
+
+// uuid v4 格式（crypto.randomUUID 或 Math.random 兜底均为该形状）
+const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
+
+const res = (status: number) => ({ ok: status < 400, status }) as unknown as Response;
+
+// 与 mergeConfig 默认值无关的测试专用 cfg（各用例按需覆盖）
+function mkCfg(over: Record<string, unknown> = {}) {
+  return {
+    siteId: "site1",
+    endpoint: "/api/v1/collect",
+    batchSize: 3,
+    flushMs: 15000,
+    maxRetry: 3,
+    queueMax: 50,
+    ...over,
+  };
+}
+
+let fetchMock: ReturnType<typeof vi.fn>;
+
+describe("transport", () => {
+  beforeEach(() => {
+    localStorage.clear(); // identity 持久化隔离
+    vi.useFakeTimers(); // 默认连 Date 一起 fake
+    fetchMock = vi.fn().mockResolvedValue(res(200));
+    vi.stubGlobal("fetch", fetchMock);
+  });
+  afterEach(() => {
+    vi.unstubAllGlobals(); // 还原 fetch / navigator 桩
+    vi.useRealTimers();
+    delete (globalThis as { __ctk_ctx?: unknown }).__ctk_ctx;
+  });
+
+  it("满 batchSize 即发：body 含 batch_id/site_id/visitor_id/session_id/sent_at/events", () => {
+    const t = new Transport(mkCfg({ batchSize: 3 }));
+    t.push({ event_id: "e1" });
+    t.push({ event_id: "e2" });
+    expect(fetchMock).not.toHaveBeenCalled();
+    t.push({ event_id: "e3" });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    const [url, init] = fetchMock.mock.calls[0];
+    expect(url).toBe("/api/v1/collect");
+    expect(init.method).toBe("POST");
+    expect(init.keepalive).toBe(true);
+    const body = JSON.parse(init.body);
+    expect(body.batch_id).toMatch(UUID_RE);
+    expect(body.site_id).toBe("site1");
+    expect(body.visitor_id).toMatch(UUID_RE);
+    expect(body.session_id).toMatch(UUID_RE);
+    expect(body.sent_at).toBe(new Date().toISOString()); // fake Date 下两处取值一致
+    expect(body.sent_at).toMatch(/^\d{4}-\d{2}-\d{2}T/);
+    expect(body.events.map((e: { event_id: string }) => e.event_id)).toEqual([
+      "e1",
+      "e2",
+      "e3",
+    ]);
+    expect("context" in body).toBe(false); // __ctk_ctx 未注入时序列化后无 context 键
+  });
+
+  it("不足 batchSize 时由 flushMs 定时器触发", async () => {
+    const t = new Transport(mkCfg({ batchSize: 10 }));
+    t.push({ event_id: "e1" });
+    expect(fetchMock).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(14999);
+    expect(fetchMock).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(1);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect(JSON.parse(fetchMock.mock.calls[0][1].body).events).toHaveLength(1);
+  });
+
+  it("context 透传：__ctk_ctx 原样进 body.context", () => {
+    const t = new Transport(mkCfg({ batchSize: 1 }));
+    (globalThis as { __ctk_ctx?: unknown }).__ctk_ctx = { device_type: "PC" };
+    t.push({ event_id: "e1" });
+    const body = JSON.parse(fetchMock.mock.calls[0][1].body);
+    expect(body.context).toEqual({ device_type: "PC" }); // 引用透传而非拷贝拼接
+  });
+
+  it("500 失败退避重试后成功：retry 归零，body 逐字节不变（同 event_id 重放）", async () => {
+    fetchMock.mockResolvedValueOnce(res(500)).mockResolvedValueOnce(res(200));
+    const t = new Transport(mkCfg({ batchSize: 2 }));
+    t.push({ event_id: "e1" });
+    t.push({ event_id: "e2" });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(2000); // 第 1 次退避 500*2^1*jitter ∈ [1000,1400]
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(fetchMock.mock.calls[1][1].body).toBe(fetchMock.mock.calls[0][1].body);
+    expect((t as unknown as { retry: number }).retry).toBe(0);
+  });
+
+  it("400 不重试（参数错，重试无意义）", async () => {
+    fetchMock.mockResolvedValueOnce(res(400));
+    const t = new Transport(mkCfg({ batchSize: 1 }));
+    t.push({ event_id: "e1" });
+    await vi.advanceTimersByTimeAsync(10000);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    expect((t as unknown as { retry: number }).retry).toBe(0);
+  });
+
+  it("403 不重试（权限错，重试只会掩盖配置问题）", async () => {
+    fetchMock.mockResolvedValueOnce(res(403));
+    const t = new Transport(mkCfg({ batchSize: 1 }));
+    t.push({ event_id: "e1" });
+    await vi.advanceTimersByTimeAsync(10000);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("429 要重试（限流是暂态）", async () => {
+    fetchMock.mockResolvedValueOnce(res(429)).mockResolvedValueOnce(res(200));
+    const t = new Transport(mkCfg({ batchSize: 1 }));
+    t.push({ event_id: "e1" });
+    await vi.advanceTimersByTimeAsync(2000);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(t.dropped).toBe(0);
+  });
+
+  it("重试耗尽按本批事件数计丢弃", async () => {
+    fetchMock.mockResolvedValue(res(500)); // 恒 500
+    const t = new Transport(mkCfg({ batchSize: 2, maxRetry: 1 }));
+    t.push({ event_id: "e1" });
+    t.push({ event_id: "e2" });
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    await vi.advanceTimersByTimeAsync(2000); // 第 1 次重试（∈[1000,1400]）后再失败
+    expect(fetchMock).toHaveBeenCalledTimes(2); // 初发 1 + 重试 1，共 2 次
+    expect(t.dropped).toBe(2); // 按入批事件数计，而非按批计
+    expect((t as unknown as { retry: number }).retry).toBe(0); // 归零供下批复用
+  });
+
+  it("queueMax 溢出丢最旧", () => {
+    const t = new Transport(mkCfg({ batchSize: 100, queueMax: 3, flushMs: 60000 }));
+    ["e1", "e2", "e3", "e4"].forEach((id) => t.push({ event_id: id }));
+    expect(t.dropped).toBe(1);
+    t.flush();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const body = JSON.parse(fetchMock.mock.calls[0][1].body);
+    expect(body.events.map((e: { event_id: string }) => e.event_id)).toEqual([
+      "e2",
+      "e3",
+      "e4",
+    ]); // 留最新 3 条
+  });
+
+  it("flush(true) 走 sendBeacon（Blob/text/plain），不调 fetch", async () => {
+    // happy-dom 无 navigator.sendBeacon——整对象替换并测后还原（unstubAllGlobals）
+    const beacon = vi.fn().mockReturnValue(true);
+    vi.stubGlobal("navigator", { sendBeacon: beacon });
+    const t = new Transport(mkCfg({ batchSize: 10 }));
+    t.push({ event_id: "e1" });
+    t.flush(true);
+
+    expect(fetchMock).not.toHaveBeenCalled();
+    expect(beacon).toHaveBeenCalledTimes(1);
+    const [url, blob] = beacon.mock.calls[0];
+    expect(url).toBe("/api/v1/collect");
+    expect(blob).toBeInstanceOf(Blob);
+    expect(blob.type).toBe("text/plain");
+    expect(JSON.parse(await blob.text()).events).toEqual([{ event_id: "e1" }]);
+  });
+
+  it("fetch 同步抛异常不冒泡到 push 调用方", async () => {
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(() => {
+        throw new Error("network gone");
+      })
+    );
+    const t = new Transport(mkCfg({ batchSize: 1, maxRetry: 0 }));
+    expect(() => t.push({ event_id: "e1" })).not.toThrow(); // try/catch 全包红线
+    await vi.advanceTimersByTimeAsync(100);
+    expect(t.dropped).toBe(1); // maxRetry=0 耗尽 → 按事件数丢弃
+  });
+});
+
+describe("mergeConfig", () => {
+  it("缺省填充默认值，显式项覆盖", () => {
+    expect(mergeConfig({ siteId: "s", endpoint: "/e" })).toEqual({
+      siteId: "s",
+      endpoint: "/e",
+      batchSize: 10,
+      flushMs: 15000,
+      maxRetry: 3,
+      queueMax: 200,
+    });
+    expect(mergeConfig({ siteId: "s", endpoint: "/e", batchSize: 5, flushMs: 1000 })).toMatchObject(
+      { batchSize: 5, flushMs: 1000, maxRetry: 3, queueMax: 200 }
+    );
+  });
+});