Files
trade-message-center/apps/chrome-extension/test/onetalk-sync-storage.test.js
T

710 lines
26 KiB
JavaScript

// 验证 OneTalk 同步账本的迁移、幂等与异常合并
import assert from "node:assert/strict";
import test from "node:test";
import { createOneTalkRenderedCardContentFingerprint } from "@trade-message-center/onetalk-contract";
import {
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_BUYER_FACT_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_SYNC_DATABASE_VERSION,
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
createOneTalkRenderedCardLedgerStore,
createOneTalkSyncStore,
createOneTalkConversationBootstrapStore,
} from "../src/onetalk/service-worker/storage.ts";
const validMessage = {
messageId: "message-1",
conversationId: "conversation-1",
senderId: "sender-1",
direction: "received",
sentAtMs: 100,
content: { version: 1, kind: "text", text: "hello" },
participantIds: ["sender-1", "login-1"],
readStatus: 0,
messageStatus: 1,
unreadCount: 0,
};
const v9IndexesByStore = new Map([
[
ONE_TALK_MESSAGE_STORE_NAME,
[["by_account_conversation", ["channelAccountId", "conversationId"]]],
],
[ONE_TALK_CHECKPOINT_STORE_NAME, [["by_account", "channelAccountId"]]],
[
ONE_TALK_CANDIDATE_STORE_NAME,
[
["by_account_status", ["channelAccountId", "status"]],
["by_account_conversation_status", ["channelAccountId", "conversationId", "status"]],
["by_account_conversation", ["channelAccountId", "conversationId"]],
],
],
[
ONE_TALK_ANOMALY_STORE_NAME,
[
["by_account", "channelAccountId"],
["by_account_conversation", ["channelAccountId", "conversationId"]],
],
],
[
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
[
["by_account", "channelAccountId"],
["by_account_pending_observed_at", ["channelAccountId", "pending.observedAtMs"]],
],
],
]);
class FakeRequest {
constructor(result, schedule = true) {
this.result = result;
this.error = null;
this.onsuccess = null;
this.onerror = null;
if (schedule) queueMicrotask(() => this.onsuccess?.());
}
}
class FakeStore {
constructor() {
this.records = new Map();
this.objectStoreGetAllCalls = 0;
this.indexNames = {
values: new Set(),
contains: (name) => this.indexNames.values.has(name),
};
}
put(record) {
this.records.set(record.key, structuredClone(record));
}
get(key) {
return new FakeRequest(this.records.get(key));
}
getAll() {
this.objectStoreGetAllCalls += 1;
return new FakeRequest([...this.records.values()].map((record) => structuredClone(record)));
}
createIndex(name, keyPath) {
this.indexNames.values.add(name);
this.indexes ??= new Map();
this.indexes.set(name, keyPath);
}
index(name) {
const keyPath = this.indexes?.get(name);
if (keyPath === undefined) throw new Error(`Missing index: ${name}`);
const keyFor = (record) => {
const valueFor = (path) =>
path.split(".").reduce((value, segment) => value?.[segment], record);
return Array.isArray(keyPath) ? keyPath.map(valueFor) : valueFor(keyPath);
};
const matches = (record, query) => {
const key = keyFor(record);
if (query?.lower !== undefined && Array.isArray(key)) {
return key[0] === query.lower[0];
}
return JSON.stringify(key) === JSON.stringify(query);
};
const matchingEntries = (query) =>
[...this.records.entries()].filter(([, record]) => matches(record, query));
return {
getAll: (query) =>
new FakeRequest(
matchingEntries(query).map(([, record]) => structuredClone(record)),
),
getAllKeys: (query) => new FakeRequest(matchingEntries(query).map(([key]) => key)),
count: (query) => new FakeRequest(matchingEntries(query).length),
};
}
clear() {
this.records.clear();
}
delete(key) {
this.records.delete(key);
}
openCursor() {
const entries = [...this.records.entries()];
const request = new FakeRequest(null, false);
let index = 0;
const advance = () => {
if (index >= entries.length) {
request.result = null;
request.onsuccess?.();
return;
}
const [key, record] = entries[index++];
request.result = {
value: structuredClone(record),
update: (updated) => this.records.set(key, structuredClone(updated)),
continue: () => queueMicrotask(advance),
};
request.onsuccess?.();
};
queueMicrotask(advance);
return request;
}
}
class FakeTransaction {
constructor(database, storeNames) {
this.database = database;
this.storeNames = Array.isArray(storeNames) ? storeNames : [storeNames];
this.error = null;
this.oncomplete = null;
this.onerror = null;
this.onabort = null;
setTimeout(() => this.oncomplete?.(), 0);
}
objectStore(name) {
assert.equal(this.storeNames.includes(name), true);
return this.database.stores.get(name);
}
}
class FakeDatabase {
constructor() {
this.stores = new Map();
this.objectStoreNames = {
contains: (name) => this.stores.has(name),
};
}
createObjectStore(name) {
const store = new FakeStore();
this.stores.set(name, store);
return store;
}
transaction(storeNames) {
return new FakeTransaction(this, storeNames);
}
}
class FakeFactory {
constructor(version = 0) {
this.database = new FakeDatabase();
this.database.version = version;
}
open(_name, requestedVersion) {
const request = new FakeRequest(this.database, false);
const version = requestedVersion ?? 1;
if (this.database.version < version) {
const transaction = new FakeTransaction(this.database, [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
]);
request.transaction = transaction;
transaction.oncomplete = () => request.onsuccess?.();
queueMicrotask(() => {
const oldVersion = this.database.version;
this.database.version = version;
request.onupgradeneeded?.({ oldVersion, newVersion: version });
});
} else {
queueMicrotask(() => request.onsuccess?.());
}
return request;
}
}
const createVersionTwoFactory = () => {
const factory = new FakeFactory(2);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
]) {
factory.database.createObjectStore(storeName);
}
return factory;
};
const checkpoint = {
key: JSON.stringify(["account-1", "conversation-1"]),
channelAccountId: "account-1",
conversationId: "conversation-1",
mode: "incremental",
phase: "awaiting_anchor",
anchorState: "awaiting_anchor",
anchorMessageId: "old-message",
pageCursor: null,
pageTimeStamp: 500,
historyComplete: false,
syncResult: "incomplete",
latestMessageId: null,
lastObservedAt: 500,
updatedAt: 500,
};
test("creates durable stores, keeps confirmed candidates, and merges anomalies", async () => {
const factory = new FakeFactory();
const store = createOneTalkSyncStore(factory, () => 500);
const first = await store.persistObservedBatch({
channelAccountId: "account-1",
conversationId: "conversation-1",
messages: [validMessage, { conversationId: "conversation-1" }],
observationSource: "incremental",
mode: "incremental",
receivedAt: 500,
});
assert.equal(first.candidates[0].status, "awaiting_anchor");
assert.equal(first.anomalies[0].code, "missing_message_id");
await store.updateCandidate(first.candidates[0], "confirmed", "transient-request");
await store.persistObservedBatch({
channelAccountId: "account-1",
conversationId: "conversation-1",
messages: [validMessage],
observationSource: "incremental",
mode: "incremental",
receivedAt: 501,
});
const candidate = await store.getCandidate("account-1", "conversation-1", "message-1");
assert.equal(candidate.status, "confirmed");
assert.equal("requestId" in candidate, false);
await store.persistObservedBatch({
channelAccountId: "account-1",
conversationId: "conversation-1",
messages: [{ conversationId: "conversation-1" }],
observationSource: "incremental",
mode: "incremental",
receivedAt: 502,
});
const anomalies = await store.listAnomalies("account-1", "conversation-1");
assert.equal(anomalies.length, 1);
assert.equal(anomalies[0].occurrenceCount, 2);
await store.putCheckpoint(checkpoint);
assert.deepEqual(await store.getCheckpoint("account-1", "conversation-1"), checkpoint);
assert.deepEqual(
[...factory.database.stores.keys()].sort(),
[
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
"onetalk_contact_profiles",
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
ONE_TALK_BUYER_FACT_STORE_NAME,
].sort(),
);
});
test("preserves existing stores while adding new stores from v7 and v8", async () => {
for (const oldVersion of [7, 8]) {
const factory = new FakeFactory(oldVersion);
const extensionStorage = new Map([
["onetalk.config", { channelAccountId: "account-1" }],
["onetalk.deviceId", "device-1"],
]);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
]) {
factory.database.createObjectStore(storeName).put({ key: storeName, legacy: true });
}
const store = createOneTalkSyncStore(factory, () => 500);
assert.deepEqual(await store.listCheckpoints("account-1"), []);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
]) {
assert.equal(factory.database.stores.get(storeName).records.size, 1);
}
assert.equal(factory.database.stores.has(ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME), true);
assert.deepEqual(
[...extensionStorage.entries()],
[
["onetalk.config", { channelAccountId: "account-1" }],
["onetalk.deviceId", "device-1"],
],
);
}
});
test("preserves all v8 ledger records while creating v9 query indexes", async () => {
const factory = new FakeFactory(8);
const v8StoreNames = [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
];
for (const storeName of v8StoreNames) {
factory.database.createObjectStore(storeName).put({ key: storeName, v8: true });
}
const store = createOneTalkSyncStore(factory, () => 500);
await store.getCheckpoint("account-1", "conversation-1");
for (const storeName of v8StoreNames) {
assert.equal(factory.database.stores.get(storeName).records.get(storeName).v8, true);
}
for (const [storeName, indexes] of v9IndexesByStore) {
assert.deepEqual([...factory.database.stores.get(storeName).indexes], indexes);
}
});
test("preserves v9 ledger records while adding the buyer facts store in v10", async () => {
const factory = new FakeFactory(9);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
]) {
factory.database.createObjectStore(storeName).put({ key: storeName, v9: true });
}
for (const [storeName, indexes] of v9IndexesByStore) {
const store = factory.database.stores.get(storeName);
for (const [name, keyPath] of indexes) store.createIndex(name, keyPath);
}
const store = createOneTalkSyncStore(factory, () => 500);
await store.getCheckpoint("account-1", "conversation-1");
assert.equal(factory.database.version, ONE_TALK_SYNC_DATABASE_VERSION);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
]) {
assert.equal(factory.database.stores.get(storeName).records.get(storeName).v9, true);
}
assert.equal(factory.database.stores.has(ONE_TALK_BUYER_FACT_STORE_NAME), true);
for (const [storeName, indexes] of v9IndexesByStore) {
assert.deepEqual([...factory.database.stores.get(storeName).indexes], indexes);
}
});
test("preserves v9 rendered-card records and reads pending ACKs through the account-status index", async () => {
const factory = new FakeFactory(9);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_CONTACT_PROFILE_STORE_NAME,
ONE_TALK_CONVERSATION_BOOTSTRAP_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
]) {
factory.database.createObjectStore(storeName);
}
const renderedLedger = factory.database.stores.get(ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME);
const content = {
version: 1,
kind: "rendered_order",
title: "Order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Address", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
const pendingAccountOne = {
key: JSON.stringify(["account-1", "conversation-1", "message-1"]),
channelAccountId: "account-1",
conversationId: "conversation-1",
messageId: "message-1",
content,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(content),
observedAtMs: 100,
baseDirection: "received",
baseSentAtMs: 99,
status: "pending_ack",
firstObservedAt: 100,
updatedAt: 100,
};
renderedLedger.put(pendingAccountOne);
renderedLedger.put({
...pendingAccountOne,
key: JSON.stringify(["account-2", "conversation-2", "message-2"]),
channelAccountId: "account-2",
conversationId: "conversation-2",
messageId: "message-2",
});
renderedLedger.put({
...pendingAccountOne,
key: JSON.stringify(["account-1", "conversation-3", "message-3"]),
conversationId: "conversation-3",
messageId: "message-3",
status: "confirmed",
});
const store = createOneTalkRenderedCardLedgerStore(factory, () => 500);
assert.deepEqual(await store.listPending("account-1"), [pendingAccountOne]);
assert.equal(factory.database.version, ONE_TALK_SYNC_DATABASE_VERSION);
assert.equal(renderedLedger.indexNames.contains("by_account_status"), true);
assert.equal(renderedLedger.indexNames.contains("by_account_conversation"), true);
assert.equal(renderedLedger.objectStoreGetAllCalls, 0);
});
test("clears only one conversation history ledger after its transaction commits", async () => {
const factory = new FakeFactory();
const store = createOneTalkSyncStore(factory, () => 500);
const renderedCardStore = createOneTalkRenderedCardLedgerStore(factory, () => 500);
const renderedCardContent = {
version: 1,
kind: "rendered_order",
title: "Order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Address", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
await renderedCardStore.observe({
channelAccountId: "account-1",
observation: {
conversationId: "conversation-1",
messageId: "message-1",
content: renderedCardContent,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(renderedCardContent),
observedAtMs: 500,
},
baseEvidence: { direction: "received", sentAtMs: 499 },
observedAt: 500,
});
for (const [channelAccountId, conversationId, messageId] of [
["account-1", "conversation-2", "message-2"],
["account-2", "conversation-1", "message-3"],
]) {
await renderedCardStore.observe({
channelAccountId,
observation: {
conversationId,
messageId,
content: renderedCardContent,
contentFingerprint:
createOneTalkRenderedCardContentFingerprint(renderedCardContent),
observedAtMs: 500,
},
baseEvidence: { direction: "received", sentAtMs: 499 },
observedAt: 500,
});
}
await store.persistObservedBatch({
channelAccountId: "account-1",
conversationId: "conversation-1",
messages: [validMessage],
observationSource: "history",
mode: "full",
});
await store.persistObservedBatch({
channelAccountId: "account-1",
conversationId: "conversation-2",
messages: [{ ...validMessage, conversationId: "conversation-2", messageId: "message-2" }],
observationSource: "history",
mode: "full",
});
await store.putCheckpoint(checkpoint);
await store.putCheckpoint({
...checkpoint,
key: JSON.stringify(["account-1", "conversation-2"]),
conversationId: "conversation-2",
});
await store.recordAnomaly({
key: "conversation-anomaly",
channelAccountId: "account-1",
conversationId: "conversation-1",
code: "invalid",
observationSource: "sync",
fields: [],
occurrenceCount: 1,
firstObservedAt: 1,
lastObservedAt: 1,
});
await store.recordAnomaly({
key: "account-anomaly",
channelAccountId: "account-1",
code: "invalid",
observationSource: "sync",
fields: [],
occurrenceCount: 1,
firstObservedAt: 1,
lastObservedAt: 1,
});
await store.clearConversationHistory("account-1", "conversation-1");
assert.equal(await renderedCardStore.get("account-1", "conversation-1", "message-1"), null);
assert.notEqual(await renderedCardStore.get("account-1", "conversation-2", "message-2"), null);
assert.notEqual(await renderedCardStore.get("account-2", "conversation-1", "message-3"), null);
assert.equal(await store.getCheckpoint("account-1", "conversation-1"), null);
assert.deepEqual(await store.listCandidates("account-1", "conversation-1"), []);
assert.deepEqual(await store.listAnomalies("account-1", "conversation-1"), []);
assert.notEqual(await store.getCheckpoint("account-1", "conversation-2"), null);
assert.equal((await store.listCandidates("account-1", "conversation-2")).length, 1);
assert.equal((await store.listPendingCandidates("account-1")).length, 1);
assert.equal((await store.listAnomalies("account-1")).length, 1);
for (const storeName of [
ONE_TALK_MESSAGE_STORE_NAME,
ONE_TALK_CHECKPOINT_STORE_NAME,
ONE_TALK_CANDIDATE_STORE_NAME,
ONE_TALK_ANOMALY_STORE_NAME,
ONE_TALK_RENDERED_CARD_LEDGER_STORE_NAME,
]) {
assert.equal(factory.database.stores.get(storeName).objectStoreGetAllCalls, 0);
}
});
test("stores bootstrap markers by account and stable migration identifier", async () => {
const factory = new FakeFactory();
const store = createOneTalkConversationBootstrapStore(factory);
await store.putConfirmed({
channelAccountId: "account-1",
migrationId: "direct-discovery-before-history-v1",
phase: "history",
confirmedBatchId: "batch-1",
discoveryOrder: ["conversation-1"],
terminalStates: { "conversation-1": "pending" },
historyRetries: { "conversation-1": 0 },
updatedAt: 500,
});
const marker = await store.get("account-1", "direct-discovery-before-history-v1");
assert.equal(marker.confirmedBatchId, "batch-1");
assert.equal(await store.get("account-2", "direct-discovery-before-history-v1"), null);
});
test("preserves the first rendered-card snapshot across concurrent same-key observations", async () => {
const factory = new FakeFactory();
const store = createOneTalkRenderedCardLedgerStore(factory, () => 500);
const firstContent = {
version: 1,
kind: "rendered_order",
title: "First order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Address", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
const secondContent = { ...firstContent, title: "Second order" };
const first = {
conversationId: "conversation-1",
messageId: "message-1",
content: firstContent,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(firstContent),
observedAtMs: 500,
};
const second = {
...first,
content: secondContent,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(secondContent),
observedAtMs: 501,
};
await Promise.all([
store.observe({
channelAccountId: "account-1",
observation: first,
baseEvidence: { direction: "received", sentAtMs: 499 },
observedAt: 500,
}),
store.observe({
channelAccountId: "account-1",
observation: second,
baseEvidence: { direction: "received", sentAtMs: 499 },
observedAt: 501,
}),
]);
const record = await store.get("account-1", "conversation-1", "message-1");
assert.deepEqual(record.content, firstContent);
assert.equal(record.lastConflictingFingerprint, second.contentFingerprint);
});
test("keeps the first pending ACK correlation while serializing same-key ledger writes", async () => {
const factory = new FakeFactory();
const store = createOneTalkRenderedCardLedgerStore(factory, () => 500);
const content = {
version: 1,
kind: "rendered_order",
title: "Order",
products: [],
productCount: 0,
status: { code: null, text: "Paid" },
payment: { totalDisplay: "$1", discountDisplay: null },
delivery: { shippingAddress: "Address", methodLabel: null, dateLabel: null },
action: { label: null, status: null },
};
const first = {
conversationId: "conversation-1",
messageId: "message-1",
content,
contentFingerprint: createOneTalkRenderedCardContentFingerprint(content),
observedAtMs: 500,
};
const second = { ...first, observedAtMs: 501 };
const baseEvidence = { direction: "received", sentAtMs: 499 };
await store.observe({ channelAccountId: "account-1", observation: first, baseEvidence });
await store.markSent({
channelAccountId: "account-1",
conversationId: first.conversationId,
messageId: first.messageId,
contentFingerprint: first.contentFingerprint,
observedAtMs: first.observedAtMs,
requestId: "request-1",
});
await Promise.all([
store.observe({ channelAccountId: "account-1", observation: second, baseEvidence }),
store.markAcknowledged({
channelAccountId: "account-1",
conversationId: first.conversationId,
messageId: first.messageId,
contentFingerprint: first.contentFingerprint,
observedAtMs: first.observedAtMs,
requestId: "request-1",
status: "accepted",
}),
]);
const record = await store.get("account-1", first.conversationId, first.messageId);
assert.equal(record.observedAtMs, first.observedAtMs);
assert.equal(record.status, "confirmed");
});