mirror of
https://github.com/sinanyuntu/trade-message-center.git
synced 2026-09-17 13:22:11 +08:00
726 lines
27 KiB
TypeScript
726 lines
27 KiB
TypeScript
// 验证 OneTalk 领域入库与锚点规则
|
|
|
|
import assert from "node:assert/strict";
|
|
import test from "node:test";
|
|
|
|
import type {
|
|
OneTalkMessage,
|
|
OneTalkObservedMessage,
|
|
} from "@trade-message-center/onetalk-contract";
|
|
|
|
import {
|
|
createOneTalkRepository,
|
|
createOneTalkService,
|
|
type OneTalkConversationState,
|
|
type OneTalkRepository,
|
|
type OneTalkSourceContext,
|
|
} from "../src/onetalk/index.ts";
|
|
import {
|
|
onetalkConversation,
|
|
onetalkMessage,
|
|
onetalkRenderedCardContent,
|
|
} from "../src/database/schema/onetalk.ts";
|
|
|
|
const context: OneTalkSourceContext = {
|
|
binding: "binding-1",
|
|
mindUserId: "mind-user-1",
|
|
workspaceId: "workspace-1",
|
|
channelAccountId: "account-1",
|
|
deviceId: "device-1",
|
|
};
|
|
|
|
const message = (overrides: Partial<OneTalkObservedMessage> = {}): OneTalkObservedMessage => {
|
|
return {
|
|
messageId: "message-1",
|
|
conversationId: "conversation-1",
|
|
senderId: "sender-1",
|
|
direction: "received",
|
|
sentAtMs: 1_700_000_000_000,
|
|
content: { version: 1, kind: "text", text: "hello" },
|
|
participantIds: ["sender-1", "login-user-1"],
|
|
readStatus: 1,
|
|
messageStatus: 2,
|
|
unreadCount: 0,
|
|
...overrides,
|
|
};
|
|
};
|
|
|
|
const createRepositoryHarness = () => {
|
|
const conversations = new Map<string, OneTalkConversationState>();
|
|
const messages = new Map<string, OneTalkMessage>();
|
|
const syncUpdates: Parameters<OneTalkRepository["updateSyncState"]>[1][] = [];
|
|
const anomalies = new Map<
|
|
string,
|
|
{ count: number; input: Parameters<OneTalkRepository["recordAnomaly"]>[0] }
|
|
>();
|
|
const observationKey = (conversationId: string, messageId: string): string =>
|
|
`${context.channelAccountId}:${conversationId}:${messageId}`;
|
|
|
|
const repository: OneTalkRepository = {
|
|
findConversation: async (_sourceContext, conversationId) => {
|
|
return conversations.get(`${context.channelAccountId}:${conversationId}`) ?? null;
|
|
},
|
|
discoverConversation: async (sourceContext, conversationId, conversationKind) => {
|
|
const key = `${sourceContext.channelAccountId}:${conversationId}`;
|
|
const existing = conversations.get(key);
|
|
if (existing) {
|
|
const updated = { ...existing, conversationKind };
|
|
conversations.set(key, updated);
|
|
return updated;
|
|
}
|
|
const created: OneTalkConversationState = {
|
|
channelAccountId: sourceContext.channelAccountId,
|
|
conversationId,
|
|
conversationKind,
|
|
syncPhase: "initial",
|
|
syncResult: "incomplete",
|
|
latestMessageId: null,
|
|
historyComplete: false,
|
|
messageCount: 0,
|
|
historyGeneration: "initial",
|
|
};
|
|
conversations.set(key, created);
|
|
return created;
|
|
},
|
|
guardedDiscoverConversation: async (
|
|
sourceContext,
|
|
conversationId,
|
|
guard,
|
|
conversationKind,
|
|
) => {
|
|
guard.assertValid();
|
|
const result = await repository.discoverConversation(
|
|
sourceContext,
|
|
conversationId,
|
|
conversationKind,
|
|
);
|
|
guard.assertValid();
|
|
return result;
|
|
},
|
|
discoverConversations: async (sourceContext, entries) =>
|
|
Promise.all(
|
|
entries.map((entry) =>
|
|
repository.discoverConversation(sourceContext, entry.conversationId, "direct"),
|
|
),
|
|
),
|
|
guardedDiscoverConversations: async (sourceContext, entries, guard) => {
|
|
guard.assertValid();
|
|
const result = await repository.discoverConversations(sourceContext, entries);
|
|
guard.assertValid();
|
|
return result;
|
|
},
|
|
listAnchors: async (sourceContext) => {
|
|
return [...conversations.values()]
|
|
.filter(
|
|
(conversation) =>
|
|
conversation.channelAccountId === sourceContext.channelAccountId,
|
|
)
|
|
.map(({ conversationId, latestMessageId, historyGeneration }) => ({
|
|
conversationId,
|
|
latestMessageId,
|
|
historyGeneration,
|
|
}));
|
|
},
|
|
listConversations: async (sourceScope) => {
|
|
return [...conversations.values()]
|
|
.filter(
|
|
(conversation) =>
|
|
conversation.channelAccountId === sourceScope.channelAccountId,
|
|
)
|
|
.sort((left, right) => left.conversationId.localeCompare(right.conversationId));
|
|
},
|
|
findConversationForRead: async (sourceScope, conversationId) => {
|
|
return conversations.get(`${sourceScope.channelAccountId}:${conversationId}`) ?? null;
|
|
},
|
|
listMessagesForRead: async (sourceScope, conversationId, cursor, limit) => {
|
|
const rows = [...messages.values()]
|
|
.filter(
|
|
(stored) =>
|
|
stored.conversationId === conversationId &&
|
|
sourceScope.channelAccountId === context.channelAccountId,
|
|
)
|
|
.filter(
|
|
(stored) =>
|
|
cursor === null ||
|
|
stored.sentAtMs > cursor.sentAtMs ||
|
|
(stored.sentAtMs === cursor.sentAtMs &&
|
|
stored.messageId > cursor.messageId),
|
|
)
|
|
.sort(
|
|
(left, right) =>
|
|
left.sentAtMs - right.sentAtMs ||
|
|
left.messageId.localeCompare(right.messageId),
|
|
);
|
|
const hasMore = rows.length > limit;
|
|
const pageRows = hasMore ? rows.slice(0, limit) : rows;
|
|
const last = pageRows.at(-1);
|
|
return {
|
|
messages: pageRows,
|
|
hasMore,
|
|
nextCursor:
|
|
hasMore && last
|
|
? {
|
|
channelAccountId: sourceScope.channelAccountId,
|
|
conversationId,
|
|
sentAtMs: last.sentAtMs,
|
|
messageId: last.messageId,
|
|
}
|
|
: null,
|
|
};
|
|
},
|
|
hasMessage: async (sourceContext, conversationId, messageId) => {
|
|
return messages.has(`${sourceContext.channelAccountId}:${conversationId}:${messageId}`);
|
|
},
|
|
guardedHasMessage: async (sourceContext, conversationId, messageId, guard) => {
|
|
guard.assertValid();
|
|
const result = messages.has(
|
|
`${sourceContext.channelAccountId}:${conversationId}:${messageId}`,
|
|
);
|
|
guard.assertValid();
|
|
return result;
|
|
},
|
|
insertMessage: async (
|
|
sourceContext,
|
|
historyGeneration,
|
|
observationSource,
|
|
observedMessage,
|
|
) => {
|
|
const conversationKey = `${sourceContext.channelAccountId}:${observedMessage.conversationId}`;
|
|
const conversation = conversations.get(conversationKey);
|
|
if (!conversation) return { status: "rejected", reason: "conversation_not_discovered" };
|
|
if (conversation.historyGeneration !== historyGeneration)
|
|
return { status: "rejected", reason: "history_generation_mismatch" };
|
|
|
|
const key = observationKey(observedMessage.conversationId, observedMessage.messageId);
|
|
const existing = messages.get(key);
|
|
if (existing) return { status: "duplicate", message: existing };
|
|
|
|
const stored: OneTalkMessage = { ...observedMessage };
|
|
messages.set(key, stored);
|
|
conversations.set(conversationKey, {
|
|
...conversation,
|
|
messageCount: conversation.messageCount + 1,
|
|
});
|
|
assert.ok(observationSource);
|
|
return { status: "accepted", message: stored };
|
|
},
|
|
guardedInsertMessage: async (
|
|
sourceContext,
|
|
historyGeneration,
|
|
observationSource,
|
|
observedMessage,
|
|
guard,
|
|
) => {
|
|
guard.assertValid();
|
|
const result = await repository.insertMessage(
|
|
sourceContext,
|
|
historyGeneration,
|
|
observationSource,
|
|
observedMessage,
|
|
);
|
|
guard.assertValid();
|
|
return result;
|
|
},
|
|
guardedObserveMessages: async (sourceContext, observations, guard) => {
|
|
const results = [];
|
|
for (const observation of observations) {
|
|
guard.assertValid();
|
|
if (!observation.normalization.ok) {
|
|
await repository.guardedRecordAnomaly(observation.normalization.anomaly, guard);
|
|
results.push({
|
|
status: "anomaly" as const,
|
|
anomalyCode: observation.normalization.anomaly.anomalyCode,
|
|
});
|
|
continue;
|
|
}
|
|
results.push(
|
|
await repository.guardedInsertMessage(
|
|
sourceContext,
|
|
observation.historyGeneration,
|
|
observation.observationSource,
|
|
observation.normalization.message,
|
|
guard,
|
|
),
|
|
);
|
|
}
|
|
return results;
|
|
},
|
|
recordAnomaly: async (input) => {
|
|
const existing = anomalies.get(input.fingerprint);
|
|
anomalies.set(input.fingerprint, {
|
|
count: (existing?.count ?? 0) + 1,
|
|
input,
|
|
});
|
|
},
|
|
guardedRecordAnomaly: async (input, guard) => {
|
|
guard.assertValid();
|
|
await repository.recordAnomaly(input);
|
|
guard.assertValid();
|
|
},
|
|
updateSyncState: async (sourceContext, update, conversationId) => {
|
|
syncUpdates.push({ ...update });
|
|
const key = `${sourceContext.channelAccountId}:${conversationId}`;
|
|
const existing = conversations.get(key);
|
|
if (!existing) return null;
|
|
const updated: OneTalkConversationState = {
|
|
...existing,
|
|
syncPhase: update.mode === "full" ? "initial" : "incremental",
|
|
syncResult: update.result,
|
|
historyComplete: update.historyComplete,
|
|
...(update.advanceAnchor ? { latestMessageId: update.latestMessageId } : {}),
|
|
};
|
|
conversations.set(key, updated);
|
|
return updated;
|
|
},
|
|
guardedUpdateSyncState: async (sourceContext, update, conversationId, guard) => {
|
|
guard.assertValid();
|
|
const result = await repository.updateSyncState(sourceContext, update, conversationId);
|
|
guard.assertValid();
|
|
return result;
|
|
},
|
|
};
|
|
|
|
return { repository, conversations, messages, anomalies, syncUpdates };
|
|
};
|
|
|
|
test("persists direct discovery and upgrades an explicit legacy rediscovery", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
const legacyConversation: OneTalkConversationState = {
|
|
channelAccountId: context.channelAccountId,
|
|
conversationId: "legacy-conversation",
|
|
conversationKind: null,
|
|
syncPhase: "initial",
|
|
syncResult: "incomplete",
|
|
latestMessageId: null,
|
|
historyComplete: false,
|
|
messageCount: 0,
|
|
historyGeneration: "initial",
|
|
};
|
|
harness.conversations.set(
|
|
`${context.channelAccountId}:legacy-conversation`,
|
|
legacyConversation,
|
|
);
|
|
|
|
assert.equal(
|
|
(await service.readConversation({ ...context }, "legacy-conversation"))?.conversationKind,
|
|
null,
|
|
);
|
|
const discovered = await service.discoverConversation(
|
|
context,
|
|
"legacy-conversation",
|
|
undefined,
|
|
"direct",
|
|
);
|
|
const repeated = await service.discoverConversation(
|
|
context,
|
|
"legacy-conversation",
|
|
undefined,
|
|
"direct",
|
|
);
|
|
|
|
assert.equal(discovered.conversationKind, "direct");
|
|
assert.equal(repeated.conversationKind, "direct");
|
|
assert.equal(
|
|
harness.conversations.get(`${context.channelAccountId}:legacy-conversation`)
|
|
?.conversationKind,
|
|
"direct",
|
|
);
|
|
});
|
|
|
|
test("requires a discovered conversation and keeps message facts idempotent", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
|
|
const unknownConversation = await service.observeMessage(context, "live", message());
|
|
assert.deepEqual(unknownConversation, {
|
|
status: "rejected",
|
|
reason: "conversation_not_discovered",
|
|
});
|
|
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
const first = await service.observeMessage(context, "history", message());
|
|
const duplicate = await service.observeMessage(context, "live", message());
|
|
|
|
assert.equal(first.status, "accepted");
|
|
assert.equal(duplicate.status, "duplicate");
|
|
assert.equal(harness.messages.size, 1);
|
|
assert.equal(harness.conversations.get("account-1:conversation-1")?.messageCount, 1);
|
|
});
|
|
|
|
test("records malformed observations as merged metadata-only anomalies", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
|
|
const malformed = {
|
|
...message(),
|
|
messageId: undefined,
|
|
participantIds: undefined,
|
|
content: { version: 1, kind: "text", text: "diagnostic text", token: "session-secret" },
|
|
} as unknown as OneTalkObservedMessage;
|
|
const first = await service.observeMessage(context, "history", malformed);
|
|
const second = await service.observeMessage(context, "history", malformed);
|
|
|
|
assert.equal(first.status, "anomaly");
|
|
assert.equal(second.status, "anomaly");
|
|
assert.equal(harness.messages.size, 0);
|
|
assert.equal(harness.anomalies.size, 1);
|
|
const anomaly = [...harness.anomalies.values()][0];
|
|
assert.equal(anomaly.count, 2);
|
|
assert.deepEqual(anomaly.input.missingFields, ["content"]);
|
|
assert.deepEqual(anomaly.input.payload, { fields: ["content"] });
|
|
});
|
|
|
|
test("persists a valid normalized content object without server-side reinterpretation", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
|
|
const result = await service.observeMessage(
|
|
context,
|
|
"send_confirmation",
|
|
message({
|
|
content: { version: 1, kind: "text", text: "hello" },
|
|
}),
|
|
);
|
|
|
|
assert.equal(result.status, "accepted");
|
|
if (result.status !== "accepted") return;
|
|
assert.deepEqual(result.message.content, { version: 1, kind: "text", text: "hello" });
|
|
});
|
|
|
|
test("persists an approved structured order without server-side card parsing", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
|
|
const content = {
|
|
version: 1 as const,
|
|
kind: "order" as const,
|
|
orderId: "order-1",
|
|
bizCode: 42,
|
|
contractId: "contract-1",
|
|
id: "id-1",
|
|
tenant: "tenant-1",
|
|
orderAmount: 12.5,
|
|
orderAmountCurrency: "USD",
|
|
paymentAmount: 10,
|
|
paymentAmountCurrency: "USD",
|
|
statusMessageKey: "order.pending_payment",
|
|
actions: [{ name: "pay", messageKey: "order.pay", payStep: "deposit" }],
|
|
};
|
|
const result = await service.observeMessage(context, "history", message({ content }));
|
|
|
|
assert.equal(result.status, "accepted");
|
|
if (result.status !== "accepted") return;
|
|
assert.deepEqual(result.message.content, content);
|
|
assert.equal(JSON.stringify(result).includes("sign"), false);
|
|
});
|
|
|
|
test("returns an existing rendered-card supplement for a batched live observation", async () => {
|
|
const observedMessage = message({
|
|
content: {
|
|
version: 1,
|
|
kind: "order",
|
|
orderId: "order-1",
|
|
bizCode: null,
|
|
contractId: null,
|
|
id: null,
|
|
tenant: null,
|
|
orderAmount: 1,
|
|
orderAmountCurrency: "USD",
|
|
paymentAmount: 1,
|
|
paymentAmountCurrency: "USD",
|
|
statusMessageKey: "paid",
|
|
actions: [],
|
|
},
|
|
});
|
|
const renderedCardContent = {
|
|
version: 1 as const,
|
|
kind: "rendered_order" as const,
|
|
title: "Late order",
|
|
products: [],
|
|
productCount: 0,
|
|
status: { code: null, text: "Paid" },
|
|
payment: { totalDisplay: "$1", discountDisplay: null },
|
|
delivery: { shippingAddress: "Hangzhou", methodLabel: null, dateLabel: null },
|
|
action: { label: null, status: null },
|
|
};
|
|
const baseRow = {
|
|
channelAccountId: context.channelAccountId,
|
|
conversationId: "conversation-1",
|
|
messageId: "message-1",
|
|
senderId: "sender-1",
|
|
binding: context.binding,
|
|
mindUserId: context.mindUserId,
|
|
workspaceId: context.workspaceId,
|
|
deviceId: context.deviceId,
|
|
direction: "received" as const,
|
|
observationType: "new" as const,
|
|
sentAtMs: 1_700_000_000_000,
|
|
content: observedMessage.content,
|
|
participantIds: ["sender-1", "login-user-1"],
|
|
readStatus: 1,
|
|
messageStatus: 2,
|
|
unreadCount: 0,
|
|
firstObservedAt: new Date("2026-09-15T00:00:00.000Z"),
|
|
lastObservedAt: new Date("2026-09-15T00:00:00.000Z"),
|
|
};
|
|
const transaction = {
|
|
select: () => ({
|
|
from: (table: unknown) => ({
|
|
where: () => ({
|
|
limit: () => {
|
|
const rows =
|
|
table === onetalkConversation
|
|
? [
|
|
{
|
|
conversationId: "conversation-1",
|
|
conversationKind: "direct" as const,
|
|
historyGeneration: "initial",
|
|
},
|
|
]
|
|
: table === onetalkRenderedCardContent
|
|
? [{ content: renderedCardContent }]
|
|
: [];
|
|
const promise = Promise.resolve(rows);
|
|
return Object.assign(promise, { for: () => promise });
|
|
},
|
|
}),
|
|
}),
|
|
}),
|
|
insert: () => ({
|
|
values: () => ({
|
|
onConflictDoNothing: () => ({ returning: async () => [baseRow] }),
|
|
}),
|
|
}),
|
|
update: () => ({ set: () => ({ where: async () => [] }) }),
|
|
};
|
|
const database = {
|
|
transaction: async (run: (value: typeof transaction) => Promise<unknown>) =>
|
|
run(transaction),
|
|
};
|
|
const repository = createOneTalkRepository(database as never);
|
|
|
|
const results = await repository.guardedObserveMessages(
|
|
context,
|
|
[
|
|
{
|
|
historyGeneration: "initial",
|
|
observationSource: "live",
|
|
normalization: { ok: true, message: observedMessage },
|
|
},
|
|
],
|
|
{ assertValid: () => undefined },
|
|
);
|
|
|
|
assert.deepEqual(results, [
|
|
{
|
|
status: "accepted",
|
|
message: observedMessage,
|
|
renderedCardContent,
|
|
},
|
|
]);
|
|
});
|
|
|
|
test("advances only valid shared anchors and preserves incomplete outcomes", async () => {
|
|
const emptyHarness = createRepositoryHarness();
|
|
const emptyService = createOneTalkService(emptyHarness.repository);
|
|
await emptyService.discoverConversation(context, "empty-conversation", undefined, "direct");
|
|
const emptyResult = await emptyService.completeSync(context, {
|
|
conversationId: "empty-conversation",
|
|
historyGeneration: "initial",
|
|
mode: "full",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
latestMessageId: null,
|
|
});
|
|
assert.equal(emptyResult.status, "accepted");
|
|
if (emptyResult.status === "accepted") {
|
|
assert.equal(emptyResult.anchorAdvanced, true);
|
|
assert.equal(emptyResult.conversation.latestMessageId, null);
|
|
assert.equal(emptyResult.conversation.historyComplete, true);
|
|
}
|
|
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
await service.observeMessage(context, "incremental", message());
|
|
|
|
const valid = await service.completeSync(context, {
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "initial",
|
|
mode: "incremental",
|
|
historyComplete: true,
|
|
result: "succeeded_with_anomalies",
|
|
latestMessageId: "message-1",
|
|
latestMessageAtMs: 1_700_000_000_500,
|
|
});
|
|
assert.equal(valid.status, "accepted");
|
|
if (valid.status === "accepted") {
|
|
assert.equal(valid.anchorAdvanced, true);
|
|
assert.equal(valid.conversation.latestMessageId, "message-1");
|
|
assert.equal(valid.conversation.syncResult, "succeeded_with_anomalies");
|
|
}
|
|
assert.equal(harness.syncUpdates.at(-1)?.latestMessageAtMs, 1_700_000_000_500);
|
|
|
|
const missing = await service.completeSync(context, {
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "initial",
|
|
mode: "incremental",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
});
|
|
assert.equal(missing.status, "accepted");
|
|
if (missing.status === "accepted") {
|
|
assert.equal(missing.anchorAdvanced, false);
|
|
assert.equal(missing.conversation.latestMessageId, "message-1");
|
|
assert.equal(missing.conversation.syncResult, "incomplete");
|
|
assert.equal(missing.conversation.historyComplete, false);
|
|
}
|
|
|
|
const anchorNotFound = await service.completeSync(context, {
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "initial",
|
|
mode: "incremental",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
latestMessageId: "message-from-another-conversation",
|
|
anomalyCode: "incremental_anchor_not_found",
|
|
});
|
|
assert.equal(anchorNotFound.status, "accepted");
|
|
if (anchorNotFound.status === "accepted") {
|
|
assert.equal(anchorNotFound.anchorAdvanced, false);
|
|
assert.equal(anchorNotFound.conversation.latestMessageId, "message-1");
|
|
assert.equal(anchorNotFound.conversation.syncResult, "incomplete");
|
|
}
|
|
|
|
const failed = await service.completeSync(context, {
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "initial",
|
|
mode: "incremental",
|
|
historyComplete: true,
|
|
result: "failed",
|
|
latestMessageId: "message-1",
|
|
});
|
|
assert.equal(failed.status, "accepted");
|
|
if (failed.status === "accepted") {
|
|
assert.equal(failed.anchorAdvanced, false);
|
|
assert.equal(failed.conversation.latestMessageId, "message-1");
|
|
assert.equal(failed.conversation.syncResult, "failed");
|
|
assert.equal(failed.conversation.historyComplete, false);
|
|
}
|
|
assert.ok(
|
|
[...harness.anomalies.values()].some(
|
|
({ input }) => input.anomalyCode === "latest_message_id_missing",
|
|
),
|
|
);
|
|
assert.ok(
|
|
[...harness.anomalies.values()].some(
|
|
({ input }) => input.anomalyCode === "incremental_anchor_not_found",
|
|
),
|
|
);
|
|
});
|
|
|
|
test("does not accept a sync completion for an unknown conversation", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
|
|
const result = await service.completeSync(context, {
|
|
conversationId: "unknown-conversation",
|
|
historyGeneration: "initial",
|
|
mode: "full",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
latestMessageId: null,
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
status: "rejected",
|
|
reason: "conversation_not_discovered",
|
|
});
|
|
});
|
|
|
|
test("rejects stale history generation for observations and sync completion", async () => {
|
|
const harness = createRepositoryHarness();
|
|
const service = createOneTalkService(harness.repository);
|
|
await service.discoverConversation(context, "conversation-1", undefined, "direct");
|
|
|
|
const observation = await harness.repository.insertMessage(
|
|
context,
|
|
"stale-generation",
|
|
"history",
|
|
message(),
|
|
);
|
|
assert.deepEqual(observation, {
|
|
status: "rejected",
|
|
reason: "history_generation_mismatch",
|
|
});
|
|
|
|
const completion = await service.completeSync(context, {
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "stale-generation",
|
|
mode: "full",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
latestMessageId: null,
|
|
});
|
|
assert.deepEqual(completion, {
|
|
status: "rejected",
|
|
reason: "history_generation_mismatch",
|
|
});
|
|
assert.equal(harness.syncUpdates.length, 0);
|
|
});
|
|
|
|
test("does not leave a sync anomaly when reset wins after the initial generation read", async () => {
|
|
const harness = createRepositoryHarness();
|
|
await harness.repository.discoverConversation(context, "conversation-1", "direct");
|
|
const baseFindConversation = harness.repository.findConversation;
|
|
const baseGuardedRecordAnomaly = harness.repository.guardedRecordAnomaly;
|
|
const baseGuardedUpdateSyncState = harness.repository.guardedUpdateSyncState;
|
|
const repository: OneTalkRepository = {
|
|
...harness.repository,
|
|
findConversation: async (...args) => baseFindConversation(...args),
|
|
guardedRecordAnomaly: async (input, guard) => {
|
|
const key = `${context.channelAccountId}:${input.conversationId}`;
|
|
const current = harness.conversations.get(key);
|
|
if (current) {
|
|
harness.conversations.set(key, {
|
|
...current,
|
|
historyGeneration: "reset-generation",
|
|
});
|
|
}
|
|
await baseGuardedRecordAnomaly(input, guard);
|
|
},
|
|
guardedUpdateSyncState: async (sourceContext, update, conversationId, guard) => {
|
|
const current = harness.conversations.get(
|
|
`${sourceContext.channelAccountId}:${conversationId}`,
|
|
);
|
|
if (current?.historyGeneration !== update.historyGeneration) return null;
|
|
return baseGuardedUpdateSyncState(sourceContext, update, conversationId, guard);
|
|
},
|
|
};
|
|
const service = createOneTalkService(repository);
|
|
|
|
const result = await service.completeSync(
|
|
context,
|
|
{
|
|
conversationId: "conversation-1",
|
|
historyGeneration: "initial",
|
|
mode: "full",
|
|
historyComplete: true,
|
|
result: "succeeded",
|
|
latestMessageId: null,
|
|
anomalyCode: "incremental_anchor_not_found",
|
|
},
|
|
{ assertValid: () => undefined },
|
|
);
|
|
|
|
assert.deepEqual(result, {
|
|
status: "rejected",
|
|
reason: "conversation_not_discovered",
|
|
});
|
|
assert.equal(harness.anomalies.size, 0);
|
|
});
|