mirror of
https://github.com/sinanyuntu/trade-message-center.git
synced 2026-09-17 13:22:11 +08:00
Merge pull request #49 from sinanyuntu/09-12-onetalk-profile-active-sync
完善 OneTalk 联系人资料主动同步链路
This commit is contained in:
YBF
@@ -40,6 +40,11 @@ OneTalkContactProfileStore.putPendingProfile(
|
||||
channelAccountId: string,
|
||||
profile: OneTalkContactProfile,
|
||||
): Promise<OneTalkContactProfileLedgerRecord>;
|
||||
OneTalkContactProfileStore.getProfile(
|
||||
channelAccountId: string,
|
||||
conversationId: string,
|
||||
): Promise<OneTalkContactProfileLedgerRecord | null>;
|
||||
OneTalkContactProfileStore.hasProfileRecord(channelAccountId: string): Promise<boolean>;
|
||||
OneTalkContactProfileStore.markProfileUploaded(input: {
|
||||
channelAccountId: string;
|
||||
conversationId: string;
|
||||
@@ -61,6 +66,9 @@ OneTalkContactProfileStore.discardPendingProfile(input: {
|
||||
### Source and identity
|
||||
|
||||
- MAIN world constructs a new whitelist object from `window.__conversationListData__` and `im-conversation-list:syncData`; it never forwards a raw row or response.
|
||||
- `onetalk.contact.snapshot` is an account-level command with the exact payload `{ action: "onetalk.contact.snapshot" }`; it reads only already-loaded direct profiles. It is attempted once only when the same account has a usable page, Bright is authenticated, and the durable ledger has no record for that account. Repeated hello, replay, selection change, Port reconnect or Bright reconnect must not create a second initial snapshot.
|
||||
- `onetalk.contact.collect` has the exact payload `{ action: "onetalk.contact.collect", conversationId: string }`. MAIN finds only that direct conversation in the current list; it does not use the selected conversation, switch UI, scan history or infer a missing identity. `syncData` may still publish a previously seen direct profile; publication is not upload eligibility.
|
||||
- A live `messageType: "new"` from either direction requests a targeted collect for its own `conversationId`. History and manual history sync never request a profile collect.
|
||||
- `channelAccountId` comes only from the logged-in page identity (`currentUserAccountId` or `IcbuIM.UserUtil.currentUser.accountId`). URL `activeAccountId` identifies the selected counterpart and is never an account fallback.
|
||||
- The durable profile key is `[channelAccountId, conversationId]`; `aliId` remains a profile field and does not define ownership. A profile may be persisted before the technical conversation row is discovered.
|
||||
- Profile observation is intentionally independent from message observation: `__conversationListData__` and message history/live callbacks may arrive in either order. Do not join a profile snapshot into a `business_card` while decoding a message; the message fact stores only the marker and the server read path may later use this current profile by the scoped key.
|
||||
@@ -69,7 +77,7 @@ OneTalkContactProfileStore.discardPendingProfile(input: {
|
||||
|
||||
### Bright WebSocket boundary
|
||||
|
||||
- Protocol version is `5`. `contact.profile.observed` is a plugin-direction frame with `profiles.length` in `1..100` and serialized UTF-8 size at most `256 KiB`; every profile has the exact 13 whitelisted fields above.
|
||||
- Protocol version is `5`. `contact.profile.observed` is a plugin-direction frame with `profiles.length` in `1..100` and serialized UTF-8 size at most `256 KiB`; the extension additionally emits at most 50 profiles per frame. Every profile has the exact 13 whitelisted fields above.
|
||||
- `contact.profile.ack` is a plugin-direction frame with `{ status: "delivered"; profileCount: number }`. It means the Bright transaction and its authorization/connection fences completed; it is not a Mind HTTP response and does not mean CRM business data was committed elsewhere.
|
||||
- Profile ingestion uses the existing plugin binding, `sync` operation and `read` permission. No second socket, page credential, Cookie, Mind user/workspace field or profile HTTP endpoint is introduced.
|
||||
- If any profile has `observedAtMs > receivedAtMs + 5 minutes`, Bright rejects the whole batch with `ws.error` code `profile_observed_at_future`; it performs no profile DB write and sends no profile ACK. The coordinator discards only the matching pending request entries so the rejected batch is not retried forever.
|
||||
@@ -77,7 +85,8 @@ OneTalkContactProfileStore.discardPendingProfile(input: {
|
||||
### Durable lifecycle
|
||||
|
||||
- `ONE_TALK_SYNC_DATABASE_VERSION` is `8`. The independent `onetalk_contact_profiles` store is keyed by the account/conversation pair; message, candidate, checkpoint and anomaly stores remain separate.
|
||||
- Observation writes pending state before Bright send. Within a record, a newer `observedAtMs` replaces an older pending snapshot. A profile with an `observedAtMs` at or below the uploaded high-water mark is skipped, even when its fingerprint differs; an equal fingerprint with a newer time advances the durable high-water mark without reopening pending.
|
||||
- The durable profile fingerprint is the only upload-deduplication gate. For every received profile, read its `[channelAccountId, conversationId]` record before `putPendingProfile()`. If either `pending.fingerprint` or `lastUploadedFingerprint` equals the incoming fingerprint, end after that read: do not update a timestamp/high-water mark, write IndexedDB, emit `profile_observed` or flush/send. A different fingerprint follows the existing durable-first pending write.
|
||||
- Same-key observations serialize the `getProfile → putPendingProfile` transition. Concurrent equal fingerprints may each perform their own read, but at most one may write, diagnose or flush. A read failure stays observable to the caller; tracker cleanup must not create a second unhandled rejection. If the coordinator is disposed while a durable read is pending, its completion performs no write, diagnostic or flush.
|
||||
- An ACK marks uploaded only after a readwrite transaction and only when the current pending fingerprint and `observedAtMs` still match. Older, unknown, duplicate or mismatched ACKs are no-ops. A future-skew error removes only the exact matching pending snapshot and records its rejected observation watermark.
|
||||
- Reconnect, Service Worker restart and a newly authenticated Bright client rebuild sends from durable pending records. There is no profile deletion, TTL, or fallback to a message row/Mind database.
|
||||
- Page/configuration callbacks use page, connection and configuration identity fences. A stale callback cannot route a snapshot to another account or operate a replacement coordinator.
|
||||
@@ -94,25 +103,28 @@ OneTalkContactProfileStore.discardPendingProfile(input: {
|
||||
| Group row, missing logged-in identity or CRM identifier mismatch | drop observation; do not write the old account or unmatched customer |
|
||||
| Extra frame key, sensitive key or wrong frame direction/scope | reject as `invalid_message`; no secret echo |
|
||||
| Empty or over-limit profile batch | `invalid_message`; no Bright DB write, send or ACK |
|
||||
| Equal/older profile observation | durable HWM/pending rules skip it; no duplicate send |
|
||||
| Equal pending/uploaded fingerprint | exactly one ledger read for that observation, then no write, diagnostic, flush or send |
|
||||
| Different fingerprint | durable-first pending write, `profile_observed` diagnostic and bounded flush are allowed |
|
||||
| Any profile beyond the five-minute future skew | whole batch `profile_observed_at_future`; no DB write/ACK, matching pending is discarded |
|
||||
| IndexedDB write/commit abort | no Bright send or uploaded state; failure remains explicit |
|
||||
| Bright offline, transport error or authorization/connection fence failure | no ACK; pending remains unless the explicit future-skew error was received |
|
||||
| ACK does not match request count or current fingerprint/time | no uploaded mark; pending/current record is preserved |
|
||||
| Page/account/configuration epoch is stale | no observer/coordinator/command side effect |
|
||||
| Auth, page/account/configuration epoch or coordinator becomes stale during the initial ledger check | no snapshot command; an old setup claim cannot block a later authenticated setup |
|
||||
| Coordinator is disposed after a durable read starts | no pending write, diagnostic or flush; a real read failure remains a caller-visible failure |
|
||||
|
||||
## 5. Good / Base / Bad Cases
|
||||
|
||||
- Good: MAIN derives a fixed profile from the logged-in account, durable state commits under `[account, conversation]`, Bright receives a bounded frame, and the exact current ACK fence marks it uploaded.
|
||||
- Good: an old profile arrives again through `syncData` or a targeted live collect; Service Worker reads its ledger fingerprint once and returns without a write, diagnostic or Bright frame.
|
||||
- Base: a profile with nullable business fields is a valid partial snapshot; a later strictly newer snapshot may overwrite every stored field, including explicit `null`.
|
||||
- Bad: serializing a conversation row, using URL `activeAccountId`, keying the ledger by `aliId`, treating an HTTP/Mind response as the upload ACK, retrying a future-skew batch, or clearing pending on send start.
|
||||
- Bad: using MAIN `seen`, page selection, a timer or a second account baseline as upload dedupe; treating an equal fingerprint as a timestamp refresh; routing a live message to a full list scan; or clearing pending on send start.
|
||||
|
||||
## 6. Tests Required
|
||||
|
||||
- Contract: exact profile/frame keys, protocol version, direction/scope, direct discovery type, sensitive/unknown key rejection, empty/over-limit batches and `256 KiB` byte limit.
|
||||
- Observer: page snapshot/update, group exclusion, logged-in identity, logout/account switch, duplicate fingerprints, CRM customer matching and avatar URL validation.
|
||||
- Ledger: the pre-v7 (`oldVersion < 7`) upgrade clears every old OneTalk store before recreating current profile and message-sync state, and the v7→v8 upgrade keeps the five existing stores; account/conversation key, durable-first ordering, abort retention, latest pending replacement, uploaded HWM, stale different-fingerprint skip, exact ACK/CAS, future-skew discard, reconnect and restart recovery. The upgrade must preserve configuration/deviceId and must not rekey/retry an old pending ledger.
|
||||
- Service Worker: existing Bright binding/read/sync authorization, profile request mapping, ACK count, future error mapping and stale callback/page identity fences; direct discovery always carries `conversationType: "direct"`.
|
||||
- Observer: initial snapshot, repeatable `syncData` publication, exact snapshot/collect payload validation, targeted direct lookup, group/unknown target exclusion, logged-in identity, logout/account switch, CRM customer matching and avatar URL validation.
|
||||
- Ledger: the pre-v7 (`oldVersion < 7`) upgrade clears every old OneTalk store before recreating current profile and message-sync state, and the v7→v8 upgrade keeps the five existing stores; account/conversation key, durable-first ordering, same-pending/uploaded one-read-zero-write behavior, same-key concurrent writes, dispose-during-read, explicit read rejection, ACK/CAS, future-skew discard, reconnect and restart recovery. The upgrade must preserve configuration/deviceId and must not rekey/retry an old pending ledger.
|
||||
- Service Worker: existing Bright binding/read/sync authorization, first-auth setup in both page-first and auth-first order, auth-loss-after-ledger-read, live sent/received targeted collect, history exclusion, 50/51 profile batches, ACK count, future error mapping and stale callback/page identity fences; direct discovery always carries `conversationType: "direct"`.
|
||||
- Direct typecheck, contract/extension focused and full tests, format check and `git diff --check` are required. Real Chromium, Bright PostgreSQL and production Mind integration are separate external checks.
|
||||
|
||||
## 7. Wrong vs Correct
|
||||
@@ -127,6 +139,14 @@ await store.markProfileUploaded({ channelAccountId, aliId, fingerprint, uploaded
|
||||
### Correct
|
||||
|
||||
```ts
|
||||
// Do not let a page-local seen set or observedAt timestamp decide upload eligibility.
|
||||
const existing = await store.getProfile(channelAccountId, profile.conversationId);
|
||||
if (
|
||||
existing?.pending?.fingerprint === profile.profileFingerprint ||
|
||||
existing?.lastUploadedFingerprint === profile.profileFingerprint
|
||||
) {
|
||||
return;
|
||||
}
|
||||
await store.putPendingProfile(channelAccountId, profile);
|
||||
const requestId = bright.sendContactProfiles({ profiles: [profile] });
|
||||
// On the matching ACK, mark only the current pending [account, conversation, fingerprint, time].
|
||||
|
||||
@@ -152,7 +152,8 @@ Profile envelope 必须显式携带当次读取的 channelAccountId。Service Wo
|
||||
- 不得广播到多个标签页；
|
||||
- 不得跨账号回退；
|
||||
- 不得随机选择页面；
|
||||
- onetalk.contact.snapshot 是唯一同账号页面的 account-level observation trigger；重复相同 hello 不重复触发，snapshot result 不阻塞消息 bootstrap；
|
||||
- `onetalk.contact.snapshot` 是唯一同账号页面的 account-level initial observation trigger；只接受精确 payload `{ action: "onetalk.contact.snapshot" }`，并且只在 Bright 已认证、页面可用且该账号 durable profile ledger 为空时由 Service Worker 请求。重复 hello、replay、选择变化或连接恢复不重复触发，snapshot result 不阻塞消息 bootstrap；
|
||||
- `onetalk.contact.collect` 是唯一同账号页面的 account-level targeted observation command；只接受精确 payload `{ action: "onetalk.contact.collect", conversationId }`，不要求或改变当前 selected 会话。MAIN 只能从已加载 direct list 找到该 conversationId；找不到或群聊完成返回 `profileCount: 0`，不猜测或广播；
|
||||
- 账号级同步命令只发送到唯一同账号页面；
|
||||
- onetalk.sync.conversation 只在同一 MAIN 页面已完成 onetalk.discover-conversations 并缓存 direct 会话后使用；它必须令 command 与 route 的 conversationId 相等，并投递到唯一同账号页面，不要求该页面当前 selected 会话就是历史目标；MAIN cache 继续是目标会话存在性的唯一证明；
|
||||
- \`onetalk.sync\` 保持 \`channelAccountId + conversationId\` 的精确页面路由；
|
||||
@@ -182,12 +183,15 @@ Profile envelope 必须显式携带当次读取的 channelAccountId。Service Wo
|
||||
| Service Worker 重新实例化 | 页面注册表为空，等待页面重新连接 |
|
||||
| Profile envelope 账号与 hello/config 账号不一致 | 丢弃 profile observation，不写 ledger、不发 Bright |
|
||||
| 同一 page identity 重复 hello | 保持已有 identity，不重复启动 snapshot lifecycle；活跃断线 latch 仅重放同账号 UI 状态 |
|
||||
| contact snapshot/collect action 缺字段、多字段或字段类型错误 | `rejected_before_send/invalid_request`；不访问 observer、SDK 或页面状态 |
|
||||
| collect target 不在当前 loaded direct list、为群聊或无登录身份 | `completed/profileCount: 0`；不切换 UI、不扫描历史、不猜测 target |
|
||||
|
||||
## 5. Good / Base / Bad Cases
|
||||
|
||||
- Good：页面先发送合法 hello，Service Worker 按账号和 command 类型选择唯一 Port；发送目标通过 command payload 传给 MAIN。
|
||||
- Good：当前页面打开会话 A，但发送目标为会话 B；只要同账号页面唯一，command 仍投递，不改变页面 selected 状态。
|
||||
- Good：全量 discovery 已在同页缓存会话 B，页面当前选中会话 A；onetalk.sync.conversation(B) 仍投递到该唯一账号页面，并由 MAIN cache 解析 B。
|
||||
- Good：live message 指向 direct 会话 B 时，Service Worker 把精确 collect(B) 投递到唯一同账号页面；B 未选中也不会改变路由结果。
|
||||
- Base：页面 selected 状态变化后重新发送 hello；旧请求不会被新身份的迟到结果完成。
|
||||
- Bad：将 URL \`conversationId\` 当成 SPA 会话身份、取第一个 selected 节点、广播到所有页面或跨账号回退。
|
||||
- Bad：把页面 Port 当作持久化队列，或在 Port 断开后自动重新发送原 command。
|
||||
@@ -201,6 +205,7 @@ Profile envelope 必须显式携带当次读取的 channelAccountId。Service Wo
|
||||
- \`onetalk.send\` 在当前 selected 为其它会话、零 selected 或多 selected 时，仍向唯一同账号页面投递。
|
||||
- \`onetalk.send\` 同账号零页面返回 \`waiting_for_page\`，多页面返回 \`ambiguous_page_route\`，均不广播。
|
||||
- 页面 post 失败、Port 断开和身份变化都返回带 reason 的 unknown。
|
||||
- contact snapshot 和 collect 都覆盖 exact payload rejection；collect 覆盖非 selected direct target、缺失 target、群聊和缺失页面登录身份。
|
||||
- 页面桥 build 产物为自包含入口，Manifest 路径、world 和 Port 名称一致。
|
||||
- connection-status 覆盖 exact-shape、错误 source/version/origin/direction、无 command-result、同账号 Port fan-out 与重复 matching hello 的离线状态重放。
|
||||
- raw `contentType`、`custom.data`、顶层 `text` 或未知 content version 均不能通过 bridge；history/live 必须复用 MAIN 的同一 decoder。
|
||||
|
||||
@@ -84,15 +84,15 @@ Bright WebSocket
|
||||
| `commands/index.ts` / `createOneTalkPageCommandHandler(pageWindow, dependencies)` | 返回现有 `OneTalkPageCommandHandler`，只按 action 委派 |
|
||||
| `commands/send.ts` / `handleOneTalkSendCommand(pageWindow, message, sendObservation)` | 校验发送输入并执行 text 或委派原 image/file 适配 |
|
||||
| `current-conversation-history/page-command.ts` | 处理 discover/sync/sync.conversation，不接收发送关联器 |
|
||||
| `contact-observer/page-command.ts` / `handleOneTalkContactProfileCommand(observer)` | 调用现有 snapshot，返回 completed/profileCount；不再承担 action 分发 |
|
||||
| `contact-observer/page-command.ts` / `handleOneTalkContactProfileCommand(observer, command)` | 严格校验并执行 snapshot 或 targeted collect，返回 completed/profileCount；不承担 action 分发 |
|
||||
|
||||
#### Contracts
|
||||
|
||||
- 组合顺序固定为：页面 UI 能力 → 单份发送关联器 → profile observer → 共用 publisher 与命令 handler → page bridge → buyer/message/manual-history。profile 仍直接使用既有 installer，不能为合并入口而提前/推迟其安装或引入 prepare/start 状态机。
|
||||
- `commands/index.ts` 将 send 委派给 send handler，将 contact.snapshot 委派给 profile handler，将三个 discover/sync 动作委派给 history handler。dispatcher 不解析业务字段，不用 handler 返回 null 逐个试路由，未知动作直接返回既有 invalid_request。
|
||||
- `commands/index.ts` 将 send 委派给 send handler，将 contact.snapshot 与 contact.collect 委派给 profile handler，将三个 discover/sync 动作委派给 history handler。dispatcher 不解析业务字段，不用 handler 返回 null 逐个试路由，未知动作直接返回既有 invalid_request。
|
||||
- publisher 对非空消息或有诊断的批次先调用 `observeSentMessages(batch.messages)`，再调用 `observedSink(batch)`；实时、历史命令和公开历史函数共用这一出口。采集只收到 observe 回调，不能调用 execute 或接管发送 pending。
|
||||
- `message-observer/send-observation.ts` 保持发送 pending、候选 ID、timeout 和 settle 的唯一 owner；命令侧通过 execute/executeImage/executeFile 使用。`image-send.ts` 保持原生媒体上传拦截器和关联状态，不复制或搬迁模块级状态容器。
|
||||
- profile snapshot 使用已安装的同一 observer，其 snapshot 自身更新 seen/CRM 相关状态并发布资料；命令只能返回统计，不能再 publish。
|
||||
- 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。
|
||||
|
||||
#### Validation & Error Matrix
|
||||
@@ -102,7 +102,9 @@ Bright WebSocket
|
||||
| batch 有消息 | 先提供给原发送关联器，再正常上报；关联成功不吞掉事实 |
|
||||
| batch 无消息但有 unsupported/invalid/anomaly 诊断 | 继续发布一次诊断批次 |
|
||||
| batch 无消息且无诊断 | 两个消费者都不调用 |
|
||||
| contact.snapshot 在 bridge 首次 hello 后立即到达 | 使用预先安装的 observer，发布一次资料并返回一个结果 |
|
||||
| initial contact.snapshot 在 Bright 认证、页面可用且 ledger 为空后到达 | 使用预先安装 observer 读取当前 loaded direct profiles；重复 hello/replay/status 不再请求全量 snapshot |
|
||||
| contact.collect 带精确 direct conversationId | 使用预先安装 observer 只读取该 profile；history/手动 history 不触发 collect |
|
||||
| contact snapshot/collect 多余字段、缺 conversationId 或错误类型 | rejected_before_send/invalid_request，不访问 observer 或 SDK |
|
||||
| 未知 action | rejected_before_send/invalid_request，不访问 SDK 或 snapshot |
|
||||
| send 的 SDK 本地接受、超时或歧义 | 保持既有发送 SOP，不增加重试或假成功 |
|
||||
| 历史命令返回 batch | 使用同一 publisher，不新建发送关联器或上报管线 |
|
||||
@@ -115,7 +117,7 @@ Bright WebSocket
|
||||
|
||||
#### Tests Required
|
||||
|
||||
- `onetalk-page-flow-boundaries.test.js` 覆盖 publisher 双消费顺序、诊断-only/空批次、snapshot/未知动作与实际页面入口安装顺序。
|
||||
- `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/异常及不重试断言。
|
||||
- 对构建后的 MAIN IIFE 执行 `scripts/verify-release-bundle.mjs`，验证实际 shared observation 驱动消息上报及单次发送确认；这属于无网络 Node VM 冒烟，真实 Chromium/OneTalk/Bright 联调需单独标记。
|
||||
|
||||
@@ -224,6 +226,7 @@ writer.sendSendConfirmation(result);
|
||||
- onetalk.send 只要求唯一同账号页面；目标会话交给 MAIN 后必须由 SDK input 的 cid 指定，conversationCode 仅为同值兼容字段，当前 selected 会话不作为发送前置条件。
|
||||
- 页面身份变化、Port 断开和 Service Worker 重启必须让旧 correlation 失效；不得由迟到结果恢复旧请求。
|
||||
- 相同 page identity 的重复 hello 不重复启动 snapshot；profile page/config/connection epoch 失效时不得操作替换后的 engine/coordinator。
|
||||
- Profile 初始 setup 同时受 Bright authenticated、page identity、configuration epoch 和 active coordinator fence 约束；任何 await 后都要重验。ledger read 后认证丢失、页面/账号切换或 coordinator 替换时不得发送 snapshot；失效 claim 不得阻塞下一次有效 setup。
|
||||
|
||||
### Durable and lifecycle boundaries
|
||||
|
||||
@@ -233,6 +236,7 @@ writer.sendSendConfirmation(result);
|
||||
- Bright 连接断开或发送结果丢失不得自动重发，不创建隐式发送任务。
|
||||
- `authorization_unavailable` 表示授权依赖暂时不可用，只关闭当前 Bright socket 并沿既有连接退避自动重连；只有凭证、授权版本、binding、scope 或协议版本等确定性错误才阻断自动重连并进入 unauthorized。
|
||||
- Service Worker 重启从 IndexedDB 恢复 checkpoint、候选和模式，不信任旧内存 cursor。
|
||||
- live `messageType: "new"`（sent 或 received）只触发所属 conversation 的 profile collect；history 不触发。profile 相同 fingerprint 在 coordinator 的唯一 durable read 后静默结束，不能通过 MAIN `seen`、时间水位或 timer 再建第二去重状态。
|
||||
- `oldVersion < 7` 的数据库先执行 v7 全量 OneTalk state 清空（v7→v8 只新增 bootstrap store、保留既有事实），再重新采集 profile、会话活动时间并执行 full sync；之后的重启/重连才从当前 ledger 恢复 pending。ACK 只在当前 `[channelAccountId, conversationId, fingerprint, observedAtMs]` 的 IndexedDB transaction commit 后生效；future-skew 整批拒绝并只丢弃匹配 pending。
|
||||
|
||||
### Send and protocol boundaries
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
{"file":".trellis/spec/chrome-extension/frontend/quality-guidelines.md","reason":"Focused test, typecheck and build verification requirements."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/contact-profile-sync.md","reason":"Verify profile durable-first, hash, batching, ACK and account isolation invariants."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/page-bridge.md","reason":"Verify command validation, exact page routing and stale identity fences."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/runtime-sync.md","reason":"Verify message/profile lifecycle separation and no duplicate page activation."}
|
||||
@@ -0,0 +1,80 @@
|
||||
# 技术设计：OneTalk 客户资料活跃同步
|
||||
|
||||
## 目标边界
|
||||
|
||||
基础客户资料是低优先级事实。它不再由 page hello、会话选择变化、Port 重连、Bright 重连或历史同步主动全量采集；仅有首次认证初始化、新会话和任一方向的 live 消息活动三类业务触发。页面资料可以重复跨 MAIN → bridge → Service Worker；唯一的去重与上传资格判断在 durable profile hash。
|
||||
|
||||
Bright WebSocket `contact.profile.observed`、服务端 profile transaction、ACK CAS、profile 白名单与 `(channelAccountId, conversationId)` ledger key 均保持不变。
|
||||
|
||||
## 现状与根因
|
||||
|
||||
`OneTalkPageRuntimeHost.handlePageIdentity()` 和 `replayTo()` 目前无条件调用 account-level `onetalk.contact.snapshot`。MAIN observer 的 `snapshot()` 每次读取完整 `__conversationListData__` 并无条件发布全部 profile；coordinator 随后逐条调用 `putPendingProfile()`。已上传的相同指纹虽不会进入 Bright frame，却仍产生 IndexedDB high-water 写入和 `profile_observed: skipped` 诊断。
|
||||
|
||||
已存在的 `profileFingerprint` 是 `v1-xxxxxxxx`：8 位十六进制 FNV-1a 结果，输入是当前白名单基础 profile 字段。它已被 ledger 以 `pending.fingerprint` / `lastUploadedFingerprint` 持久化，不新建 hash 或第二份 profile 状态。
|
||||
|
||||
## 目标数据流
|
||||
|
||||
```text
|
||||
首次安装账号 + ws.accepted + 页面可用
|
||||
-> account snapshot（页面已加载 direct profiles）
|
||||
-> 全部 durable 写入
|
||||
-> 每帧最多 50 条发送
|
||||
|
||||
新会话 / syncData
|
||||
-> page profile observed（允许旧会话再次经过 bridge）
|
||||
-> IndexedDB hash compare
|
||||
-> 相同 hash 静默结束；不同 hash 才 durable write + upload
|
||||
|
||||
任一 live message（sent / received）
|
||||
-> 唯一 conversationId
|
||||
-> account-level targeted collect（不切换页面、不扫全表）
|
||||
-> IndexedDB hash compare
|
||||
-> hash 变化才 durable write + Bright upload
|
||||
```
|
||||
|
||||
history、手动历史同步和 `messageType: "history"` 不走 targeted collect。
|
||||
|
||||
## MAIN 页面观察器与命令
|
||||
|
||||
扩展 `OneTalkContactProfileObserver` 为三个明确操作：
|
||||
|
||||
- `snapshot()`：读取当前已加载 direct profile，并发布完整 profile 列表；仅首次认证初始化使用。
|
||||
- `collectConversation(conversationId)`：从当前 `__conversationListData__` 精确查找一个 direct conversation，并只发布该 profile；找不到、群聊或不完整身份返回 `profileCount: 0`，不猜测、不切换 UI。
|
||||
|
||||
`syncData` 与 CRM 观察保留页面侧已有的资料发布能力；它们不负责也不拥有 profile 上传去重。旧会话即使再次发布 profile，也由 Service Worker 的 durable hash 读比较静默结束。MAIN 不保存 `seen` 作为资料资格或上传 gate；新会话和 live activity 的识别仅决定何时尝试读取页面资料。
|
||||
|
||||
新增两个 MAIN 命令：
|
||||
|
||||
- `onetalk.contact.collect`：account-level，命令 payload 带严格的 `conversationId`；它不是“当前选择会话”路由，必须只投递到唯一同账号页面。
|
||||
|
||||
保留 `onetalk.contact.snapshot` 作为首次初始化命令。两个命令在 MAIN dispatcher 中显式校验 action 与 exact payload，未知/多余字段走既有 `invalid_request`。
|
||||
|
||||
## Service Worker 生命周期
|
||||
|
||||
`OneTalkPageRuntimeHost` 将 page identity 的消息同步职责与 profile 初始化职责解耦：
|
||||
|
||||
- page identity 仍立刻驱动 sync engine 和 buyer fact 的既有 lifecycle。
|
||||
- host 记录当前 authenticated account；`ws.accepted` 与 page identity 任何先后顺序都调用同一个 guarded profile setup。
|
||||
- profile setup 读取 profile ledger 是否已有该 account 的任何记录：无记录时 claim 一次首次 snapshot；有记录时不发 profile page command。claim 绑定 configuration epoch、page connection epoch、active profile coordinator，防止 hello/replay/status 的竞态重复请求。
|
||||
- 失败、页面不可用、配置/Port/coordinator 失效不会留下成功 claim；下一个有效认证或页面 lifecycle 可重试。成功后同一 active setup 不重复请求。
|
||||
- `replayTo()` 不再直接请求 snapshot，只复用这个 setup gate。
|
||||
|
||||
在 profile page bridge 消息抵达时，host 先执行既有 account/coordinator fence，再交给 coordinator。对 live 消息，host 从 `OneTalkPageObservedMessage.batch` 提取 `messageType === "new"` 的唯一 conversationId，发起 `onetalk.contact.collect`。收发方向不参与筛选；history 不触发。每个 conversationId 有短生命周期 in-flight correlation，避免同一批或未完成命令重复路由；完成后下一条 live 消息仍会重新读取并比较 hash。
|
||||
|
||||
## Durable 去重与发送
|
||||
|
||||
在调用 `putPendingProfile()` 前，coordinator 对每个收到的 profile 读取同 key ledger；这是唯一去重闸门：
|
||||
|
||||
- `pending.fingerprint` 或 `lastUploadedFingerprint` 等于新 `profileFingerprint`：读取后直接返回。不得更新 observed timestamp/high-water、不得写 IndexedDB、不得发 `profile_observed` 诊断、不得调用 flush。
|
||||
- 指纹不同或记录不存在：保持 current durable-first `putPendingProfile()`，再进入 flush。
|
||||
|
||||
initial snapshot 的所有输入必须先完成 durable 写入，再开始一次 flush。发送分组策略从协议上限 100 收紧为扩展本地 `50`，同时继续执行 256 KiB 实际 frame 大小检查。服务端仍可接收协议允许的 1..100，不改 contract 或数据库。
|
||||
|
||||
`profile_observed` 仅在实际进入 pending 的 profile 上报。首次 snapshot 可使用一条不含 profile 值的聚合诊断；相同 hash 是正常静默路径。
|
||||
|
||||
## 风险与回滚
|
||||
|
||||
- 首次 snapshot 只包含 OneTalk 页面已经加载的资料；不会新增 SDK/CRM 请求。若列表随后更新，new-conversation `syncData` 与 live message targeted collect 仍可补齐。
|
||||
- 旧会话再次穿过 MAIN、bridge 和 Service Worker 是允许的低成本路径；相同 hash 只做一个 IndexedDB 读取。不要重新引入 MAIN-side `seen` 或账号级 refresh state 作为第二个去重来源。
|
||||
- 8 位 FNV hash 有理论碰撞风险，但沿用已发布、低优先级资料的现有 fingerprint；本任务不改变字段或 hash 算法，避免重新采集和兼容性迁移。
|
||||
- 回滚只需恢复原触发逻辑；ledger 与 server schema 均不迁移，已有 pending 仍按现有 ACK 规则恢复。
|
||||
@@ -0,0 +1,5 @@
|
||||
{"file":".trellis/spec/guides/index.md","reason":"Cross-layer trigger and ownership review guide."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/index.md","reason":"Chrome extension package baseline and required quality checks."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/contact-profile-sync.md","reason":"Profile whitelist, durable ledger, ACK, identity and sensitive-field contract."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/page-bridge.md","reason":"MAIN/ISOLATED/Service Worker routing and page identity constraints."}
|
||||
{"file":".trellis/spec/chrome-extension/frontend/onetalk/runtime-sync.md","reason":"Page command ownership, lifecycle and runtime composition rules."}
|
||||
@@ -0,0 +1,36 @@
|
||||
# 执行计划：OneTalk 客户资料活跃同步
|
||||
|
||||
## 1. MAIN 资料观察与命令
|
||||
|
||||
1. 更新 `contact-observer/entry.ts`：保留页面资料发布行为，增加 targeted conversation collect；不使用 MAIN `seen` 作为 profile 去重或上传资格判断。
|
||||
2. 更新 `contact-observer/page-command.ts` 与 `main-page/commands/index.ts`：显式处理 targeted collect，严格校验 `conversationId`，保留 snapshot。
|
||||
3. 更新 page command/bridge routing：新命令以 account-level 方式精确路由到唯一同账号页面，禁止按当前选择会话猜测或广播。
|
||||
4. 增加 observer / MAIN command focused tests：首次 snapshot、重复资料可再次发布、targeted collect、live target missing/group、pagehide 与账号切换。
|
||||
|
||||
## 2. Service Worker lifecycle 与 live trigger
|
||||
|
||||
1. 从 `handlePageIdentity()`、`replayTo()` 移除无条件 profile snapshot；保持消息/buyer 的原有 page-ready 行为。
|
||||
2. 在 Bright authenticated 状态和 page identity 两端接入相同 profile setup gate；无 profile ledger record 时执行 initial snapshot，有记录时不发初始化资料命令，并覆盖两者乱序、重复 hello、reconnect、配置替换和 Port 失效。
|
||||
3. 在 page observed message 的 Service Worker 边界提取唯一 live conversationId，路由 targeted collect；history 不触发，收发方向均触发。
|
||||
4. 增加 runtime/controller/session tests，确认首次认证只全量一次、已初始化账号不再发初始化命令、会话切换不 snapshot、live/history 分流和 account/page fence。
|
||||
|
||||
## 3. Ledger 去重与发送策略
|
||||
|
||||
1. 为 profile store 增加只读的 account-record existence 查询，不改变 store schema 或记录形状。
|
||||
2. coordinator 对每个收到的 profile 在 `putPendingProfile()` 前读取 pending / uploaded fingerprint；相同 hash 在这次唯一 durable 读取后完全短路，不写 timestamp、不发诊断、不 flush。
|
||||
3. 将 extension outbound profile batch policy 固定为 50 条；保留 shared contract 的 100 条接收上限与 256 KiB frame 检查。
|
||||
4. 收敛 profile diagnostics：只报告真正 pending 的 observation，initial snapshot 用安全 aggregate，移除正常 skipped 风暴。
|
||||
5. 增加 store/coordinator tests：重复跨桥 profile 的一次读/零写/零诊断/零发送、same pending、same uploaded、different hash、50/51/101 条分包、ACK CAS、offline/reconnect、future-skew 与无 profile record 的初始化判定。
|
||||
|
||||
## 4. 质量门与实际验证
|
||||
|
||||
1. 运行扩展 focused tests（profile observer、page commands/bridge、service-worker runtime、sync runtime/controller、profile store/coordinator）。
|
||||
2. 运行 `pnpm --filter @trade-message-center/chrome-extension typecheck`、相关包 test 与根级 format/diff 检查。
|
||||
3. 运行 `gitnexus detect_changes` 审核受影响 symbol / flow；检查 diff 不混入既有 `apps/mind-test-harness/src/config.ts` 改动。
|
||||
4. 用 Chromium 验证：首次 auth 一次性 profile frames（每帧 ≤50）；连续 hello/会话切换无全量 frame；live sent 和 received 各触发目标会话检查；相同 hash 无 IndexedDB 写/诊断/WS frame。
|
||||
|
||||
## 回滚点
|
||||
|
||||
- MAIN command/observer 改动完成后，先跑 observer 与 bridge tests。
|
||||
- lifecycle gate 完成后，先跑 runtime/controller tests；若 page identity 或 auth 顺序有回归，先回滚该 gate，不改 ACK/ledger 语义。
|
||||
- batch / hash short-circuit 完成后，先跑 coordinator/store tests；不得以隐藏诊断代替验证无写入、无 flush 的真实短路。
|
||||
@@ -0,0 +1,40 @@
|
||||
# 收敛 OneTalk 客户资料活跃同步
|
||||
|
||||
## Goal
|
||||
|
||||
将 OneTalk 基础客户资料采集收敛为低优先级、由实际业务活动驱动的流程：首次安装后的首次 Bright 认证采集一次已加载资料；后续新会话或新消息活动可以再次把资料送到 Service Worker。已持久化的资料指纹是唯一去重闸门：相同则只做一次 IndexedDB 读取，不产生写入、`profile_observed` 诊断或 Bright 上传。
|
||||
|
||||
## Confirmed Facts
|
||||
|
||||
- 当前页面 hello、会话选择变化和 runtime replay 可触发全量 `onetalk.contact.snapshot`，其每次都会读取并发布当前 `__conversationListData__`，造成重复 `profile_observed: skipped`。
|
||||
- `profileFingerprint` 已是白名单基础资料生成的 `v1-xxxxxxxx`（8 位十六进制 hash），以 `[channelAccountId, conversationId]` 为 key 持久化在 profile IndexedDB ledger 的 pending/last-uploaded 状态中。
|
||||
- 当前 profile coordinator 会在每次 `observe()` 完成写入后立即 flush；协议上限为 100 条，但产品要求后续实际上传以 50 条为一组。
|
||||
- 已归一化 live 消息带有 `messageType: "new"`、`direction` 与 `conversationId`，因此可作为目标会话资料检查的精确触发，而不必扫描完整会话列表。
|
||||
|
||||
## Requirements
|
||||
|
||||
1. 首次采集：仅当本次扩展安装对应的 channel account 尚无 profile ledger 记录、Bright 已认证且 OneTalk 页面可用时，采集页面已经加载的基础客户资料一次。页面 hello、会话切换、Port 重连与 Bright 重连不得重复触发该全量采集。
|
||||
2. 新会话：`im-conversation-list:syncData` 的 direct profile 可以正常经过 MAIN → bridge → Service Worker；MAIN 不维护用于决定是否上传的 profile 去重状态，durable hash 是唯一的上传资格判断。
|
||||
3. 新消息：收到任一方向的 live 新消息时，只检查消息所属的一个 direct conversation 的资料；不得把 history、手动历史同步或全量会话扫描当作资料采集触发。资料是否上传仍完全由 durable hash 比较决定。
|
||||
4. 指纹去重：每个到达 Service Worker 的 profile 都用已有白名单字段计算现有 8 位 hash，并对同一 account/conversation 执行一次 durable profile ledger 读取；与 pending 或已上传 hash 相同则完全无副作用。不同或没有 ledger record 才 durable-first 写入并进入上传。
|
||||
5. 上传：一次初始采集或其他收集完成后，先完成所有待发 profile 的 durable 写入，再按最多 50 条 profile 一个 `contact.profile.observed` frame 发送。保留现有 256 KiB 限制、ACK CAS、future-skew 拒绝、离线 pending 恢复和账号隔离。
|
||||
6. 诊断：`profile_observed` 只表示实际需要发送的目标会话资料；首次采集可以使用一个安全的聚合诊断，不能为已知/相同 hash 的资料逐条记录 `skipped`。
|
||||
|
||||
## Out of Scope
|
||||
|
||||
- 不增加 24 小时定时或机会式全量刷新。
|
||||
- 不引入微批 timer、第二条 WebSocket、主动 CRM/HTTP 请求，或扫描/切换页面会话。
|
||||
- 不修改 Bright 服务端协议、数据库 schema 或 profile 的白名单字段。
|
||||
|
||||
## Acceptance Criteria
|
||||
|
||||
- [ ] 新安装、认证完成且页面可用时，恰好进行一次当前已加载资料的全量采集；重复 hello、会话切换、Port/Bright 重连均不重复全量采集。
|
||||
- [ ] 初始采集后的旧会话资料即使再次经过 MAIN → bridge → Service Worker，相同 hash 也只产生一次 IndexedDB 读取；没有写入、`profile_observed` 诊断或 Bright frame。
|
||||
- [ ] 一个合格的 live 新消息只请求/检查其所属 conversationId 的 profile；history 不触发资料检查。
|
||||
- [ ] 相同 8 位 profile hash 不写 ledger、不产生 `profile_observed`、不向 Bright 发送；不同 hash 仍按 durable-first、ACK CAS 正确上传。
|
||||
- [ ] 初始或多会话待发资料按每帧最多 50 条拆分，同时满足既有 256 KiB 限制；50 以下不被人为拆分。
|
||||
- [ ] group、缺少登录身份、未知目标会话和账号不匹配继续 fail closed，不推断 identity。
|
||||
|
||||
## Key Decision
|
||||
|
||||
- live 新消息不区分 `direction`；客户入站和用户发送均触发该会话的一次目标资料检查。相同 hash 是正常无副作用路径，不产生 ledger 写入、诊断或上传。
|
||||
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"id": "onetalk-profile-active-sync",
|
||||
"name": "onetalk-profile-active-sync",
|
||||
"title": "收敛 OneTalk 客户资料活跃同步",
|
||||
"description": "仅在首次认证和会话活跃时采集客户资料，按持久化指纹去重并以 50 条分包上传",
|
||||
"status": "completed",
|
||||
"dev_type": null,
|
||||
"scope": null,
|
||||
"package": null,
|
||||
"priority": "P2",
|
||||
"creator": "ybf",
|
||||
"assignee": "ybf",
|
||||
"createdAt": "2026-09-12",
|
||||
"completedAt": "2026-09-13",
|
||||
"branch": "09-12-onetalk-profile-active-sync",
|
||||
"base_branch": "main",
|
||||
"worktree_path": null,
|
||||
"commit": null,
|
||||
"pr_url": null,
|
||||
"subtasks": [],
|
||||
"children": [],
|
||||
"parent": null,
|
||||
"relatedFiles": [],
|
||||
"notes": "",
|
||||
"meta": {}
|
||||
}
|
||||
@@ -8,8 +8,8 @@
|
||||
|
||||
<!-- @@@auto:current-status -->
|
||||
- **Active File**: `journal-1.md`
|
||||
- **Total Sessions**: 64
|
||||
- **Last Active**: 2026-09-12
|
||||
- **Total Sessions**: 65
|
||||
- **Last Active**: 2026-09-13
|
||||
<!-- @@@/auto:current-status -->
|
||||
|
||||
---
|
||||
@@ -19,7 +19,7 @@
|
||||
<!-- @@@auto:active-documents -->
|
||||
| File | Lines | Status |
|
||||
|------|-------|--------|
|
||||
| `journal-1.md` | ~1399 | Active |
|
||||
| `journal-1.md` | ~1433 | Active |
|
||||
<!-- @@@/auto:active-documents -->
|
||||
|
||||
---
|
||||
@@ -29,6 +29,7 @@
|
||||
<!-- @@@auto:session-history -->
|
||||
| # | Date | Title | Commits | Branch |
|
||||
|---|------|-------|---------|--------|
|
||||
| 65 | 2026-09-13 | OneTalk profile active sync | `f93255e` | `09-12-onetalk-profile-active-sync` |
|
||||
| 65 | 2026-09-12 | OneTalk 采集与命令流程边界整理 | `15efc84` | `09-12-onetalk-flow-boundaries` |
|
||||
| 64 | 2026-09-12 | Harden Mind send liveness | `440df49`, `cc2ee8f` | `09-12-mind-send-liveness` |
|
||||
| 63 | 2026-09-12 | OneTalk 结构化消息信息采集分类 | `f92f00d`, `31400fc`, `86447cc` | `main` |
|
||||
|
||||
@@ -1409,3 +1409,25 @@ Fail-closed plugin lease admission plus Harness heartbeat and HTTP deadline term
|
||||
### Status
|
||||
|
||||
[OK] **Completed**
|
||||
|
||||
|
||||
## Session 65: OneTalk profile active sync
|
||||
<!-- trellis-session: v=2 fp=2ef84c1ed02f1730 -->
|
||||
|
||||
**Date**: 2026-09-13
|
||||
**Task**: OneTalk profile active sync
|
||||
**Branch**: `09-12-onetalk-profile-active-sync`
|
||||
|
||||
### Summary
|
||||
|
||||
收敛首次认证与活跃会话的 OneTalk profile 同步：durable hash 去重、50 条分包、精确 collect 命令与异步 lifecycle fence；377 个扩展测试、typecheck、build 通过，Chromium/Bright 实机 smoke deferred。
|
||||
|
||||
### Git Commits
|
||||
|
||||
| Hash | Message |
|
||||
|------|---------|
|
||||
| `f93255e` | feat(onetalk): sync active contact profiles |
|
||||
|
||||
### Status
|
||||
|
||||
[OK] **Completed**
|
||||
|
||||
@@ -13,7 +13,7 @@ import { handleOneTalkSendCommand } from "./send.ts";
|
||||
export type OneTalkPageCommandDependencies = {
|
||||
sendObservation: Pick<SendObservationCorrelator, "execute" | "executeImage" | "executeFile">;
|
||||
observedSink: OneTalkObservedMessageSink;
|
||||
profileObserver: Pick<OneTalkContactProfileObserver, "snapshot">;
|
||||
profileObserver: Pick<OneTalkContactProfileObserver, "snapshot" | "collectConversation">;
|
||||
historyBootstrapProgress: HistoryBootstrapProgressTooltip;
|
||||
};
|
||||
|
||||
@@ -27,7 +27,11 @@ export const createOneTalkPageCommandHandler = (
|
||||
case "onetalk.send":
|
||||
return handleOneTalkSendCommand(pageWindow, message, dependencies.sendObservation);
|
||||
case "onetalk.contact.snapshot":
|
||||
return handleOneTalkContactProfileCommand(dependencies.profileObserver);
|
||||
case "onetalk.contact.collect":
|
||||
return handleOneTalkContactProfileCommand(
|
||||
dependencies.profileObserver,
|
||||
message.command,
|
||||
);
|
||||
case "onetalk.discover-conversations":
|
||||
case "onetalk.sync":
|
||||
case "onetalk.sync.conversation":
|
||||
|
||||
@@ -23,6 +23,7 @@ type PageEventBus = {
|
||||
|
||||
export type OneTalkContactProfileObserver = {
|
||||
snapshot: () => OneTalkContactProfile[];
|
||||
collectConversation: (conversationId: string) => OneTalkContactProfile[];
|
||||
dispose: () => void;
|
||||
};
|
||||
|
||||
@@ -36,10 +37,6 @@ const unsubscribeFor = (value: unknown): (() => void) | undefined => {
|
||||
return typeof value === "function" ? (value as () => void) : undefined;
|
||||
};
|
||||
|
||||
const observedProfileKey = (channelAccountId: string, conversationId: string): string => {
|
||||
return JSON.stringify([channelAccountId, conversationId]);
|
||||
};
|
||||
|
||||
/** 安装独立联系人观察器；初始快照仅由 account-level command 触发。 */
|
||||
export const installOneTalkContactProfileObserver = (
|
||||
pageWindow: OneTalkPageWindow,
|
||||
@@ -49,7 +46,6 @@ export const installOneTalkContactProfileObserver = (
|
||||
): OneTalkContactProfileObserver => {
|
||||
let stopped = false;
|
||||
let observedChannelAccountId: string | undefined;
|
||||
const seen = new Map<string, string>();
|
||||
const crmObservations = new Map<string, OneTalkCrmAvatarObservation>();
|
||||
const now = (): number => Date.now();
|
||||
const emit = (profiles: OneTalkContactProfile[], channelAccountId: string): void => {
|
||||
@@ -65,23 +61,11 @@ export const installOneTalkContactProfileObserver = (
|
||||
};
|
||||
const clearPageIdentity = (): void => {
|
||||
observedChannelAccountId = undefined;
|
||||
seen.clear();
|
||||
crmObservations.clear();
|
||||
};
|
||||
const profilesFromObservationMap = (value: unknown): OneTalkContactProfile[] => {
|
||||
return profilesFromConversationMap(value, now(), crmObservations);
|
||||
};
|
||||
const changedProfiles = (
|
||||
profiles: OneTalkContactProfile[],
|
||||
channelAccountId: string,
|
||||
): OneTalkContactProfile[] => {
|
||||
return profiles.filter((profile) => {
|
||||
const key = observedProfileKey(channelAccountId, profile.conversationId);
|
||||
const previous = seen.get(key);
|
||||
seen.set(key, profile.profileFingerprint);
|
||||
return previous !== profile.profileFingerprint;
|
||||
});
|
||||
};
|
||||
const selectedSdkProfile = (conversationId: string): OneTalkContactProfile | null => {
|
||||
const profiles = profilesFromObservationMap(pageWindow.__conversationListData__);
|
||||
const matches = profiles.filter((profile) => profile.conversationId === conversationId);
|
||||
@@ -120,7 +104,7 @@ export const installOneTalkContactProfileObserver = (
|
||||
}
|
||||
crmObservations.set(sdkProfile.conversationId, crmObservation);
|
||||
const nextProfile = contactProfileWithCrmObservation(sdkProfile, crmObservation, now());
|
||||
emit(changedProfiles([nextProfile], channelAccountId), channelAccountId);
|
||||
emit([nextProfile], channelAccountId);
|
||||
};
|
||||
const consumeUpdates = (value: unknown): void => {
|
||||
const channelAccountId = readChannelAccountId(pageWindow);
|
||||
@@ -134,16 +118,12 @@ export const installOneTalkContactProfileObserver = (
|
||||
observedChannelAccountId !== channelAccountId
|
||||
) {
|
||||
observedChannelAccountId = channelAccountId;
|
||||
seen.clear();
|
||||
crmObservations.clear();
|
||||
onDiagnostic?.({ code: "profile_login_identity_changed" });
|
||||
return;
|
||||
}
|
||||
observedChannelAccountId = channelAccountId;
|
||||
emit(
|
||||
changedProfiles(profilesFromObservationMap(value), channelAccountId),
|
||||
channelAccountId,
|
||||
);
|
||||
emit(profilesFromObservationMap(value), channelAccountId);
|
||||
};
|
||||
const eventBus = eventBusFor(pageWindow);
|
||||
const unsubscribe = eventBus?.on?.("im-conversation-list:syncData", consumeUpdates);
|
||||
@@ -171,20 +151,23 @@ export const installOneTalkContactProfileObserver = (
|
||||
return [];
|
||||
}
|
||||
if (observedChannelAccountId !== channelAccountId) {
|
||||
seen.clear();
|
||||
crmObservations.clear();
|
||||
}
|
||||
observedChannelAccountId = channelAccountId;
|
||||
const profiles = profilesFromObservationMap(pageWindow.__conversationListData__);
|
||||
for (const profile of profiles) {
|
||||
seen.set(
|
||||
observedProfileKey(channelAccountId, profile.conversationId),
|
||||
profile.profileFingerprint,
|
||||
);
|
||||
}
|
||||
emit(profiles, channelAccountId);
|
||||
return profiles;
|
||||
},
|
||||
collectConversation: (conversationId) => {
|
||||
const channelAccountId = readChannelAccountId(pageWindow);
|
||||
if (stopped || !channelAccountId || conversationId.length === 0) return [];
|
||||
if (observedChannelAccountId !== channelAccountId) crmObservations.clear();
|
||||
observedChannelAccountId = channelAccountId;
|
||||
const profile = selectedSdkProfile(conversationId);
|
||||
if (!profile) return [];
|
||||
emit([profile], channelAccountId);
|
||||
return [profile];
|
||||
},
|
||||
dispose: onPageHide,
|
||||
};
|
||||
};
|
||||
|
||||
@@ -1,13 +1,37 @@
|
||||
// 处理 Service Worker 请求的联系人资料快照命令
|
||||
|
||||
import type { PageCommandResult } from "../../page-bridge/model.ts";
|
||||
import type { PageCommand, PageCommandResult } from "../../page-bridge/model.ts";
|
||||
import type { OneTalkContactProfileObserver } from "./entry.ts";
|
||||
|
||||
/** 执行联系人资料 account-level snapshot 命令。 */
|
||||
const isSnapshotCommand = (command: PageCommand): boolean => {
|
||||
return command.action === "onetalk.contact.snapshot" && Object.keys(command).length === 1;
|
||||
};
|
||||
|
||||
const isCollectConversationCommand = (
|
||||
command: PageCommand,
|
||||
): command is PageCommand & {
|
||||
action: "onetalk.contact.collect";
|
||||
conversationId: string;
|
||||
} => {
|
||||
return (
|
||||
command.action === "onetalk.contact.collect" &&
|
||||
Object.keys(command).length === 2 &&
|
||||
typeof command.conversationId === "string" &&
|
||||
command.conversationId.length > 0
|
||||
);
|
||||
};
|
||||
|
||||
/** 执行联系人资料 account-level snapshot 或 targeted collect 命令。 */
|
||||
export const handleOneTalkContactProfileCommand = (
|
||||
observer: Pick<OneTalkContactProfileObserver, "snapshot">,
|
||||
observer: Pick<OneTalkContactProfileObserver, "snapshot" | "collectConversation">,
|
||||
command: PageCommand,
|
||||
): PageCommandResult => {
|
||||
const profiles = observer.snapshot();
|
||||
const profiles = isSnapshotCommand(command)
|
||||
? observer.snapshot()
|
||||
: isCollectConversationCommand(command)
|
||||
? observer.collectConversation(command.conversationId)
|
||||
: null;
|
||||
if (profiles === null) return { status: "rejected_before_send", reason: "invalid_request" };
|
||||
return {
|
||||
status: "completed",
|
||||
profileCount: profiles.length,
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
OneTalkBrightClient,
|
||||
OneTalkBrightClientOptions,
|
||||
OneTalkBrightDiagnostic,
|
||||
OneTalkBrightClientState,
|
||||
} from "./transport/bright-client.ts";
|
||||
import { createOneTalkBrightClient } from "./transport/bright-client.ts";
|
||||
import type { OneTalkExtensionConfig } from "../config.ts";
|
||||
@@ -61,6 +62,10 @@ export type OneTalkConfiguredSyncSessionOptions = {
|
||||
onPageDiagnostic?: (event: OneTalkPageDiagnostic) => void;
|
||||
onEngineDiagnostic?: (event: OneTalkSyncEngineDiagnostic) => void;
|
||||
onProfileDiagnostic?: (event: OneTalkContactProfileDiagnostic) => void;
|
||||
onProfileStatus?: (
|
||||
profile: OneTalkContactProfileCoordinator,
|
||||
state: OneTalkBrightClientState,
|
||||
) => void;
|
||||
now?: () => number;
|
||||
createRequestId?: (kind: string) => string;
|
||||
};
|
||||
@@ -98,6 +103,9 @@ export class OneTalkConfiguredSyncSession {
|
||||
private readonly onProfileDiagnostic:
|
||||
| ((event: OneTalkContactProfileDiagnostic) => void)
|
||||
| undefined;
|
||||
private readonly onProfileStatus:
|
||||
| ((profile: OneTalkContactProfileCoordinator, state: OneTalkBrightClientState) => void)
|
||||
| undefined;
|
||||
private readonly now: (() => number) | undefined;
|
||||
private readonly createRequestId: ((kind: string) => string) | undefined;
|
||||
private store: OneTalkSyncStore | null = null;
|
||||
@@ -132,6 +140,7 @@ export class OneTalkConfiguredSyncSession {
|
||||
this.onPageDiagnostic = options.onPageDiagnostic;
|
||||
this.onEngineDiagnostic = options.onEngineDiagnostic;
|
||||
this.onProfileDiagnostic = options.onProfileDiagnostic;
|
||||
this.onProfileStatus = options.onProfileStatus;
|
||||
this.now = options.now;
|
||||
this.createRequestId = options.createRequestId;
|
||||
}
|
||||
@@ -241,6 +250,7 @@ export class OneTalkConfiguredSyncSession {
|
||||
? bright.subscribeStatus?.((state) => {
|
||||
if (currentRevision !== this.revision) return;
|
||||
activeProfile?.handleStatus(state);
|
||||
if (activeProfile) this.onProfileStatus?.(activeProfile, state);
|
||||
activeBuyer?.handleStatus(state);
|
||||
})
|
||||
: undefined;
|
||||
|
||||
@@ -19,6 +19,8 @@ import {
|
||||
type OneTalkContactProfileStore,
|
||||
} from "./storage.ts";
|
||||
|
||||
const ONE_TALK_CONTACT_PROFILE_OUTBOUND_BATCH_SIZE = 50;
|
||||
|
||||
export type OneTalkContactProfileDiagnostic = {
|
||||
event: "profile_observed" | "profile_send" | "profile_ack" | "profile_snapshot";
|
||||
status?:
|
||||
@@ -45,6 +47,8 @@ export type OneTalkContactProfileDiagnostic = {
|
||||
|
||||
export type OneTalkContactProfileCoordinator = {
|
||||
getChannelAccountId: () => string;
|
||||
hasProfileRecord: () => Promise<boolean>;
|
||||
isAuthenticated: () => boolean;
|
||||
observe: (profiles: OneTalkContactProfile[]) => Promise<void>;
|
||||
handleFrame: (frame: OneTalkFrame) => void;
|
||||
handleStatus: (state: OneTalkBrightClientState) => void;
|
||||
@@ -110,7 +114,11 @@ const profilesFor = (records: OneTalkContactProfileLedgerRecord[]): OneTalkConta
|
||||
};
|
||||
|
||||
const fitsProfileBatch = (scope: OneTalkPluginScope, batch: ProfileBatch): boolean => {
|
||||
if (batch.records.length > ONETALK_CONTACT_PROFILE_MAX_BATCH_SIZE) return false;
|
||||
if (
|
||||
batch.records.length > ONE_TALK_CONTACT_PROFILE_OUTBOUND_BATCH_SIZE ||
|
||||
batch.records.length > ONETALK_CONTACT_PROFILE_MAX_BATCH_SIZE
|
||||
)
|
||||
return false;
|
||||
return (
|
||||
byteLength(
|
||||
createOneTalkContactProfileObservedFrame(
|
||||
@@ -203,12 +211,14 @@ export const createOneTalkContactProfileCoordinator = (options: {
|
||||
const createRequestId =
|
||||
options.createRequestId ?? ((kind: string) => `profile-${kind}-${++requestSequence}`);
|
||||
const requests = new Map<string, RequestEntry[]>();
|
||||
const profileWrites = new Map<string, Promise<OneTalkContactProfileLedgerRecord | null>>();
|
||||
const writer = createOneTalkContactProfileFrameWriter({
|
||||
scope: options.scope,
|
||||
send: (frame) => options.bright.send(frame),
|
||||
createRequestId,
|
||||
});
|
||||
let disposed = false;
|
||||
let authenticated = false;
|
||||
let flushing: Promise<void> | null = null;
|
||||
let followUpFlushRequested = false;
|
||||
const reportError = (error: unknown): void => {
|
||||
@@ -315,22 +325,54 @@ export const createOneTalkContactProfileCoordinator = (options: {
|
||||
}
|
||||
validProfiles.push(profile);
|
||||
}
|
||||
const persistProfile = (
|
||||
profile: OneTalkContactProfile,
|
||||
): Promise<OneTalkContactProfileLedgerRecord | null> => {
|
||||
const key = contactProfileKey(options.scope.channelAccountId, profile.conversationId);
|
||||
const previous = profileWrites.get(key) ?? Promise.resolve(null);
|
||||
const operation = previous.then(async () => {
|
||||
const existing = await options.store.getProfile(
|
||||
options.scope.channelAccountId,
|
||||
profile.conversationId,
|
||||
);
|
||||
if (disposed) return null;
|
||||
if (
|
||||
existing?.pending?.fingerprint === profile.profileFingerprint ||
|
||||
existing?.lastUploadedFingerprint === profile.profileFingerprint
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const record = await options.store.putPendingProfile(
|
||||
options.scope.channelAccountId,
|
||||
profile,
|
||||
);
|
||||
return record.pending?.fingerprint === profile.profileFingerprint &&
|
||||
record.pending.observedAtMs === profile.observedAtMs
|
||||
? record
|
||||
: null;
|
||||
});
|
||||
let tracked: Promise<null> | null = null;
|
||||
const clearTracker = (): null => {
|
||||
if (tracked && profileWrites.get(key) === tracked) profileWrites.delete(key);
|
||||
return null;
|
||||
};
|
||||
tracked = operation.then(clearTracker, clearTracker);
|
||||
profileWrites.set(key, tracked);
|
||||
return operation;
|
||||
};
|
||||
let pendingCount = 0;
|
||||
for (const profile of stableDedupeProfiles(options.scope.channelAccountId, validProfiles)) {
|
||||
const record = await options.store.putPendingProfile(
|
||||
options.scope.channelAccountId,
|
||||
profile,
|
||||
);
|
||||
const record = await persistProfile(profile);
|
||||
if (!record || disposed) continue;
|
||||
pendingCount += 1;
|
||||
emit(options.onDiagnostic, {
|
||||
event: "profile_observed",
|
||||
status:
|
||||
record.pending?.fingerprint === profile.profileFingerprint
|
||||
? "pending"
|
||||
: "skipped",
|
||||
status: "pending",
|
||||
profileCount: 1,
|
||||
fieldNames: PROFILE_FIELD_NAMES,
|
||||
});
|
||||
}
|
||||
await flush();
|
||||
if (pendingCount > 0) await flush();
|
||||
};
|
||||
|
||||
const handleFrame = (frame: OneTalkFrame): void => {
|
||||
@@ -421,6 +463,7 @@ export const createOneTalkContactProfileCoordinator = (options: {
|
||||
};
|
||||
|
||||
const handleStatus = (state: OneTalkBrightClientState): void => {
|
||||
authenticated = state.status === "authenticated";
|
||||
if (state.status !== "authenticated") {
|
||||
requests.clear();
|
||||
followUpFlushRequested = false;
|
||||
@@ -431,6 +474,8 @@ export const createOneTalkContactProfileCoordinator = (options: {
|
||||
|
||||
return {
|
||||
getChannelAccountId: () => options.scope.channelAccountId,
|
||||
hasProfileRecord: () => options.store.hasProfileRecord(options.scope.channelAccountId),
|
||||
isAuthenticated: () => authenticated,
|
||||
observe,
|
||||
handleFrame,
|
||||
handleStatus,
|
||||
@@ -441,6 +486,8 @@ export const createOneTalkContactProfileCoordinator = (options: {
|
||||
},
|
||||
dispose: () => {
|
||||
disposed = true;
|
||||
authenticated = false;
|
||||
profileWrites.clear();
|
||||
requests.clear();
|
||||
followUpFlushRequested = false;
|
||||
},
|
||||
|
||||
@@ -34,6 +34,13 @@ type PageLifecycleToken = {
|
||||
pageEpoch: number;
|
||||
};
|
||||
|
||||
type ProfileSetup = {
|
||||
channelAccountId: string;
|
||||
configurationEpoch: number;
|
||||
profile: OneTalkContactProfileCoordinator;
|
||||
state: "checking" | "requested";
|
||||
};
|
||||
|
||||
const diagnosticForProfileSnapshot = (
|
||||
status: "failed" | "rejected",
|
||||
code: "snapshot_unavailable" | "snapshot_failed",
|
||||
@@ -49,6 +56,8 @@ export class OneTalkPageRuntimeHost {
|
||||
private readonly options: OneTalkPageRuntimeHostOptions;
|
||||
private lastPageIdentity: OneTalkPageIdentity | null = null;
|
||||
private profileSnapshotRequestSequence = 0;
|
||||
private profileSetup: ProfileSetup | null = null;
|
||||
private readonly liveProfileCollections = new Set<string>();
|
||||
private connectionEpoch = 0;
|
||||
private pageEpoch = 0;
|
||||
private disconnectedChannelAccountId: string | null = null;
|
||||
@@ -96,38 +105,105 @@ export class OneTalkPageRuntimeHost {
|
||||
}
|
||||
}
|
||||
|
||||
private requestProfileSnapshot(
|
||||
channelAccountId: string,
|
||||
token: PageLifecycleToken,
|
||||
identity: OneTalkPageIdentity,
|
||||
engine: OneTalkSyncEngine | null,
|
||||
profile: OneTalkContactProfileCoordinator | null,
|
||||
): void {
|
||||
if (!this.isCurrentToken(token, identity, engine, profile)) return;
|
||||
void this.runtime
|
||||
.routePageCommand({
|
||||
channelAccountId,
|
||||
requestId: `profile-snapshot-${++this.profileSnapshotRequestSequence}`,
|
||||
command: { action: "onetalk.contact.snapshot" },
|
||||
})
|
||||
.then((result) => {
|
||||
if (!this.isCurrentToken(token, identity, engine, profile)) return;
|
||||
private isCurrentProfileSetup(setup: ProfileSetup): boolean {
|
||||
return (
|
||||
this.profileSetup === setup &&
|
||||
setup.configurationEpoch === this.configurationEpoch() &&
|
||||
this.lastPageIdentity?.channelAccountId === setup.channelAccountId &&
|
||||
this.options.getActiveProfileCoordinator?.() === setup.profile &&
|
||||
(this.options.getActiveChannelAccountId?.() ?? setup.channelAccountId) ===
|
||||
setup.channelAccountId
|
||||
);
|
||||
}
|
||||
|
||||
private requestInitialProfileSnapshot(profile: OneTalkContactProfileCoordinator): void {
|
||||
const identity = this.lastPageIdentity;
|
||||
if (
|
||||
!identity ||
|
||||
!profile.isAuthenticated() ||
|
||||
profile.getChannelAccountId() !== identity.channelAccountId
|
||||
)
|
||||
return;
|
||||
const existing = this.profileSetup;
|
||||
if (
|
||||
existing &&
|
||||
existing.channelAccountId === identity.channelAccountId &&
|
||||
existing.configurationEpoch === this.configurationEpoch() &&
|
||||
existing.profile === profile &&
|
||||
this.isCurrentProfileSetup(existing)
|
||||
)
|
||||
return;
|
||||
const setup: ProfileSetup = {
|
||||
channelAccountId: identity.channelAccountId,
|
||||
configurationEpoch: this.configurationEpoch(),
|
||||
profile,
|
||||
state: "checking",
|
||||
};
|
||||
this.profileSetup = setup;
|
||||
void profile
|
||||
.hasProfileRecord()
|
||||
.then(async (hasRecord) => {
|
||||
if (!this.isCurrentProfileSetup(setup) || !setup.profile.isAuthenticated()) return;
|
||||
if (hasRecord) {
|
||||
setup.state = "requested";
|
||||
return;
|
||||
}
|
||||
const result = await this.runtime.routePageCommand({
|
||||
channelAccountId: setup.channelAccountId,
|
||||
requestId: `profile-snapshot-${++this.profileSnapshotRequestSequence}`,
|
||||
command: { action: "onetalk.contact.snapshot" },
|
||||
});
|
||||
if (!this.isCurrentProfileSetup(setup)) return;
|
||||
if (
|
||||
result.status === "rejected_before_send" ||
|
||||
result.status === "delivery_unknown"
|
||||
) {
|
||||
this.profileSetup = null;
|
||||
this.emitDiagnostic(
|
||||
diagnosticForProfileSnapshot("rejected", "snapshot_unavailable"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
setup.state = "requested";
|
||||
})
|
||||
.catch(() => {
|
||||
if (this.isCurrentToken(token, identity, engine, profile)) {
|
||||
this.emitDiagnostic(diagnosticForProfileSnapshot("failed", "snapshot_failed"));
|
||||
}
|
||||
if (!this.isCurrentProfileSetup(setup)) return;
|
||||
this.profileSetup = null;
|
||||
this.emitDiagnostic(diagnosticForProfileSnapshot("failed", "snapshot_failed"));
|
||||
});
|
||||
}
|
||||
|
||||
private requestLiveProfileCollection(channelAccountId: string, conversationId: string): void {
|
||||
const profile = this.options.getActiveProfileCoordinator?.() ?? null;
|
||||
if (!profile || profile.getChannelAccountId() !== channelAccountId) return;
|
||||
const key = JSON.stringify([channelAccountId, conversationId]);
|
||||
if (this.liveProfileCollections.has(key)) return;
|
||||
this.liveProfileCollections.add(key);
|
||||
void this.runtime
|
||||
.routePageCommand({
|
||||
channelAccountId,
|
||||
requestId: `profile-collect-${++this.profileSnapshotRequestSequence}`,
|
||||
command: { action: "onetalk.contact.collect", conversationId },
|
||||
})
|
||||
.catch((error) => this.options.onError(error))
|
||||
.finally(() => this.liveProfileCollections.delete(key));
|
||||
}
|
||||
|
||||
private requestLiveProfileCollections(
|
||||
message: Parameters<
|
||||
NonNullable<OneTalkServiceWorkerRuntimeOptions["persistPageObservation"]>
|
||||
>[0],
|
||||
channelAccountId: string,
|
||||
): void {
|
||||
for (const conversationId of new Set(
|
||||
message.batch
|
||||
.filter((entry) => entry.messageType === "new")
|
||||
.map((entry) => entry.conversationId),
|
||||
)) {
|
||||
this.requestLiveProfileCollection(channelAccountId, conversationId);
|
||||
}
|
||||
}
|
||||
|
||||
private resumeBuyerDelivery(
|
||||
token: PageLifecycleToken,
|
||||
identity: OneTalkPageIdentity,
|
||||
@@ -168,7 +244,7 @@ export class OneTalkPageRuntimeHost {
|
||||
return;
|
||||
this.replayConnectionStatus(channelAccountId);
|
||||
this.resumeBuyerDelivery(token, identity, engine, profile, buyer);
|
||||
this.requestProfileSnapshot(channelAccountId, token, identity, engine, profile);
|
||||
if (profile) this.requestInitialProfileSnapshot(profile);
|
||||
await engine?.handlePageReady(channelAccountId, conversationId);
|
||||
if (!this.isCurrentToken(token, identity, engine, profile)) return;
|
||||
await profile?.handlePageReady();
|
||||
@@ -196,6 +272,7 @@ export class OneTalkPageRuntimeHost {
|
||||
const engine = options.getActiveEngine();
|
||||
if (!engine) return;
|
||||
await engine.handlePageObservation(message, channelAccountId);
|
||||
this.requestLiveProfileCollections(message, channelAccountId);
|
||||
void sender;
|
||||
},
|
||||
persistPageProfileObservation: async (
|
||||
@@ -225,6 +302,7 @@ export class OneTalkPageRuntimeHost {
|
||||
this.connectionEpoch += 1;
|
||||
this.pageEpoch += 1;
|
||||
this.lastPageIdentity = null;
|
||||
this.liveProfileCollections.clear();
|
||||
options.getActiveEngine()?.handlePageDisconnected();
|
||||
options.getActiveProfileCoordinator?.()?.handlePageDisconnected();
|
||||
options.getActiveBuyerFactCoordinator?.()?.handlePageDisconnected();
|
||||
@@ -268,6 +346,16 @@ export class OneTalkPageRuntimeHost {
|
||||
this.runtime.publishConnectionStatus(previousChannelAccountId, false);
|
||||
}
|
||||
|
||||
public handleProfileStatus(
|
||||
profile: OneTalkContactProfileCoordinator,
|
||||
status: OneTalkBrightConnectionStatus,
|
||||
): void {
|
||||
if (status !== "authenticated" && this.profileSetup?.profile === profile) {
|
||||
this.profileSetup = null;
|
||||
}
|
||||
if (status === "authenticated") this.requestInitialProfileSnapshot(profile);
|
||||
}
|
||||
|
||||
public async replayTo(
|
||||
engine: OneTalkSyncEngine,
|
||||
profileCoordinator?: OneTalkContactProfileCoordinator | null,
|
||||
@@ -289,7 +377,7 @@ export class OneTalkPageRuntimeHost {
|
||||
)
|
||||
return;
|
||||
this.resumeBuyerDelivery(token, identity, engine, profile, buyer);
|
||||
this.requestProfileSnapshot(identity.channelAccountId, token, identity, engine, profile);
|
||||
if (profile) this.requestInitialProfileSnapshot(profile);
|
||||
await engine.handlePageReady(identity.channelAccountId, identity.conversationId);
|
||||
if (!this.isCurrentToken(token, identity, engine, profile)) return;
|
||||
await profile?.handlePageReady();
|
||||
|
||||
@@ -327,6 +327,7 @@ const dispatchPageObservation = (
|
||||
|
||||
const dispatchPageProfileObservation = (
|
||||
options: OneTalkServiceWorkerRuntimeOptions,
|
||||
connections: Map<string, PageConnection>,
|
||||
connection: PageConnection,
|
||||
message: OneTalkPageProfileObservedMessage,
|
||||
): void => {
|
||||
@@ -343,6 +344,8 @@ const dispatchPageProfileObservation = (
|
||||
}
|
||||
const onProfileMessage = options.onPageProfileMessage;
|
||||
if (!onProfileMessage) return;
|
||||
const conversationId = connection.conversationId;
|
||||
const conversationSelection = connection.conversationSelection;
|
||||
try {
|
||||
const persist = options.persistPageProfileObservation;
|
||||
const notify = (): void | Promise<void> => onProfileMessage(message, connection.sender);
|
||||
@@ -353,7 +356,16 @@ const dispatchPageProfileObservation = (
|
||||
return;
|
||||
}
|
||||
void Promise.resolve(persist(message, connection.sender, channelAccountId))
|
||||
.then(notify)
|
||||
.then(() => {
|
||||
if (
|
||||
!isCurrentConnection(connections, connection) ||
|
||||
connection.channelAccountId !== channelAccountId ||
|
||||
connection.conversationId !== conversationId ||
|
||||
connection.conversationSelection !== conversationSelection
|
||||
)
|
||||
return;
|
||||
return notify();
|
||||
})
|
||||
.catch((error) => {
|
||||
reportRuntimeError(options.onError, error);
|
||||
});
|
||||
@@ -448,7 +460,7 @@ const createPageMessageHandler = (
|
||||
dispatchPageObservation(options, connection, message);
|
||||
return;
|
||||
case "onetalk.page.profile-observed":
|
||||
dispatchPageProfileObservation(options, connection, message);
|
||||
dispatchPageProfileObservation(options, connections, connection, message);
|
||||
return;
|
||||
case "onetalk.page.buyer-facts-observed":
|
||||
dispatchPageBuyerFactsObservation(options, connection, message);
|
||||
@@ -491,7 +503,17 @@ const isAccountLevelCommand = (command: PageCommand): boolean => {
|
||||
return (
|
||||
command.action === "onetalk.sync" ||
|
||||
command.action === "onetalk.discover-conversations" ||
|
||||
command.action === "onetalk.contact.snapshot"
|
||||
command.action === "onetalk.contact.snapshot" ||
|
||||
command.action === "onetalk.contact.collect"
|
||||
);
|
||||
};
|
||||
|
||||
const isProfileCollectCommand = (command: PageCommand): boolean => {
|
||||
return (
|
||||
command.action === "onetalk.contact.collect" &&
|
||||
Object.keys(command).length === 2 &&
|
||||
typeof command.conversationId === "string" &&
|
||||
command.conversationId.length > 0
|
||||
);
|
||||
};
|
||||
|
||||
@@ -624,6 +646,12 @@ export const createOneTalkServiceWorkerRuntime = (
|
||||
});
|
||||
return rejectedBeforeSend("invalid_request");
|
||||
}
|
||||
if (
|
||||
route.command.action === "onetalk.contact.collect" &&
|
||||
!isProfileCollectCommand(route.command)
|
||||
) {
|
||||
return rejectedBeforeSend("invalid_request");
|
||||
}
|
||||
|
||||
let connection: PageConnection | null;
|
||||
if (route.conversationId === undefined) {
|
||||
|
||||
@@ -183,6 +183,7 @@ export type OneTalkContactProfileStore = {
|
||||
listPendingProfiles: (
|
||||
channelAccountId?: string,
|
||||
) => Promise<OneTalkContactProfileLedgerRecord[]>;
|
||||
hasProfileRecord: (channelAccountId: string) => Promise<boolean>;
|
||||
putPendingProfile: (
|
||||
channelAccountId: string,
|
||||
profile: OneTalkContactProfile,
|
||||
@@ -815,6 +816,15 @@ export const createOneTalkContactProfileStore = (
|
||||
);
|
||||
};
|
||||
|
||||
const hasProfileRecord = async (channelAccountId: string): Promise<boolean> => {
|
||||
const database = await getDatabase();
|
||||
const records = await readAll<OneTalkContactProfileLedgerRecord>(
|
||||
database,
|
||||
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
|
||||
);
|
||||
return records.some((record) => record.channelAccountId === channelAccountId);
|
||||
};
|
||||
|
||||
const putPendingProfile = async (
|
||||
channelAccountId: string,
|
||||
profile: OneTalkContactProfile,
|
||||
@@ -954,6 +964,7 @@ export const createOneTalkContactProfileStore = (
|
||||
return {
|
||||
getProfile,
|
||||
listPendingProfiles,
|
||||
hasProfileRecord,
|
||||
putPendingProfile,
|
||||
markProfileUploaded,
|
||||
discardPendingProfile,
|
||||
|
||||
@@ -156,6 +156,7 @@ export const createOneTalkServiceWorkerSyncController = (
|
||||
onPageDiagnostic: options.onPageDiagnostic,
|
||||
onEngineDiagnostic: options.onEngineDiagnostic,
|
||||
onProfileDiagnostic: options.onProfileDiagnostic,
|
||||
onProfileStatus: (profile, state) => pageHost.handleProfileStatus(profile, state.status),
|
||||
now: options.now,
|
||||
createRequestId: options.createRequestId,
|
||||
});
|
||||
|
||||
@@ -67,9 +67,6 @@ export const createOneTalkServiceWorkerSyncRuntime = (
|
||||
onDiagnostic: options.onProfileDiagnostic,
|
||||
})
|
||||
: undefined;
|
||||
const unsubscribeProfileStatus = options.bright.subscribeStatus?.((state) =>
|
||||
profile?.handleStatus(state),
|
||||
);
|
||||
pageHost = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => engine,
|
||||
getActiveProfileCoordinator: () => profile ?? null,
|
||||
@@ -77,6 +74,10 @@ export const createOneTalkServiceWorkerSyncRuntime = (
|
||||
onProfileDiagnostic: options.onProfileDiagnostic,
|
||||
onError: options.onError ?? (() => undefined),
|
||||
});
|
||||
const unsubscribeProfileStatus = options.bright.subscribeStatus?.((state) => {
|
||||
profile?.handleStatus(state);
|
||||
if (profile) pageHost.handleProfileStatus(profile, state.status);
|
||||
});
|
||||
let requestSequence = 0;
|
||||
const createRequestId =
|
||||
options.createRequestId ?? ((kind: string) => `runtime-${kind}-${++requestSequence}`);
|
||||
|
||||
@@ -66,8 +66,8 @@ class ProfileStore {
|
||||
this.records = new Map();
|
||||
}
|
||||
|
||||
async getProfile(channelAccountId, aliId) {
|
||||
return this.records.get(JSON.stringify([channelAccountId, aliId])) ?? null;
|
||||
async getProfile(channelAccountId, conversationId) {
|
||||
return this.records.get(JSON.stringify([channelAccountId, conversationId])) ?? null;
|
||||
}
|
||||
|
||||
async listPendingProfiles(channelAccountId) {
|
||||
@@ -78,8 +78,14 @@ class ProfileStore {
|
||||
);
|
||||
}
|
||||
|
||||
async hasProfileRecord(channelAccountId) {
|
||||
return [...this.records.values()].some(
|
||||
(record) => record.channelAccountId === channelAccountId,
|
||||
);
|
||||
}
|
||||
|
||||
async putPendingProfile(channelAccountId, profile) {
|
||||
const key = JSON.stringify([channelAccountId, profile.aliId]);
|
||||
const key = JSON.stringify([channelAccountId, profile.conversationId]);
|
||||
const current = this.records.get(key);
|
||||
if (current?.lastUploadedFingerprint === profile.profileFingerprint && !current.pending) {
|
||||
return current;
|
||||
@@ -87,6 +93,7 @@ class ProfileStore {
|
||||
const record = {
|
||||
key,
|
||||
channelAccountId,
|
||||
conversationId: profile.conversationId,
|
||||
aliId: profile.aliId,
|
||||
lastUploadedFingerprint: current?.lastUploadedFingerprint ?? null,
|
||||
pending: {
|
||||
@@ -100,8 +107,8 @@ class ProfileStore {
|
||||
return record;
|
||||
}
|
||||
|
||||
async markProfileUploaded({ channelAccountId, aliId, fingerprint, uploadedAt }) {
|
||||
const key = JSON.stringify([channelAccountId, aliId]);
|
||||
async markProfileUploaded({ channelAccountId, conversationId, fingerprint, uploadedAt }) {
|
||||
const key = JSON.stringify([channelAccountId, conversationId]);
|
||||
const current = this.records.get(key);
|
||||
if (current?.pending?.fingerprint !== fingerprint) return false;
|
||||
this.records.set(key, {
|
||||
@@ -113,6 +120,30 @@ class ProfileStore {
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
async discardPendingProfile({
|
||||
channelAccountId,
|
||||
conversationId,
|
||||
fingerprint,
|
||||
observedAtMs,
|
||||
rejectedAt,
|
||||
}) {
|
||||
const key = JSON.stringify([channelAccountId, conversationId]);
|
||||
const current = this.records.get(key);
|
||||
if (
|
||||
current?.pending?.fingerprint !== fingerprint ||
|
||||
current.pending.observedAtMs !== observedAtMs
|
||||
)
|
||||
return false;
|
||||
this.records.set(key, {
|
||||
...current,
|
||||
pending: undefined,
|
||||
lastRejectedFingerprint: fingerprint,
|
||||
lastRejectedObservedAtMs: observedAtMs,
|
||||
updatedAt: rejectedAt,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
const profile = {
|
||||
@@ -325,14 +356,16 @@ test("creates an account-scoped profile coordinator that carries avatarUrl", asy
|
||||
{ profiles: [profile], requestId: "profile-batch-1" },
|
||||
]);
|
||||
assert.equal(
|
||||
(await profileStore.getProfile("account-1", profile.aliId)).pending.profile.avatarUrl,
|
||||
(await profileStore.getProfile("account-1", profile.conversationId)).pending.profile
|
||||
.avatarUrl,
|
||||
profile.avatarUrl,
|
||||
);
|
||||
|
||||
await session.configure({ ...config, channelAccountId: "account-2", deviceId: "device-2" });
|
||||
assert.equal(session.getActive().profile.getChannelAccountId(), "account-2");
|
||||
assert.equal(
|
||||
(await profileStore.getProfile("account-1", profile.aliId)).pending.profile.avatarUrl,
|
||||
(await profileStore.getProfile("account-1", profile.conversationId)).pending.profile
|
||||
.avatarUrl,
|
||||
profile.avatarUrl,
|
||||
);
|
||||
});
|
||||
|
||||
@@ -32,9 +32,11 @@ const profile = (
|
||||
observationStatus: "confirmed",
|
||||
});
|
||||
|
||||
const createFixture = ({ holdFirstList = false, onDiagnostic } = {}) => {
|
||||
const createFixture = ({ holdFirstList = false, onDiagnostic, getProfile } = {}) => {
|
||||
const records = new Map();
|
||||
const sent = [];
|
||||
let getCount = 0;
|
||||
let putCount = 0;
|
||||
let listCount = 0;
|
||||
let listStarted;
|
||||
let releaseList;
|
||||
@@ -42,6 +44,13 @@ const createFixture = ({ holdFirstList = false, onDiagnostic } = {}) => {
|
||||
listStarted = resolve;
|
||||
});
|
||||
const store = {
|
||||
getProfile: async (channelAccountId, conversationId) => {
|
||||
getCount += 1;
|
||||
if (getProfile) return getProfile(channelAccountId, conversationId);
|
||||
return records.get(JSON.stringify([channelAccountId, conversationId])) ?? null;
|
||||
},
|
||||
hasProfileRecord: async (channelAccountId) =>
|
||||
[...records.values()].some((record) => record.channelAccountId === channelAccountId),
|
||||
listPendingProfiles: async () => {
|
||||
const snapshot = [...records.values()].filter((record) => record.pending);
|
||||
listCount += 1;
|
||||
@@ -53,6 +62,7 @@ const createFixture = ({ holdFirstList = false, onDiagnostic } = {}) => {
|
||||
return snapshot;
|
||||
},
|
||||
putPendingProfile: async (channelAccountId, next) => {
|
||||
putCount += 1;
|
||||
const key = JSON.stringify([channelAccountId, next.conversationId]);
|
||||
const current = records.get(key);
|
||||
if (
|
||||
@@ -144,7 +154,15 @@ const createFixture = ({ holdFirstList = false, onDiagnostic } = {}) => {
|
||||
now: () => 1_700_000_000_100,
|
||||
onDiagnostic,
|
||||
});
|
||||
return { coordinator, records, sent, firstListStarted, releaseList: () => releaseList?.() };
|
||||
return {
|
||||
coordinator,
|
||||
records,
|
||||
sent,
|
||||
getCount: () => getCount,
|
||||
putCount: () => putCount,
|
||||
firstListStarted,
|
||||
releaseList: () => releaseList?.(),
|
||||
};
|
||||
};
|
||||
|
||||
test("writes pending before send, replaces pending avatar, and ignores stale ACK", async () => {
|
||||
@@ -216,6 +234,91 @@ test("writes pending before send, replaces pending avatar, and ignores stale ACK
|
||||
assert.equal(fixture.sent.length, 2);
|
||||
});
|
||||
|
||||
test("silently short-circuits an uploaded fingerprint after one durable read", async () => {
|
||||
const diagnostics = [];
|
||||
const fixture = createFixture({ onDiagnostic: (event) => diagnostics.push(event) });
|
||||
const first = profile("First", "v1-first");
|
||||
|
||||
await fixture.coordinator.observe([first]);
|
||||
fixture.coordinator.handleFrame({
|
||||
type: "contact.profile.ack",
|
||||
protocolVersion: 2,
|
||||
connectionType: "plugin",
|
||||
requestId: "profile-batch-1",
|
||||
scope,
|
||||
payload: { status: "delivered", profileCount: 1 },
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
const getCount = fixture.getCount();
|
||||
const putCount = fixture.putCount();
|
||||
const sentCount = fixture.sent.length;
|
||||
const observedCount = diagnostics.filter((event) => event.event === "profile_observed").length;
|
||||
await fixture.coordinator.observe([first]);
|
||||
|
||||
assert.equal(fixture.getCount(), getCount + 1);
|
||||
assert.equal(fixture.putCount(), putCount);
|
||||
assert.equal(fixture.sent.length, sentCount);
|
||||
assert.equal(
|
||||
diagnostics.filter((event) => event.event === "profile_observed").length,
|
||||
observedCount,
|
||||
);
|
||||
});
|
||||
|
||||
test("does not write the same fingerprint twice when profile observations overlap", async () => {
|
||||
const fixture = createFixture();
|
||||
const first = profile("First", "v1-first");
|
||||
|
||||
await Promise.all([fixture.coordinator.observe([first]), fixture.coordinator.observe([first])]);
|
||||
|
||||
assert.equal(fixture.getCount(), 2);
|
||||
assert.equal(fixture.putCount(), 1);
|
||||
assert.equal(fixture.sent.length, 1);
|
||||
});
|
||||
|
||||
test("does not leave a rejected profile-write tracker promise unhandled", async () => {
|
||||
const fixture = createFixture({
|
||||
getProfile: async () => {
|
||||
throw new Error("profile read failed");
|
||||
},
|
||||
});
|
||||
const unhandled = [];
|
||||
const onUnhandledRejection = (reason) => unhandled.push(reason);
|
||||
process.on("unhandledRejection", onUnhandledRejection);
|
||||
|
||||
try {
|
||||
await assert.rejects(fixture.coordinator.observe([profile("First", "v1-first")]), {
|
||||
message: "profile read failed",
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
assert.deepEqual(unhandled, []);
|
||||
} finally {
|
||||
process.off("unhandledRejection", onUnhandledRejection);
|
||||
}
|
||||
});
|
||||
|
||||
test("does not write a profile after the coordinator is disposed during its durable read", async () => {
|
||||
let releaseProfileRead;
|
||||
const profileRead = new Promise((resolve) => {
|
||||
releaseProfileRead = resolve;
|
||||
});
|
||||
const diagnostics = [];
|
||||
const fixture = createFixture({
|
||||
getProfile: async () => profileRead,
|
||||
onDiagnostic: (event) => diagnostics.push(event),
|
||||
});
|
||||
const observation = fixture.coordinator.observe([profile("First", "v1-first")]);
|
||||
await Promise.resolve();
|
||||
|
||||
fixture.coordinator.dispose();
|
||||
releaseProfileRead(null);
|
||||
await observation;
|
||||
|
||||
assert.equal(fixture.putCount(), 0);
|
||||
assert.equal(fixture.sent.length, 0);
|
||||
assert.deepEqual(diagnostics, []);
|
||||
});
|
||||
|
||||
test("keeps the first maximum batch observation and isolates shared aliIds by conversation", async () => {
|
||||
const fixture = createFixture();
|
||||
const older = profile("Older", "v1-older", "shared-ali", null, "conversation-a", 100);
|
||||
@@ -274,14 +377,40 @@ test("keeps the first maximum batch observation and isolates shared aliIds by co
|
||||
assert.equal(otherRecord.pending, undefined);
|
||||
});
|
||||
|
||||
test("does not let an earlier ACK settle a newer same-fingerprint observation", async () => {
|
||||
test("caps extension outbound profile frames at fifty records", async () => {
|
||||
const profiles = (count) =>
|
||||
Array.from({ length: count }, (_, index) =>
|
||||
profile(
|
||||
`Profile ${index}`,
|
||||
`v1-${count}-${index}`,
|
||||
`ali-${count}-${index}`,
|
||||
null,
|
||||
`conversation-${count}-${index}`,
|
||||
),
|
||||
);
|
||||
const fifty = createFixture();
|
||||
await fifty.coordinator.observe(profiles(50));
|
||||
assert.deepEqual(
|
||||
fifty.sent.map((entry) => entry.profiles.length),
|
||||
[50],
|
||||
);
|
||||
|
||||
const fiftyOne = createFixture();
|
||||
await fiftyOne.coordinator.observe(profiles(51));
|
||||
assert.deepEqual(
|
||||
fiftyOne.sent.map((entry) => entry.profiles.length),
|
||||
[50, 1],
|
||||
);
|
||||
});
|
||||
|
||||
test("makes a newer same-fingerprint observation a read-only no-op", async () => {
|
||||
const fixture = createFixture();
|
||||
const earlier = profile("Same customer", "v1-same", "shared-ali", null, "conversation-1", 100);
|
||||
const later = profile("Same customer", "v1-same", "shared-ali", null, "conversation-1", 101);
|
||||
|
||||
await fixture.coordinator.observe([earlier]);
|
||||
await fixture.coordinator.observe([later]);
|
||||
assert.equal(fixture.sent.length, 2);
|
||||
assert.equal(fixture.sent.length, 1);
|
||||
|
||||
fixture.coordinator.handleFrame({
|
||||
type: "contact.profile.ack",
|
||||
@@ -292,28 +421,13 @@ test("does not let an earlier ACK settle a newer same-fingerprint observation",
|
||||
payload: { status: "delivered", profileCount: 1 },
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
assert.equal(
|
||||
fixture.records.get(JSON.stringify([scope.channelAccountId, later.conversationId])).pending
|
||||
.observedAtMs,
|
||||
later.observedAtMs,
|
||||
);
|
||||
|
||||
fixture.coordinator.handleFrame({
|
||||
type: "contact.profile.ack",
|
||||
protocolVersion: 2,
|
||||
connectionType: "plugin",
|
||||
requestId: "profile-batch-2",
|
||||
scope,
|
||||
payload: { status: "delivered", profileCount: 1 },
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
assert.equal(
|
||||
fixture.records.get(JSON.stringify([scope.channelAccountId, later.conversationId])).pending,
|
||||
undefined,
|
||||
);
|
||||
});
|
||||
|
||||
test("does not let a future error discard a newer same-fingerprint observation", async () => {
|
||||
test("does not reopen a pending profile for a newer same fingerprint", async () => {
|
||||
const fixture = createFixture();
|
||||
const earlier = profile("Same customer", "v1-same", "shared-ali", null, "conversation-1", 100);
|
||||
const later = profile("Same customer", "v1-same", "shared-ali", null, "conversation-1", 101);
|
||||
@@ -333,23 +447,8 @@ test("does not let a future error discard a newer same-fingerprint observation",
|
||||
const record = fixture.records.get(
|
||||
JSON.stringify([scope.channelAccountId, later.conversationId]),
|
||||
);
|
||||
assert.equal(record.pending.observedAtMs, later.observedAtMs);
|
||||
assert.equal(record.lastUploadedFingerprint, null);
|
||||
assert.equal(record.lastRejectedFingerprint, undefined);
|
||||
|
||||
fixture.coordinator.handleFrame({
|
||||
type: "contact.profile.ack",
|
||||
protocolVersion: 2,
|
||||
connectionType: "plugin",
|
||||
requestId: "profile-batch-2",
|
||||
scope,
|
||||
payload: { status: "delivered", profileCount: 1 },
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
assert.equal(
|
||||
fixture.records.get(JSON.stringify([scope.channelAccountId, later.conversationId])).pending,
|
||||
undefined,
|
||||
);
|
||||
assert.equal(record.pending, undefined);
|
||||
assert.equal(record.lastRejectedFingerprint, earlier.profileFingerprint);
|
||||
});
|
||||
|
||||
test("discards only the request-scoped future batch without marking it uploaded", async () => {
|
||||
@@ -603,7 +702,7 @@ test("sends a chunk update after a multi-chunk flush has already started", async
|
||||
|
||||
assert.deepEqual(
|
||||
fixture.sent.map((entry) => entry.profiles.length),
|
||||
[100, 1, 1],
|
||||
[50, 50, 1, 1],
|
||||
);
|
||||
assert.equal(fixture.sent.at(-1).profiles[0].profileFingerprint, "v2-50");
|
||||
});
|
||||
|
||||
@@ -71,7 +71,7 @@ test("constructs a whitelist profile without leaking sensitive row fields", () =
|
||||
assert.equal(JSON.stringify(profile).includes("secret"), false);
|
||||
});
|
||||
|
||||
test("emits snapshot, changed syncData, skips groups and unsubscribes on pagehide", () => {
|
||||
test("emits snapshot and every direct syncData observation, then unsubscribes on pagehide", () => {
|
||||
const fixture = createPage({ first: row() });
|
||||
const batches = [];
|
||||
const observer = installOneTalkContactProfileObserver(fixture.page, (profiles) => {
|
||||
@@ -84,19 +84,19 @@ test("emits snapshot, changed syncData, skips groups and unsubscribes on pagehid
|
||||
assert.equal(batches[0][0].aliId, "2208314000798");
|
||||
|
||||
fixture.page.emitSyncData({ first: row() });
|
||||
assert.equal(batches.length, 1);
|
||||
assert.equal(batches.length, 2);
|
||||
fixture.page.emitSyncData({
|
||||
first: row("Changed Name"),
|
||||
group: { ...row(), cid: "group-1", isGroup: true },
|
||||
incomplete: { cid: "missing-ali", name: "ignored" },
|
||||
});
|
||||
assert.equal(batches.length, 2);
|
||||
assert.equal(batches[1][0].name, "Changed Name");
|
||||
assert.equal(batches.length, 3);
|
||||
assert.equal(batches[2][0].name, "Changed Name");
|
||||
|
||||
fixture.page.pagehide();
|
||||
assert.equal(fixture.isUnsubscribed(), true);
|
||||
fixture.page.emitSyncData({ first: row("After Hide") });
|
||||
assert.equal(batches.length, 2);
|
||||
assert.equal(batches.length, 3);
|
||||
});
|
||||
|
||||
test("requires logged-in identity for a snapshot and never uses activeAccountId", () => {
|
||||
@@ -190,7 +190,7 @@ test("uses the SDK avatar priority, skips invalid candidates, and fingerprints a
|
||||
);
|
||||
});
|
||||
|
||||
test("snapshot and syncData share the same SDK extraction and avatar fingerprint", () => {
|
||||
test("snapshot and syncData share the same SDK extraction without MAIN-side profile dedupe", () => {
|
||||
const fixture = createPage({ first: row() });
|
||||
const batches = [];
|
||||
const observer = installOneTalkContactProfileObserver(fixture.page, (profiles) => {
|
||||
@@ -201,10 +201,32 @@ test("snapshot and syncData share the same SDK extraction and avatar fingerprint
|
||||
fixture.page.emitSyncData({ first: row() });
|
||||
assert.equal(snapshot[0].avatarUrl, batches[0][0].avatarUrl);
|
||||
assert.equal(snapshot[0].profileFingerprint, batches[0][0].profileFingerprint);
|
||||
assert.equal(batches.length, 1);
|
||||
fixture.page.emitSyncData({ first: row("Heena Liu", "https://cdn.example.test/changed.jpg") });
|
||||
assert.equal(batches.length, 2);
|
||||
assert.equal(batches[1][0].avatarUrl, "https://cdn.example.test/changed.jpg");
|
||||
assert.deepEqual(batches[1], batches[0]);
|
||||
fixture.page.emitSyncData({ first: row("Heena Liu", "https://cdn.example.test/changed.jpg") });
|
||||
assert.equal(batches.length, 3);
|
||||
assert.equal(batches[2][0].avatarUrl, "https://cdn.example.test/changed.jpg");
|
||||
});
|
||||
|
||||
test("collects only a requested direct conversation from the loaded SDK map", () => {
|
||||
const fixture = createPage({
|
||||
first: row(),
|
||||
second: { ...row("Other"), cid: "conversation-2" },
|
||||
});
|
||||
const batches = [];
|
||||
const observer = installOneTalkContactProfileObserver(fixture.page, (profiles) => {
|
||||
batches.push(profiles);
|
||||
});
|
||||
|
||||
assert.deepEqual(
|
||||
observer.collectConversation("conversation-2").map((profile) => profile.conversationId),
|
||||
["conversation-2"],
|
||||
);
|
||||
assert.deepEqual(observer.collectConversation("missing"), []);
|
||||
assert.deepEqual(
|
||||
batches.map((profiles) => profiles.map((profile) => profile.conversationId)),
|
||||
[["conversation-2"]],
|
||||
);
|
||||
});
|
||||
|
||||
const crmSeam = (capture) => ({
|
||||
|
||||
@@ -229,6 +229,14 @@ test("resets legacy OneTalk stores before v7 while opening the current database
|
||||
}
|
||||
});
|
||||
|
||||
test("reports whether an account has any durable profile ledger record", async () => {
|
||||
const store = createOneTalkContactProfileStore(new Factory(), () => 100);
|
||||
assert.equal(await store.hasProfileRecord("account-1"), false);
|
||||
await store.putPendingProfile("account-1", profile());
|
||||
assert.equal(await store.hasProfileRecord("account-1"), true);
|
||||
assert.equal(await store.hasProfileRecord("account-2"), false);
|
||||
});
|
||||
|
||||
test("does not return true before the readwrite transaction commits", async () => {
|
||||
const factory = new Factory();
|
||||
const store = createOneTalkContactProfileStore(factory, () => 100);
|
||||
|
||||
@@ -85,6 +85,7 @@ test("connects one publisher batch to the real send correlator and one page sink
|
||||
|
||||
test("routes a profile snapshot once and rejects an unknown page action", async () => {
|
||||
let snapshots = 0;
|
||||
const collected = [];
|
||||
const handler = createOneTalkPageCommandHandler(
|
||||
{},
|
||||
{
|
||||
@@ -95,6 +96,10 @@ test("routes a profile snapshot once and rejects an unknown page action", async
|
||||
snapshots += 1;
|
||||
return [{ conversationId: "conversation-1" }];
|
||||
},
|
||||
collectConversation: (conversationId) => {
|
||||
collected.push(conversationId);
|
||||
return [{ conversationId }];
|
||||
},
|
||||
},
|
||||
historyBootstrapProgress: {},
|
||||
},
|
||||
@@ -105,6 +110,22 @@ test("routes a profile snapshot once and rejects an unknown page action", async
|
||||
profileCount: 1,
|
||||
});
|
||||
assert.equal(snapshots, 1);
|
||||
assert.deepEqual(
|
||||
await handler({ command: { action: "onetalk.contact.snapshot", extra: true } }),
|
||||
{ status: "rejected_before_send", reason: "invalid_request" },
|
||||
);
|
||||
assert.equal(snapshots, 1);
|
||||
assert.deepEqual(
|
||||
await handler({
|
||||
command: { action: "onetalk.contact.collect", conversationId: "conversation-2" },
|
||||
}),
|
||||
{ status: "completed", profileCount: 1 },
|
||||
);
|
||||
assert.deepEqual(collected, ["conversation-2"]);
|
||||
assert.deepEqual(
|
||||
await handler({ command: { action: "onetalk.contact.collect", conversationId: "" } }),
|
||||
{ status: "rejected_before_send", reason: "invalid_request" },
|
||||
);
|
||||
assert.deepEqual(await handler({ command: { action: "unexpected.action" } }), {
|
||||
status: "rejected_before_send",
|
||||
reason: "invalid_request",
|
||||
|
||||
@@ -46,6 +46,14 @@ const profile = {
|
||||
observationStatus: "confirmed",
|
||||
};
|
||||
|
||||
const profileCoordinator = (channelAccountId, hasProfileRecord = false) => ({
|
||||
getChannelAccountId: () => channelAccountId,
|
||||
hasProfileRecord: async () => hasProfileRecord,
|
||||
isAuthenticated: () => true,
|
||||
handlePageReady: async () => {},
|
||||
handlePageDisconnected: () => {},
|
||||
});
|
||||
|
||||
class FakePort {
|
||||
constructor(sender, name = ONE_TALK_PAGE_PORT_NAME) {
|
||||
this.name = name;
|
||||
@@ -259,8 +267,10 @@ test("rejects the removed buyer collection command without posting it to the pag
|
||||
|
||||
test("requests only one profile snapshot for repeated hello from the same page identity", async () => {
|
||||
const engine = { handlePageReady: async () => {} };
|
||||
const activeProfile = profileCoordinator("account-unique");
|
||||
const host = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => engine,
|
||||
getActiveProfileCoordinator: () => activeProfile,
|
||||
getActiveChannelAccountId: () => "account-unique",
|
||||
onError: () => {},
|
||||
});
|
||||
@@ -287,6 +297,7 @@ test("requests a new profile snapshot when the page identity changes", async ()
|
||||
const engine = { handlePageReady: async () => {} };
|
||||
const host = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => engine,
|
||||
getActiveProfileCoordinator: () => null,
|
||||
onError: () => {},
|
||||
});
|
||||
const port = new FakePort(pageSender(31));
|
||||
@@ -297,7 +308,7 @@ test("requests a new profile snapshot when the page identity changes", async ()
|
||||
|
||||
assert.deepEqual(
|
||||
port.posted.map((message) => message.command?.action),
|
||||
["onetalk.contact.snapshot", "onetalk.contact.snapshot"],
|
||||
[],
|
||||
);
|
||||
});
|
||||
|
||||
@@ -323,6 +334,36 @@ test("does not pass an old page observation to a replacement-account engine", as
|
||||
assert.deepEqual(handled, []);
|
||||
});
|
||||
|
||||
test("routes only live message conversations to targeted profile collection", async () => {
|
||||
const engine = {
|
||||
handlePageObservation: async () => {},
|
||||
handlePageReady: async () => {},
|
||||
};
|
||||
const activeProfile = profileCoordinator("account-live", true);
|
||||
const host = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => engine,
|
||||
getActiveProfileCoordinator: () => activeProfile,
|
||||
getActiveChannelAccountId: () => "account-live",
|
||||
onError: () => {},
|
||||
});
|
||||
const port = new FakePort(pageSender(39));
|
||||
host.runtime.handleConnect(port);
|
||||
port.dispatchMessage(createOneTalkPageHelloMessage("account-live"));
|
||||
port.dispatchMessage(
|
||||
createOneTalkPageObservedMessage([
|
||||
observedMessage,
|
||||
{ ...observedMessage, messageId: "history-1", messageType: "history" },
|
||||
{ ...observedMessage, messageId: "live-duplicate" },
|
||||
]),
|
||||
);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
assert.deepEqual(
|
||||
port.posted.map((message) => message.command),
|
||||
[{ action: "onetalk.contact.collect", conversationId: "conversation-1" }],
|
||||
);
|
||||
});
|
||||
|
||||
test("durably gates profile messages and routes the account-level profile snapshot command", async () => {
|
||||
const events = [];
|
||||
const runtime = createOneTalkServiceWorkerRuntime({
|
||||
@@ -377,7 +418,7 @@ test("rejects a profile update whose page identity does not match the last hello
|
||||
assert.deepEqual(events, []);
|
||||
});
|
||||
|
||||
test("keeps profile observations on the current page account across an identity change", async () => {
|
||||
test("notifies only the current page identity after profile persistence", async () => {
|
||||
const persisted = [];
|
||||
const handled = [];
|
||||
const runtime = createOneTalkServiceWorkerRuntime({
|
||||
@@ -404,7 +445,35 @@ test("keeps profile observations on the current page account across an identity
|
||||
["account-a", profile.avatarUrl],
|
||||
["account-b", null],
|
||||
]);
|
||||
assert.deepEqual(handled, ["account-a", "account-b"]);
|
||||
assert.deepEqual(handled, ["account-b"]);
|
||||
});
|
||||
|
||||
test("does not notify the profile handler after the page identity changes during persistence", async () => {
|
||||
let releasePersistence;
|
||||
const persistenceStarted = new Promise((resolve) => {
|
||||
releasePersistence = resolve;
|
||||
});
|
||||
const persisted = [];
|
||||
const handled = [];
|
||||
const runtime = createOneTalkServiceWorkerRuntime({
|
||||
persistPageProfileObservation: async (message, _sender, channelAccountId) => {
|
||||
persisted.push([channelAccountId, message.profiles[0].avatarUrl]);
|
||||
await persistenceStarted;
|
||||
},
|
||||
onPageProfileMessage: (message) => handled.push(message.channelAccountId),
|
||||
onPageMessage: () => undefined,
|
||||
});
|
||||
const port = new FakePort(pageSender(38));
|
||||
connectPage(runtime, port, "account-a", undefined);
|
||||
port.dispatchMessage(createOneTalkPageProfileObservedMessage([profile], "account-a"));
|
||||
await Promise.resolve();
|
||||
|
||||
port.dispatchMessage(createOneTalkPageHelloMessage("account-b", undefined));
|
||||
releasePersistence();
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
assert.deepEqual(persisted, [["account-a", profile.avatarUrl]]);
|
||||
assert.deepEqual(handled, []);
|
||||
});
|
||||
|
||||
test("profile persistence failure does not block the independent message observer", async () => {
|
||||
|
||||
@@ -395,9 +395,16 @@ test("replay only resumes durable buyer delivery after a page hello", async () =
|
||||
test("reports snapshot route failure without blocking page readiness", async () => {
|
||||
const diagnostics = [];
|
||||
const engine = { handlePageReady: async () => {} };
|
||||
const activeProfile = {
|
||||
getChannelAccountId: () => "account-failure",
|
||||
hasProfileRecord: async () => false,
|
||||
isAuthenticated: () => true,
|
||||
handlePageReady: async () => {},
|
||||
handlePageDisconnected: () => {},
|
||||
};
|
||||
const host = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => engine,
|
||||
getActiveProfileCoordinator: () => null,
|
||||
getActiveProfileCoordinator: () => activeProfile,
|
||||
getActiveChannelAccountId: () => "account-failure",
|
||||
getConfigurationEpoch: () => 1,
|
||||
onProfileDiagnostic: (event) => diagnostics.push(event),
|
||||
@@ -421,6 +428,75 @@ test("reports snapshot route failure without blocking page readiness", async ()
|
||||
);
|
||||
});
|
||||
|
||||
test("standalone sync runtime requests the initial snapshot when Bright authenticates after page hello", async () => {
|
||||
const bright = new FakeBright({});
|
||||
const runtime = createOneTalkServiceWorkerSyncRuntime({
|
||||
scope: { channelAccountId: "account-standalone", deviceId: "device-standalone" },
|
||||
bright,
|
||||
store: new EmptyStore(),
|
||||
bootstrapStore: new EmptyBootstrapStore(),
|
||||
profileStore: {
|
||||
getProfile: async () => null,
|
||||
listPendingProfiles: async () => [],
|
||||
hasProfileRecord: async () => false,
|
||||
putPendingProfile: async () => {
|
||||
throw new Error("unexpected profile write");
|
||||
},
|
||||
markProfileUploaded: async () => false,
|
||||
discardPendingProfile: async () => false,
|
||||
},
|
||||
onError: () => {},
|
||||
});
|
||||
const port = new FakePort(pageSender);
|
||||
runtime.runtime.handleConnect(port);
|
||||
port.dispatchMessage(createOneTalkPageHelloMessage("account-standalone"));
|
||||
bright.emitStatus("authenticated");
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
assert.deepEqual(
|
||||
port.posted.map((message) => message.command?.action),
|
||||
["onetalk.contact.snapshot"],
|
||||
);
|
||||
runtime.dispose();
|
||||
});
|
||||
|
||||
test("does not route an initial snapshot after Bright authentication is lost during the ledger check", async () => {
|
||||
let authenticated = true;
|
||||
let releaseLedgerCheck;
|
||||
const ledgerCheck = new Promise((resolve) => {
|
||||
releaseLedgerCheck = resolve;
|
||||
});
|
||||
const profile = {
|
||||
getChannelAccountId: () => "account-auth-race",
|
||||
hasProfileRecord: () => ledgerCheck,
|
||||
isAuthenticated: () => authenticated,
|
||||
handleStatus: (state) => {
|
||||
authenticated = state.status === "authenticated";
|
||||
},
|
||||
handlePageReady: async () => {},
|
||||
handlePageDisconnected: () => {},
|
||||
};
|
||||
const host = new OneTalkPageRuntimeHost({
|
||||
getActiveEngine: () => ({ handlePageReady: async () => {} }),
|
||||
getActiveProfileCoordinator: () => profile,
|
||||
getActiveChannelAccountId: () => "account-auth-race",
|
||||
onError: () => {},
|
||||
});
|
||||
const port = new FakePort(pageSender);
|
||||
host.runtime.handleConnect(port);
|
||||
port.dispatchMessage(createOneTalkPageHelloMessage("account-auth-race"));
|
||||
await Promise.resolve();
|
||||
|
||||
profile.handleStatus({ status: "closed", permissions: [] });
|
||||
releaseLedgerCheck(false);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
assert.deepEqual(
|
||||
port.posted.filter((message) => message.type === "onetalk.page.command"),
|
||||
[],
|
||||
);
|
||||
});
|
||||
|
||||
test("disposes the runtime's Bright subscriptions and socket", () => {
|
||||
const bright = new FakeBright({});
|
||||
const runtime = createOneTalkServiceWorkerSyncRuntime({
|
||||
@@ -463,11 +539,17 @@ test("does not let a delayed old page callback touch the replacement account", a
|
||||
let oldProfileReadyCalls = 0;
|
||||
let nextProfileReadyCalls = 0;
|
||||
const oldProfile = {
|
||||
getChannelAccountId: () => "account-a",
|
||||
hasProfileRecord: async () => false,
|
||||
isAuthenticated: () => true,
|
||||
handlePageReady: async () => {
|
||||
oldProfileReadyCalls += 1;
|
||||
},
|
||||
};
|
||||
const nextProfile = {
|
||||
getChannelAccountId: () => "account-b",
|
||||
hasProfileRecord: async () => false,
|
||||
isAuthenticated: () => true,
|
||||
handlePageReady: async () => {
|
||||
nextProfileReadyCalls += 1;
|
||||
},
|
||||
@@ -502,6 +584,6 @@ test("does not let a delayed old page callback touch the replacement account", a
|
||||
assert.equal(oldProfileReadyCalls, 0);
|
||||
assert.equal(port.posted.length, initialPostCount);
|
||||
assert.equal(nextProfileReadyCalls, 1);
|
||||
assert.equal(replacementPort.posted.length, replacementPostCount);
|
||||
assert.equal(replacementPort.posted.length, replacementPostCount + 1);
|
||||
assert.equal(replacementPort.posted.at(-1).command.action, "onetalk.contact.snapshot");
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user