YBF/trade-message-center
1
0
Fork 0
mirror of https://github.com/sinanyuntu/trade-message-center.git synced 2026-09-17 13:22:11 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: commit rendered-card supplements independently

Browse Source
This commit is contained in:
YBF
2026-09-15 17:25:36 +08:00
parent c19e709c93
commit 971f84bb3d
19 changed files with 749 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.trellis/spec/chrome-extension/frontend/onetalk/runtime-sync.md
+4
View File
@@ -94,6 +94,7 @@ Bright WebSocket
- `message-observer/send-observation.ts` 保持发送 pending、候选 ID、timeout 和 settle 的唯一 owner；命令侧通过 execute/executeImage/executeFile 使用。`image-send.ts` 保持原生媒体上传拦截器和关联状态，不复制或搬迁模块级状态容器。
- profile observer 可将 `syncData` 的 direct profiles 再次发布；它不维护 `seen`、baseline 或其他上传资格状态。snapshot 只读取当前已加载资料，collect 只读取一个明确 conversationId；两者都复用已安装 observer，命令只返回统计，不能再二次 publish。
- 新 installer、publisher 和 dispatcher 不拥有 Map、Set、timer、账本或连接。消息投递、profile/buyer 账本、ACK、页面请求关联及 Bright 生命周期继续归各原 Service Worker owner。
- `card-observer/entry.ts` 以 MutationObserver 批次作为 rendered-card 的唯一主动触发。`.message-item-wrapper` 只表示 React 已挂载，不表示 Fiber/template 已齐备：首次扫描及该 wrapper 子树的 `childList`、attributes 或 text 变化都必须立即重新读取；只有 reader 成功产生 exact 白名单 observation 后按 `conversationId + messageId + fingerprint` 去重。reader 返回 `null` 不写 bridge/ledger，并等待下一次该 wrapper 的 DOM 变化；不得全局轮询、用 DOM 文案兜底或把 raw Fiber/props 跨 MAIN。
#### Validation & Error Matrix
@@ -108,6 +109,8 @@ Bright WebSocket
| 未知 action | rejected_before_send/invalid_request，不访问 SDK 或 snapshot |
| send 的 SDK 本地接受、超时或歧义 | 保持既有发送 SOP，不增加重试或假成功 |
| 历史命令返回 batch | 使用同一 publisher，不新建发送关联器或上报管线 |
| rendered-card wrapper 先挂载、后补齐 Fiber/template | 每个相关 DOM 变化批次立即重新读取；成功只发布一次 |
| rendered-card reader 仍无完整白名单模板 | 不发布、不写 ledger；仅等待该 wrapper 的下一次子树变化 |
#### Good / Base / Bad Cases
@@ -119,6 +122,7 @@ Bright WebSocket
- `onetalk-page-flow-boundaries.test.js` 覆盖 publisher 双消费顺序、诊断-only/空批次、snapshot/collect exact payload、targeted collect、未知动作与实际页面入口安装顺序。
- `onetalk-send-page.test.js` 保留 send/history 的输入、结果、tooltip 和发布行为；`onetalk-send-observation.test.js` 与媒体测试保留唯一确认、timeout/异常及不重试断言。
- `onetalk-rendered-card-observer.test.js` 覆盖 wrapper 首次读取失败、后续子树/属性/文本变化后立即重新读取并只生成一次 approved observation。
- 对构建后的 MAIN IIFE 执行 `scripts/verify-release-bundle.mjs`，验证实际 shared observation 驱动消息上报及单次发送确认；这属于无网络 Node VM 冒烟，真实 Chromium/OneTalk/Bright 联调需单独标记。
#### Wrong vs Correct
apps/chrome-extension/src/onetalk/main-page/card-observer/entry.ts
+53 -14
View File
@@ -1,10 +1,12 @@
// 安装仅以真实 mount 为触发的渲染卡片 observer。
// 安装渲染卡片 observer：DOM mutation 即时读取，另以当前列表周期重扫补足 Fiber 延迟就绪。
import type { OneTalkRenderedCardObservation } from "@trade-message-center/onetalk-contract";
import type { OneTalkPageWindow } from "../model.ts";
import type { OneTalkPageRenderedCardBaseEvidence } from "../../page-bridge/model.ts";
import { readOneTalkRenderedCard } from "./react-card-reader.ts";
const RENDERED_CARD_RESCAN_INTERVAL_MS = 1_000;
const messageListRootFor = (
document: NonNullable<OneTalkPageWindow["document"]>,
): Element | null => {
@@ -18,7 +20,14 @@ const messageListRootFor = (
return firstMessage.parentElement;
};
/** 安装仅以真实 mount 为触发的 Fiber card reader；未验证类别一律不上传。 */
const renderedCardWrapperFor = (node: Node): Element | null => {
const element = node instanceof Element ? node : node.parentElement;
return element?.matches(".message-item-wrapper")
? element
: (element?.closest(".message-item-wrapper") ?? null);
};
/** 安装 Fiber card reader；未验证类别一律不上传。 */
export const installOneTalkRenderedCardObserver = (
pageWindow: OneTalkPageWindow,
sink: (
@@ -28,17 +37,20 @@ export const installOneTalkRenderedCardObserver = (
): void => {
const document = pageWindow.document;
if (!document || !pageWindow.MutationObserver) return;
const seen = new WeakSet<Element>();
const observedCardKeys = new Set<string>();
let disposed = false;
const inspect = (wrapper: Element): void => {
if (seen.has(wrapper)) return;
seen.add(wrapper);
setTimeout(() => {
if (disposed) return;
const read = readOneTalkRenderedCard(pageWindow, wrapper);
if (!read) return;
sink([read.observation], [read.baseEvidence]);
}, 0);
if (disposed) return;
const read = readOneTalkRenderedCard(pageWindow, wrapper);
if (!read) return;
const observationKey = JSON.stringify([
read.observation.conversationId,
read.observation.messageId,
read.observation.contentFingerprint,
]);
if (observedCardKeys.has(observationKey)) return;
observedCardKeys.add(observationKey);
sink([read.observation], [read.baseEvidence]);
};
let root: Element | null = null;
let observingDocument = false;
@@ -50,12 +62,17 @@ export const installOneTalkRenderedCardObserver = (
}
for (const record of records) {
if (root && record.target !== root && !root.contains(record.target)) continue;
const wrappers = new Set<Element>();
const targetWrapper = renderedCardWrapperFor(record.target);
if (targetWrapper) wrappers.add(targetWrapper);
for (const node of Array.from(record.addedNodes)) {
if (!(node instanceof Element)) continue;
if (node.matches(".message-item-wrapper")) inspect(node);
const wrapper = renderedCardWrapperFor(node);
if (wrapper) wrappers.add(wrapper);
for (const wrapper of Array.from(node.querySelectorAll(".message-item-wrapper")))
inspect(wrapper);
wrappers.add(wrapper);
}
for (const wrapper of wrappers) inspect(wrapper);
}
});
const bind = (): void => {
@@ -72,14 +89,36 @@ export const installOneTalkRenderedCardObserver = (
observingDocument = false;
for (const wrapper of Array.from(root.querySelectorAll(".message-item-wrapper")))
inspect(wrapper);
observer.observe(root, { childList: true, subtree: true });
observer.observe(root, {
attributes: true,
characterData: true,
childList: true,
subtree: true,
});
if (root.parentElement) observer.observe(root.parentElement, { childList: true });
};
const rescanCurrentMessageList = (): void => {
if (disposed) return;
if (messageListRootFor(document) !== root) {
bind();
return;
}
if (!root) return;
for (const wrapper of Array.from(root.querySelectorAll(".message-item-wrapper")))
inspect(wrapper);
};
bind();
const rescanTimer = globalThis.setInterval(
rescanCurrentMessageList,
RENDERED_CARD_RESCAN_INTERVAL_MS,
);
pageWindow.addEventListener("popstate", bind);
pageWindow.addEventListener("hashchange", bind);
pageWindow.addEventListener("pagehide", () => {
disposed = true;
globalThis.clearInterval(rescanTimer);
observer.disconnect();
});
};
apps/chrome-extension/src/onetalk/service-worker/configured-sync-session.ts
-1
View File
@@ -274,7 +274,6 @@ export class OneTalkConfiguredSyncSession {
renderedCard = createOneTalkRenderedCardCoordinator({
scope: pluginScope,
ledger: nextRenderedCardLedgerStore,
syncStore: nextStore,
bright,
createRequestId,
onError: reportCurrentError,
apps/chrome-extension/src/onetalk/service-worker/rendered-card-coordinator.ts
+1 -20
View File
@@ -8,7 +8,6 @@ import {
} from "@trade-message-center/onetalk-contract";
import type { OneTalkBrightClient, OneTalkBrightClientState } from "./transport/bright-client.ts";
import type { OneTalkRenderedCardLedgerStore } from "./storage.ts";
import type { OneTalkSyncStore } from "./storage.ts";
import type { OneTalkPageRenderedCardBaseEvidence } from "../page-bridge/model.ts";
export type OneTalkRenderedCardCoordinator = {
@@ -18,17 +17,15 @@ export type OneTalkRenderedCardCoordinator = {
) => Promise<void>;
handleFrame: (frame: OneTalkFrame) => void;
handleStatus: (state: OneTalkBrightClientState) => void;
handleBaseCandidateProgress: () => void;
handlePageReady: () => Promise<void>;
handlePageDisconnected: () => void;
dispose: () => void;
};
/** 创建独立卡片协调器；基础消息确认前不允许补全上传。 */
/** 创建独立卡片协调器；补全账本不等待基础消息 ACK。 */
export const createOneTalkRenderedCardCoordinator = (options: {
scope: OneTalkPluginScope;
ledger: OneTalkRenderedCardLedgerStore;
syncStore: OneTalkSyncStore;
bright: OneTalkBrightClient;
createRequestId?: (kind: string) => string;
onError?: (error: unknown) => void;
@@ -54,18 +51,6 @@ export const createOneTalkRenderedCardCoordinator = (options: {
try {
for (const record of await options.ledger.listPending(options.scope.channelAccountId)) {
if (disposed || !options.bright.isOnline() || sent.has(record.key)) continue;
const base = await options.syncStore.getCandidate(
record.channelAccountId,
record.conversationId,
record.messageId,
);
if (
disposed ||
base?.status !== "confirmed" ||
base.message.direction !== record.baseDirection ||
base.message.sentAtMs !== record.baseSentAtMs
)
continue;
const requestId = createRequestId("observed");
const pendingSend = {
requestId,
@@ -171,10 +156,6 @@ export const createOneTalkRenderedCardCoordinator = (options: {
}
void flush();
},
handleBaseCandidateProgress: () => {
if (disposed) return;
setTimeout(() => void flush(), 0);
},
handlePageReady: flush,
handlePageDisconnected: () => undefined,
dispose: () => {
apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts
+5 -12
View File
@@ -37,10 +37,7 @@ export const createOneTalkServiceWorkerFrameRouter = (options: {
sync: Pick<OneTalkSyncEngine, "handleServerFrame">;
profile?: Pick<OneTalkContactProfileCoordinator, "handleFrame">;
buyer?: Pick<OneTalkBuyerFactCoordinator, "handleFrame">;
renderedCard?: Pick<
OneTalkRenderedCardCoordinator,
"handleFrame" | "handleBaseCandidateProgress"
>;
renderedCard?: Pick<OneTalkRenderedCardCoordinator, "handleFrame">;
send?: Pick<OneTalkSendCommandFlow, "handle">;
rebuild?: Pick<OneTalkHistoryRebuildFlow, "handle">;
}): OneTalkServiceWorkerFrameRouter => {
@@ -48,14 +45,10 @@ export const createOneTalkServiceWorkerFrameRouter = (options: {
defineOneTalkBusinessRoute("anchor.snapshot", (frame) =>
options.sync.handleServerFrame(frame),
),
defineOneTalkBusinessRoute("message.ack", (frame) => {
options.sync.handleServerFrame(frame);
options.renderedCard?.handleBaseCandidateProgress();
}),
defineOneTalkBusinessRoute("messages.ack", (frame) => {
options.sync.handleServerFrame(frame);
options.renderedCard?.handleBaseCandidateProgress();
}),
defineOneTalkBusinessRoute("message.ack", (frame) => options.sync.handleServerFrame(frame)),
defineOneTalkBusinessRoute("messages.ack", (frame) =>
options.sync.handleServerFrame(frame),
),
defineOneTalkBusinessRoute("conversation.ack", (frame) =>
options.sync.handleServerFrame(frame),
),
apps/chrome-extension/src/onetalk/service-worker/storage.ts
+2 -2
View File
@@ -144,9 +144,9 @@ export type OneTalkRenderedCardLedgerRecord = {
content: OneTalkRenderedCardContent;
contentFingerprint: string;
observedAtMs: number;
/** MAIN Fiber identity cross-check retained only for restart-safe candidate validation. */
/** MAIN Fiber identity evidence retained with the independently uploaded card fact. */
baseDirection: "received" | "sent";
/** MAIN Fiber sendTime cross-check retained only for restart-safe candidate validation. */
/** MAIN Fiber sendTime evidence retained with the independently uploaded card fact. */
baseSentAtMs: number;
status: "pending_ack" | "confirmed" | "rejected";
requestId?: string;
apps/chrome-extension/src/onetalk/service-worker/sync-runtime.ts
-1
View File
@@ -79,7 +79,6 @@ export const createOneTalkServiceWorkerSyncRuntime = (
? createOneTalkRenderedCardCoordinator({
scope: options.scope,
ledger: options.renderedCardLedgerStore,
syncStore: options.store,
bright: options.bright,
...(options.createRequestId === undefined
? {}
apps/chrome-extension/test/onetalk-rendered-card-coordinator.test.js
+2 -61
View File
@@ -39,10 +39,9 @@ const baseEvidence = [
},
];
test("writes the card ledger before send, gates on base confirmation, resends after reconnect, and exact-matches ACK", async () => {
test("writes the card ledger before immediate send, resends after reconnect, and exact-matches ACK", async () => {
const timeline = [];
const records = new Map();
let baseStatus = "pending_ack";
const acknowledgements = [];
const ledger = {
observe: async ({ channelAccountId, observation: value, baseEvidence: evidence }) => {
@@ -102,21 +101,11 @@ test("writes the card ledger before send, gates on base confirmation, resends af
const coordinator = createOneTalkRenderedCardCoordinator({
scope,
ledger,
syncStore: {
getCandidate: async () => ({
status: baseStatus,
message: { direction: "received", sentAtMs: 99 },
}),
},
bright,
createRequestId: () => "card-request",
});
await coordinator.observe([observation], baseEvidence);
assert.deepEqual(timeline, ["write"]);
baseStatus = "confirmed";
coordinator.handleStatus({ status: "authenticated", permissions: [] });
await new Promise((resolve) => setImmediate(resolve));
assert.equal(frames.length, 1);
assert.equal(frames[0].type, "rendered.card.observed");
assert.deepEqual(timeline, ["write", "send"]);
@@ -155,49 +144,7 @@ test("writes the card ledger before send, gates on base confirmation, resends af
assert.equal(records.get(key).status, "confirmed");
});
test("does not send when MAIN base evidence differs from the confirmed candidate", async () => {
const writes = [];
const coordinator = createOneTalkRenderedCardCoordinator({
scope,
ledger: {
observe: async ({ channelAccountId, observation: value, baseEvidence: evidence }) => ({
key,
channelAccountId,
...value,
baseDirection: evidence.direction,
baseSentAtMs: evidence.sentAtMs,
status: "pending_ack",
firstObservedAt: 1,
updatedAt: 1,
}),
listPending: async () => [
{
key,
channelAccountId: scope.channelAccountId,
...observation,
baseDirection: "received",
baseSentAtMs: 99,
status: "pending_ack",
firstObservedAt: 1,
updatedAt: 1,
},
],
markSent: async () => assert.fail("must not persist a send request"),
markAcknowledged: async () => false,
},
syncStore: {
getCandidate: async () => ({
status: "confirmed",
message: { direction: "sent", sentAtMs: 99 },
}),
},
bright: { isOnline: () => true, send: (frame) => writes.push(frame) },
});
await coordinator.observe([observation], baseEvidence);
assert.deepEqual(writes, []);
});
test("recovers a pending record after worker restart using durable base cross-check metadata", async () => {
test("recovers a pending record after worker restart without a base candidate", async () => {
const records = new Map([
[
key,
@@ -227,12 +174,6 @@ test("recovers a pending record after worker restart using durable base cross-ch
},
markAcknowledged: async () => false,
},
syncStore: {
getCandidate: async () => ({
status: "confirmed",
message: { direction: "received", sentAtMs: 99 },
}),
},
bright: { isOnline: () => true, send: (frame) => frames.push(frame) || true },
createRequestId: () => "recovered-request",
});
apps/chrome-extension/test/onetalk-rendered-card-observer.test.js
+423
View File
@@ -37,6 +37,7 @@ test("installs a mutation observer before the first message wrapper exists", ()
assert.equal(observers.length, 1);
assert.equal(observers[0].targets.length, 1);
assert.equal(listeners.has("pagehide"), true);
listeners.get("pagehide")();
});
test("rebinds the scoped message observer when its list root is replaced", () => {
@@ -79,3 +80,425 @@ test("rebinds the scoped message observer when its list root is replaced", () =>
assert.ok(observers[0].targets.includes(secondRoot));
listeners.get("pagehide")();
});
test("reads a card immediately when its wrapper subtree changes", async () => {
const originalElement = globalThis.Element;
class TestElement {
matches() {
return false;
}
closest() {
return this.parentWrapper ?? null;
}
querySelectorAll() {
return [];
}
}
globalThis.Element = TestElement;
try {
const observers = [];
const listeners = new Map();
const templateFiber = { memoizedProps: {} };
const wrapper = new TestElement();
const root = {
parentElement: null,
querySelectorAll: (selector) => {
assert.equal(selector, ".message-item-wrapper");
return [wrapper];
},
contains: (node) => node === wrapper,
};
wrapper.parentElement = root;
wrapper.parentWrapper = wrapper;
wrapper.__reactFiber$fixture = {
memoizedProps: {
itemData: {
messageId: "message-1",
conversationCode: "conversation-1",
messageType: "rec",
sendTime: 1_700_000_000_000,
msgType: 10010,
originalData: { cardType: 9 },
},
},
child: templateFiber,
};
const page = {
document: {
querySelectorAll: (selector) => {
if (selector === ".message-item-wrapper") return [wrapper];
if (selector === ".contact-item-container.selected[data-cid]") {
return [{ getAttribute: () => "conversation-1" }];
}
assert.fail(`unexpected selector: ${selector}`);
},
},
MutationObserver: class {
constructor(callback) {
this.callback = callback;
observers.push(this);
}
observe() {}
disconnect() {}
},
addEventListener: (type, listener) => listeners.set(type, listener),
};
const observations = [];
installOneTalkRenderedCardObserver(page, (next) => observations.push(...next));
templateFiber.memoizedProps = {
data: {
cardTitle: "Order #1",
productInfoList: [
{ productName: "Widget", productImage: "https://img.alicdn.com/item.jpg" },
],
orderStatusText: "Paid",
shouldPayAmount: "$10",
shippingAddress: "Hangzhou",
},
};
observers[0].callback([{ target: wrapper, addedNodes: [new TestElement()] }]);
assert.deepEqual(
observations.map((observation) => observation.content.kind),
["rendered_order"],
);
listeners.get("pagehide")();
} finally {
if (originalElement === undefined) delete globalThis.Element;
else globalThis.Element = originalElement;
}
});
test("re-reads a wrapper immediately when React changes text inside it", async () => {
const originalElement = globalThis.Element;
class TestElement {
matches() {
return false;
}
closest() {
return this.parentWrapper ?? null;
}
querySelectorAll() {
return [];
}
}
globalThis.Element = TestElement;
try {
const observers = [];
const observeCalls = [];
const listeners = new Map();
const templateFiber = { memoizedProps: {} };
const wrapper = new TestElement();
const textNode = { parentElement: wrapper };
const root = {
parentElement: null,
querySelectorAll: (selector) => {
assert.equal(selector, ".message-item-wrapper");
return [wrapper];
},
contains: (node) => node === wrapper || node?.parentElement === wrapper,
};
wrapper.parentElement = root;
wrapper.parentWrapper = wrapper;
wrapper.__reactFiber$fixture = {
memoizedProps: {
itemData: {
messageId: "message-1",
conversationCode: "conversation-1",
messageType: "rec",
sendTime: 1_700_000_000_000,
msgType: 10010,
originalData: { cardType: 9 },
},
},
child: templateFiber,
};
const page = {
document: {
querySelectorAll: (selector) => {
if (selector === ".message-item-wrapper") return [wrapper];
if (selector === ".contact-item-container.selected[data-cid]") {
return [{ getAttribute: () => "conversation-1" }];
}
assert.fail(`unexpected selector: ${selector}`);
},
},
MutationObserver: class {
constructor(callback) {
this.callback = callback;
observers.push(this);
}
observe(target, options) {
observeCalls.push({ target, options });
}
disconnect() {}
},
addEventListener: (type, listener) => listeners.set(type, listener),
};
const observations = [];
installOneTalkRenderedCardObserver(page, (next) => observations.push(...next));
templateFiber.memoizedProps = {
data: {
cardTitle: "Order #1",
productInfoList: [
{ productName: "Widget", productImage: "https://img.alicdn.com/item.jpg" },
],
orderStatusText: "Paid",
shouldPayAmount: "$10",
shippingAddress: "Hangzhou",
},
};
observers[0].callback([{ target: textNode, addedNodes: [] }]);
assert.ok(
observeCalls.some(
({ target, options }) =>
target === root &&
options.attributes === true &&
options.characterData === true,
),
);
assert.deepEqual(
observations.map((observation) => observation.content.kind),
["rendered_order"],
);
listeners.get("pagehide")();
} finally {
if (originalElement === undefined) delete globalThis.Element;
else globalThis.Element = originalElement;
}
});
test("re-reads a virtualized wrapper for a new card but deduplicates the same card fingerprint", async () => {
const originalElement = globalThis.Element;
class TestElement {
matches() {
return false;
}
closest() {
return this.parentWrapper ?? null;
}
querySelectorAll() {
return [];
}
}
globalThis.Element = TestElement;
try {
const observers = [];
const listeners = new Map();
const templateFiber = { memoizedProps: {} };
const wrapper = new TestElement();
const root = {
parentElement: null,
querySelectorAll: (selector) => {
assert.equal(selector, ".message-item-wrapper");
return [wrapper];
},
contains: (node) => node === wrapper,
};
wrapper.parentElement = root;
wrapper.parentWrapper = wrapper;
wrapper.__reactFiber$fixture = {
memoizedProps: {
itemData: {
messageId: "message-1",
conversationCode: "conversation-1",
messageType: "rec",
sendTime: 1_700_000_000_000,
msgType: 10010,
originalData: { cardType: 9 },
},
},
child: templateFiber,
};
const page = {
document: {
querySelectorAll: (selector) => {
if (selector === ".message-item-wrapper") return [wrapper];
if (selector === ".contact-item-container.selected[data-cid]") {
return [{ getAttribute: () => "conversation-1" }];
}
assert.fail(`unexpected selector: ${selector}`);
},
},
MutationObserver: class {
constructor(callback) {
this.callback = callback;
observers.push(this);
}
observe() {}
disconnect() {}
},
addEventListener: (type, listener) => listeners.set(type, listener),
};
const observations = [];
const setOrderTemplate = (title) => {
templateFiber.memoizedProps = {
data: {
cardTitle: title,
productInfoList: [
{ productName: "Widget", productImage: "https://img.alicdn.com/item.jpg" },
],
orderStatusText: "Paid",
shouldPayAmount: "$10",
shippingAddress: "Hangzhou",
},
};
};
setOrderTemplate("Order #1");
installOneTalkRenderedCardObserver(page, (next) => observations.push(...next));
await new Promise((resolve) => setTimeout(resolve, 120));
wrapper.__reactFiber$fixture.memoizedProps.itemData.messageId = "message-2";
setOrderTemplate("Order #2");
observers[0].callback([{ target: wrapper, addedNodes: [new TestElement()] }]);
await new Promise((resolve) => setTimeout(resolve, 120));
observers[0].callback([{ target: wrapper, addedNodes: [new TestElement()] }]);
await new Promise((resolve) => setTimeout(resolve, 120));
assert.deepEqual(
observations.map((value) => [value.messageId, value.content.title]),
[
["message-1", "Order #1"],
["message-2", "Order #2"],
],
);
listeners.get("pagehide")();
} finally {
if (originalElement === undefined) delete globalThis.Element;
else globalThis.Element = originalElement;
}
});
test("rescans the current list every second when a card Fiber becomes ready without a DOM mutation", () => {
const originalElement = globalThis.Element;
const originalSetInterval = globalThis.setInterval;
const originalClearInterval = globalThis.clearInterval;
class TestElement {
matches() {
return false;
}
closest() {
return this.parentWrapper ?? null;
}
querySelectorAll() {
return [];
}
}
globalThis.Element = TestElement;
try {
const observers = [];
const intervals = [];
const clearedIntervals = [];
const listeners = new Map();
const templateFiber = { memoizedProps: {} };
const wrapper = new TestElement();
const root = {
parentElement: null,
querySelectorAll: (selector) => {
assert.equal(selector, ".message-item-wrapper");
return [wrapper];
},
contains: (node) => node === wrapper,
};
wrapper.parentElement = root;
wrapper.parentWrapper = wrapper;
wrapper.__reactFiber$fixture = {
memoizedProps: {
itemData: {
messageId: "message-1",
conversationCode: "conversation-1",
messageType: "rec",
sendTime: 1_700_000_000_000,
msgType: 10010,
originalData: { cardType: 9 },
},
},
child: templateFiber,
};
globalThis.setInterval = (callback, delay) => {
intervals.push({ callback, delay });
return intervals.length;
};
globalThis.clearInterval = (interval) => clearedIntervals.push(interval);
const page = {
document: {
querySelectorAll: (selector) => {
if (selector === ".message-item-wrapper") return [wrapper];
if (selector === ".contact-item-container.selected[data-cid]") {
return [{ getAttribute: () => "conversation-1" }];
}
assert.fail(`unexpected selector: ${selector}`);
},
},
MutationObserver: class {
constructor(callback) {
this.callback = callback;
observers.push(this);
}
observe() {}
disconnect() {}
},
addEventListener: (type, listener) => listeners.set(type, listener),
};
const observations = [];
installOneTalkRenderedCardObserver(page, (next) => observations.push(...next));
assert.deepEqual(
intervals.map((interval) => interval.delay),
[1_000],
);
templateFiber.memoizedProps = {
data: {
cardTitle: "Order #1",
productInfoList: [
{ productName: "Widget", productImage: "https://img.alicdn.com/item.jpg" },
],
orderStatusText: "Paid",
shouldPayAmount: "$10",
shippingAddress: "Hangzhou",
},
};
intervals[0].callback();
assert.deepEqual(
observations.map((observation) => observation.content.kind),
["rendered_order"],
);
listeners.get("pagehide")();
assert.deepEqual(clearedIntervals, [1]);
} finally {
if (originalElement === undefined) delete globalThis.Element;
else globalThis.Element = originalElement;
globalThis.setInterval = originalSetInterval;
globalThis.clearInterval = originalClearInterval;
}
});
apps/server/src/onetalk/model.ts
+11 -2
View File
@@ -13,6 +13,7 @@ import type {
OneTalkSyncAnomalyCode,
OneTalkSyncMode,
OneTalkSyncResult,
OneTalkRenderedCardContent,
} from "@trade-message-center/onetalk-contract";
export class OneTalkDatabaseError extends Error {
@@ -65,7 +66,11 @@ export type OneTalkConversationState = {
};
export type OneTalkMessageInsertResult =
| { status: "accepted"; message: OneTalkMessage }
| {
status: "accepted";
message: OneTalkMessage;
renderedCardContent?: OneTalkRenderedCardContent;
}
| { status: "duplicate"; message: OneTalkMessage }
| { status: "rejected"; reason: "conversation_not_discovered" | "history_generation_mismatch" };
@@ -217,7 +222,11 @@ export type OneTalkMessageNormalization =
| { ok: false; anomaly: OneTalkAnomalyInput };
export type OneTalkObservationResult =
| { status: "accepted" | "duplicate"; message: OneTalkMessage }
| {
status: "accepted" | "duplicate";
message: OneTalkMessage;
renderedCardContent?: OneTalkRenderedCardContent;
}
| { status: "anomaly"; anomalyCode: string }
| { status: "rejected"; reason: "conversation_not_discovered" | "history_generation_mismatch" };
apps/server/src/onetalk/rendered-card-model.ts
+15 -4
View File
@@ -8,10 +8,21 @@ import type {
import type { OneTalkCommitGuard } from "./model.ts";
export type OneTalkRenderedCardStoreResult =
| { status: "accepted"; content: OneTalkRenderedCardContent; message: OneTalkCenterMessage }
| { status: "duplicate"; content: OneTalkRenderedCardContent; message: OneTalkCenterMessage }
| { status: "conflict"; content: OneTalkRenderedCardContent; message: OneTalkCenterMessage }
| { status: "rejected"; reason: "base_message_missing" };
| {
status: "accepted";
content: OneTalkRenderedCardContent;
message: OneTalkCenterMessage | null;
}
| {
status: "duplicate";
content: OneTalkRenderedCardContent;
message: OneTalkCenterMessage | null;
}
| {
status: "conflict";
content: OneTalkRenderedCardContent;
message: OneTalkCenterMessage | null;
};
export type OneTalkRenderedCardRepository = {
store: (input: {
apps/server/src/onetalk/rendered-card-repository.ts
+22 -12
View File
@@ -44,6 +44,17 @@ const effectiveMessageFor = (
sentAtMs: row.sentAtMs,
content,
});
const existingBaseMessageFor = async (
transaction: Parameters<Parameters<Database["transaction"]>[0]>[0],
input: Parameters<OneTalkRenderedCardRepository["store"]>[0],
) => {
const base = await transaction
.select()
.from(onetalkMessage)
.where(baseMessageCondition(input))
.limit(1);
return base[0] ?? null;
};
/** 在独立事务中写入首个补全快照，重复只更新时间，冲突只记元数据。 */
const store = async (
@@ -52,15 +63,6 @@ const store = async (
): Promise<OneTalkRenderedCardStoreResult> =>
database.transaction(async (transaction) => {
input.commitGuard?.assertValid();
const base = await transaction
.select()
.from(onetalkMessage)
.where(baseMessageCondition(input))
.limit(1)
.for("update");
input.commitGuard?.assertValid();
const baseMessage = base[0];
if (!baseMessage) return { status: "rejected", reason: "base_message_missing" };
const existing = await transaction
.select()
.from(onetalkRenderedCardContent)
@@ -81,10 +83,14 @@ const store = async (
lastObservedAt: input.receivedAt,
});
input.commitGuard?.assertValid();
const baseMessage = await existingBaseMessageFor(transaction, input);
input.commitGuard?.assertValid();
return {
status: "accepted",
content: input.observation.content,
message: effectiveMessageFor(baseMessage, input.observation.content),
message: baseMessage
? effectiveMessageFor(baseMessage, input.observation.content)
: null,
};
}
const saved = record.renderedCardContent as OneTalkRenderedCardContent;
@@ -94,10 +100,12 @@ const store = async (
.set({ lastObservedAt: input.receivedAt })
.where(keyCondition(input));
input.commitGuard?.assertValid();
const baseMessage = await existingBaseMessageFor(transaction, input);
input.commitGuard?.assertValid();
return {
status: "duplicate",
content: saved,
message: effectiveMessageFor(baseMessage, saved),
message: baseMessage ? effectiveMessageFor(baseMessage, saved) : null,
};
}
await transaction
@@ -109,10 +117,12 @@ const store = async (
})
.where(keyCondition(input));
input.commitGuard?.assertValid();
const baseMessage = await existingBaseMessageFor(transaction, input);
input.commitGuard?.assertValid();
return {
status: "conflict",
content: saved,
message: effectiveMessageFor(baseMessage, saved),
message: baseMessage ? effectiveMessageFor(baseMessage, saved) : null,
};
});
apps/server/src/onetalk/repository.ts
+32 -1
View File
@@ -8,6 +8,7 @@ import type {
OneTalkHistoryGeneration,
OneTalkMessage,
OneTalkObservationSource,
OneTalkRenderedCardContent,
} from "@trade-message-center/onetalk-contract";
import type { Database } from "../database/index.ts";
@@ -95,6 +96,30 @@ const messageCondition = (
);
};
const renderedCardCondition = (
context: OneTalkSourceContext,
conversationId: string,
messageId: string,
) =>
and(
eq(onetalkRenderedCardContent.channelAccountId, context.channelAccountId),
eq(onetalkRenderedCardContent.conversationId, conversationId),
eq(onetalkRenderedCardContent.messageId, messageId),
);
const renderedCardContentFor = async (
transaction: Parameters<Parameters<Database["transaction"]>[0]>[0],
context: OneTalkSourceContext,
message: Pick<OneTalkMessage, "conversationId" | "messageId">,
): Promise<OneTalkRenderedCardContent | null> => {
const rows = await transaction
.select({ content: onetalkRenderedCardContent.renderedCardContent })
.from(onetalkRenderedCardContent)
.where(renderedCardCondition(context, message.conversationId, message.messageId))
.limit(1);
return rows[0]?.content ?? null;
};
const observationTypeFor = (source: OneTalkObservationSource): "new" | "history" => {
return source === "history" ? "history" : "new";
};
@@ -460,7 +485,13 @@ const insertMessage = async (
})
.where(conversationCondition(context, message.conversationId));
guard?.assertValid();
return { status: "accepted", message: toMessage(insertedRow) };
const renderedCardContent = await renderedCardContentFor(transaction, context, message);
guard?.assertValid();
return {
status: "accepted",
message: toMessage(insertedRow),
...(renderedCardContent === null ? {} : { renderedCardContent }),
};
}
guard?.assertValid();
apps/server/src/websocket/plugin/flows/rendered-card-flow.ts
+8 -11
View File
@@ -33,8 +33,8 @@ const scopeFor = (context: OneTalkSourceContext) => ({
});
const ackPayloadFor = (
frame: OneTalkRenderedCardObservedFrame,
status: "accepted" | "duplicate" | "conflict" | "rejected",
rejectionCode?: "base_message_missing" | "content_conflict" | "invalid_card",
status: "accepted" | "duplicate" | "conflict",
rejectionCode?: "content_conflict",
) => ({
conversationId: frame.payload.conversationId,
messageId: frame.payload.messageId,
@@ -85,17 +85,14 @@ export const createOneTalkRenderedCardFlow = (options: {
}
if (options.registry.getCanonicalConnection(socket) !== canonical) return;
if (!authorization.ok) return options.handleAuthorizationFailure(frame, authorization.code);
const ack =
result.status === "rejected"
? ackPayloadFor(frame, "rejected", result.reason)
: ackPayloadFor(
frame,
result.status,
result.status === "conflict" ? "content_conflict" : undefined,
);
const ack = ackPayloadFor(
frame,
result.status,
result.status === "conflict" ? "content_conflict" : undefined,
);
if (!options.sendFrame(createOneTalkRenderedCardAckFrame(frame, ack)))
return options.closeForAcknowledgementFailure();
if (result.status !== "accepted") return;
if (result.status !== "accepted" || result.message === null) return;
guard.assertValid();
await options.registry.publishMessageUpdated({
message: result.message,
apps/server/src/websocket/plugin/flows/send-confirmation-flow.ts
+4 -1
View File
@@ -53,7 +53,10 @@ export const createOneTalkSendConfirmationFlow = (
if (result.status === "accepted") {
pending.commitGuard.assertValid();
await options.registry.publishMessageCreated({
message: toOneTalkCenterMessage(result.message),
message: toOneTalkCenterMessage({
...result.message,
content: result.renderedCardContent ?? result.message.content,
}),
requestId: pending.frame.requestId,
scope: pending.mind.mindScope,
policyEpoch: pending.policyEpoch,
apps/server/src/websocket/plugin/flows/sync-flows.ts
+4 -1
View File
@@ -393,7 +393,10 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
queued.guard.assertValid();
const scope = scopeForContext(request.context);
await options.publishMessageCreated({
message: toOneTalkCenterMessage(result.message),
message: toOneTalkCenterMessage({
...result.message,
content: result.renderedCardContent ?? result.message.content,
}),
requestId: queued.frame.requestId,
scope,
policyEpoch: queued.policyEpoch,
apps/server/test/onetalk-rendered-card-flow.test.ts
+45 -10
View File
@@ -49,16 +49,55 @@ const message = {
content,
};
for (const status of ["accepted", "duplicate", "conflict", "rejected"] as const) {
test("ACKs an independently committed card when its base message has not arrived", async () => {
const sent: OneTalkFrame[] = [];
const published: unknown[] = [];
const canonical = {} as never;
const flow = createOneTalkRenderedCardFlow({
service: {
ingest: async () => ({ status: "accepted" as const, content, message: null }),
},
registry: {
getCanonicalConnection: () => canonical,
publishMessageUpdated: async (input: unknown) => published.push(input),
} as never,
sendFrame: (outbound) => {
sent.push(outbound);
return true;
},
isPolicyCurrent: () => true,
isCommitGuardFailure: () => false,
closeForPause: () => assert.fail("unexpected pause"),
closeForDatabaseFailure: () => assert.fail("unexpected database failure"),
closeForAcknowledgementFailure: () => assert.fail("unexpected ack failure"),
reauthorize: async () => ({ ok: true }),
handleAuthorizationFailure: () => assert.fail("unexpected authorization failure"),
});
await flow.handle({
socket: {} as never,
frame,
context,
guard: { assertValid: () => undefined },
policyEpoch: 1,
canonical,
});
assert.deepEqual(
sent.map((value) => value.type),
["rendered.card.ack"],
);
assert.equal((sent[0]?.payload as { status?: string } | undefined)?.status, "accepted");
assert.deepEqual(published, []);
});
for (const status of ["accepted", "duplicate", "conflict"] as const) {
test(`ACKs rendered-card ${status} and publishes only acceptance`, async () => {
const sent: OneTalkFrame[] = [];
const published: unknown[] = [];
const flow = createOneTalkRenderedCardFlow({
service: {
ingest: async () =>
status === "rejected"
? { status, reason: "base_message_missing" as const }
: { status, content, message },
ingest: async () => ({ status, content, message }),
},
registry: {
getCanonicalConnection: () => canonical,
@@ -93,11 +132,7 @@ for (const status of ["accepted", "duplicate", "conflict", "rejected"] as const)
contentFingerprint: frame.payload.contentFingerprint,
observedAtMs: 100,
status,
...(status === "rejected"
? { rejectionCode: "base_message_missing" }
: status === "conflict"
? { rejectionCode: "content_conflict" }
: {}),
...(status === "conflict" ? { rejectionCode: "content_conflict" } : {}),
});
assert.equal(published.length, status === "accepted" ? 1 : 0);
if (status === "accepted") {
apps/server/test/onetalk-rendered-card-repository.test.ts
+71
View File
@@ -0,0 +1,71 @@
// 验证 rendered-card supplement 可独立于基础消息提交。
import assert from "node:assert/strict";
import test from "node:test";
import { createOneTalkRenderedCardContentFingerprint } from "@trade-message-center/onetalk-contract";
import { createOneTalkRenderedCardRepository } from "../src/onetalk/rendered-card-repository.ts";
const content = {
version: 1 as const,
kind: "rendered_order" as const,
title: "Order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
test("stores a rendered-card supplement when its base message is absent", async () => {
const inserted: unknown[] = [];
const selectResults = [[], []];
const transaction = {
select: () => ({
from: () => ({
where: () => ({
limit: () => {
const result = selectResults.shift();
return Object.assign(Promise.resolve(result), { for: () => result });
},
}),
}),
}),
insert: () => ({
values: async (value: unknown) => inserted.push(value),
}),
update: () => assert.fail("unexpected update"),
};
const database = {
transaction: async (run: (value: typeof transaction) => Promise<unknown>) =>
run(transaction),
};
const repository = createOneTalkRenderedCardRepository(database as never);
const result = await repository.store({
channelAccountId: "account-1",
observation: {
conversationId: "conversation-1",
messageId: "message-1",
content,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(content),
observedAtMs: 100,
},
receivedAt: new Date("2026-09-15T00:00:00.000Z"),
});
assert.deepEqual(result, { status: "accepted", content, message: null });
assert.deepEqual(inserted, [
{
channelAccountId: "account-1",
conversationId: "conversation-1",
messageId: "message-1",
renderedCardContent: content,
renderedCardContentFingerprint: createOneTalkRenderedCardContentFingerprint(content),
renderedCardObservedAtMs: 100,
firstConfirmedAt: new Date("2026-09-15T00:00:00.000Z"),
lastObservedAt: new Date("2026-09-15T00:00:00.000Z"),
},
]);
});
apps/server/test/onetalk-websocket.test.ts
+47
View File
@@ -1687,6 +1687,53 @@ test("flushes observations before completing sync and publishes accepted live fa
}
});
test("publishes an existing rendered-card supplement when its base live message arrives", async () => {
const renderedCardContent = {
version: 1 as const,
kind: "rendered_order" as const,
title: "Late order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
const service = createService(async (_context, _source, observedMessage) => ({
status: "accepted" as const,
message: observedMessage,
renderedCardContent,
}));
const app = createTestApp({
database: createDatabaseStub(),
authorization: createMockAuthorizationReader([authorizationRecord]),
oneTalkService: service,
});
const mind = await openSocket(app, "/ws/mind");
const plugin = await openSocket(app);
try {
await connectMindPage(mind);
const online = nextPluginStatus(mind, "online");
await connectPlugin(plugin);
await online;
const acknowledgement = nextMessage(plugin);
const created = nextFrame(mind, (value) => value.type === "message.created");
plugin.send(JSON.stringify(observedFrame("late-base-message", message("late-message"))));
assert.equal((await acknowledgement).type, "message.ack");
assert.deepEqual((await created).payload, {
message: {
...centerMessage(message("late-message")),
content: renderedCardContent,
},
});
} finally {
await closeApp(app, [mind, plugin]);
}
});
test("does not publish duplicate live observations", async () => {
const service = createService(async () => ({
status: "duplicate" as const,