trade-message-center/apps/server/test/observation-batcher.test.ts

// 验证 OneTalk 观察批次的边界与失败屏障

import assert from "node:assert/strict";
import test from "node:test";

import { ONETALK_PROTOCOL_VERSION } from "@trade-message-center/onetalk-contract";
import type {
    OneTalkMessage,
    OneTalkMessageObservedFrame,
    OneTalkMessagesObservedFrame,
    OneTalkPluginScope,
} from "@trade-message-center/onetalk-contract";

import {
    ObservationBatcher,
    type QueuedObservationFrame,
} from "../src/websocket/plugin/flows/observation-batcher.ts";
import type {
    OneTalkCommitGuard,
    OneTalkObservationInput,
    OneTalkObservationResult,
    OneTalkService,
    OneTalkSourceContext,
} from "../src/onetalk/index.ts";

const pluginScope: OneTalkPluginScope = {
    channelAccountId: "account-1",
    deviceId: "device-1",
};

const sourceContext: OneTalkSourceContext = {
    binding: "binding-1",
    mindUserId: "mind-user-1",
    workspaceId: "workspace-1",
    channelAccountId: pluginScope.channelAccountId,
    deviceId: pluginScope.deviceId,
};

const message = (messageId: string): OneTalkMessage => ({
    messageId,
    conversationId: "conversation-1",
    senderId: "sender-1",
    direction: "received",
    sentAtMs: 1_700_000_000_000,
    content: { version: 1, kind: "text", text: messageId },
    participantIds: ["sender-1", "login-1"],
    readStatus: 0,
    messageStatus: 1,
    unreadCount: 0,
});

const guard: OneTalkCommitGuard = { assertValid: () => {} };

const singleFrame = (requestId: string, observed: OneTalkMessage): OneTalkMessageObservedFrame => ({
    protocolVersion: ONETALK_PROTOCOL_VERSION,
    connectionType: "plugin",
    type: "message.observed",
    requestId,
    scope: pluginScope,
    payload: { observationSource: "live", message: observed },
});

const arrayFrame = (
    requestId: string,
    messages: OneTalkMessage[],
): OneTalkMessagesObservedFrame => ({
    protocolVersion: ONETALK_PROTOCOL_VERSION,
    connectionType: "plugin",
    type: "messages.observed",
    requestId,
    scope: pluginScope,
    payload: { observationSource: "history", messages },
});

const queued = (
    frame: OneTalkMessageObservedFrame | OneTalkMessagesObservedFrame,
): QueuedObservationFrame => ({
    frame,
    items:
        frame.type === "message.observed"
            ? [{ observationSource: "live", message: frame.payload.message }]
            : frame.payload.messages.map((observed) => ({
                  observationSource: "history",
                  message: observed,
              })),
    guard,
    policyEpoch: 1,
});

const acceptedResults = (observations: OneTalkObservationInput[]): OneTalkObservationResult[] =>
    observations.map((observation) => ({ status: "accepted", message: observation.message }));

const createService = (observeMessages: OneTalkService["observeMessages"]): OneTalkService =>
    ({ observeMessages }) as unknown as OneTalkService;

test("keeps a 95-item window plus a 10-item array in one flush", async () => {
    const calls: OneTalkObservationInput[][] = [];
    const deliveries: Array<{
        frames: QueuedObservationFrame[];
        results: OneTalkObservationResult[];
    }> = [];
    const batcher = new ObservationBatcher({
        service: createService(async (_context, observations) => {
            calls.push(observations);
            return acceptedResults(observations);
        }),
        context: sourceContext,
        windowMs: 60_000,
        deliver: async (frames, results) => {
            deliveries.push({ frames, results });
        },
        onFailure: () => assert.fail("the batch should not fail"),
    });

    for (let index = 0; index < 95; index += 1) {
        batcher.enqueue(queued(singleFrame(`single-${index}`, message(`single-${index}`))));
    }
    const arrayMessages = Array.from({ length: 10 }, (_, index) => message(`array-${index}`));
    batcher.enqueue(queued(arrayFrame("array-1", arrayMessages)));

    await batcher.flush();

    assert.equal(calls.length, 1);
    assert.equal(calls[0].length, 105);
    assert.deepEqual(
        deliveries[0].frames.map((frame) => frame.items.length),
        [...Array(95).fill(1), 10],
    );
    assert.equal(deliveries[0].results.length, 105);
    batcher.discard();
});

test("puts frames arriving after timer capture into the next serialized flush", async () => {
    const calls: string[][] = [];
    const deliveryOrder: string[] = [];
    let markFirstStarted: (() => void) | undefined;
    const firstStarted = new Promise<void>((resolve) => {
        markFirstStarted = resolve;
    });
    let releaseFirst: (() => void) | undefined;
    const firstRelease = new Promise<void>((resolve) => {
        releaseFirst = resolve;
    });
    const batcher = new ObservationBatcher({
        service: createService(async (_context, observations) => {
            const ids = observations.map((observation) => observation.message.messageId);
            calls.push(ids);
            if (calls.length === 1) {
                markFirstStarted?.();
                await firstRelease;
            }
            return acceptedResults(observations);
        }),
        context: sourceContext,
        windowMs: 1,
        maxItems: 100,
        deliver: async (frames) => {
            deliveryOrder.push(
                ...frames.flatMap((frame) => frame.items.map((item) => item.message.messageId)),
            );
        },
        onFailure: (error) => assert.fail(error instanceof Error ? error : String(error)),
    });

    batcher.enqueue(queued(singleFrame("first", message("first"))));
    await firstStarted;
    batcher.enqueue(queued(singleFrame("second", message("second"))));
    releaseFirst?.();
    await batcher.flush();

    assert.deepEqual(calls, [["first"], ["second"]]);
    assert.deepEqual(deliveryOrder, ["first", "second"]);
    batcher.discard();
});

test("does not deliver a failed transaction or retain its queued frames", async () => {
    const failure = new Error("database failed");
    let deliveries = 0;
    let failures = 0;
    const batcher = new ObservationBatcher({
        service: createService(async () => {
            throw failure;
        }),
        context: sourceContext,
        windowMs: 60_000,
        deliver: async () => {
            deliveries += 1;
        },
        onFailure: (error) => {
            failures += 1;
            assert.equal(error, failure);
        },
    });

    batcher.enqueue(queued(singleFrame("failed", message("failed"))));
    await assert.rejects(() => batcher.flush(), failure);

    assert.equal(failures, 1);
    assert.equal(deliveries, 0);
    batcher.discard();
});