mirror of
https://github.com/sinanyuntu/trade-message-center.git
synced 2026-09-17 13:22:11 +08:00
chore: remove PostgreSQL integration release gate
This commit is contained in:
YBF
@@ -157,63 +157,9 @@ jobs:
|
||||
if-no-files-found: error
|
||||
retention-days: 7
|
||||
|
||||
postgres-integration:
|
||||
name: release-database-gate
|
||||
needs: verify-main-tag
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
env:
|
||||
CI: "true"
|
||||
NODE_ENV: test
|
||||
MIND_AUTH_BASE_URL: http://127.0.0.1:8787
|
||||
MIND_PAGE_ORIGIN: http://127.0.0.1:7878
|
||||
ONETALK_PLUGIN_ORIGINS: chrome-extension://ogdbffjakeeidblabkeakakdecfbcmlf
|
||||
MIND_AUTH_TIMEOUT_MS: "3000"
|
||||
DATABASE_URL: postgres://postgres:postgres@127.0.0.1:5432/trade_message_center_test
|
||||
TEST_DATABASE_URL: postgres://postgres:postgres@127.0.0.1:5432/trade_message_center_test
|
||||
services:
|
||||
postgres:
|
||||
image: postgres:17-alpine
|
||||
env:
|
||||
POSTGRES_DB: trade_message_center_test
|
||||
POSTGRES_USER: postgres
|
||||
POSTGRES_PASSWORD: postgres
|
||||
ports:
|
||||
- 5432:5432
|
||||
options: >-
|
||||
--health-cmd "pg_isready -U postgres -d trade_message_center_test"
|
||||
--health-interval 5s
|
||||
--health-timeout 5s
|
||||
--health-retries 12
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up pnpm
|
||||
uses: pnpm/action-setup@v4
|
||||
with:
|
||||
version: 11.7.0
|
||||
run_install: false
|
||||
|
||||
- name: Set up Node.js
|
||||
uses: actions/setup-node@v4
|
||||
with:
|
||||
node-version-file: .nvmrc
|
||||
cache: pnpm
|
||||
cache-dependency-path: pnpm-lock.yaml
|
||||
|
||||
- name: Install dependencies
|
||||
run: pnpm install --frozen-lockfile --registry=https://registry.npmjs.org
|
||||
|
||||
- name: Apply database migrations
|
||||
run: pnpm --filter @trade-message-center/server db:migrate
|
||||
|
||||
- name: Run PostgreSQL integration tests
|
||||
run: timeout 60s pnpm --filter @trade-message-center/server test:integration
|
||||
|
||||
deploy:
|
||||
name: deploy-server
|
||||
needs: [quality, postgres-integration, publish-extension]
|
||||
needs: [quality, publish-extension]
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
@@ -517,7 +463,7 @@ jobs:
|
||||
|
||||
publish-extension:
|
||||
name: publish-extension
|
||||
needs: [quality, postgres-integration]
|
||||
needs: quality
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 15
|
||||
|
||||
@@ -310,7 +310,7 @@ OSS Bucket、Endpoint 和 Region 已直接写在 `.github/workflows/release_ci.y
|
||||
|
||||
PR 和 `main` 分支推送本身不触发发布。发布新版本时先更新根版本并同步子包，合入 `main` 后再推送对应 tag；不要只更换 tag 名而重复使用旧的根版本。CI 自动构建并打包，不需要提交本地 `dist/` 或手动上传 ZIP。
|
||||
|
||||
发布顺序为质量检查与 PostgreSQL 集成测试 → 上传扩展 ZIP 到 OSS → 部署 server。上传失败时不会部署新服务，避免下载接口指向尚不存在的新版本包。扩展使用 `production` Environment 的 `VITE_BRIGHT_WEBSOCKET_URL` 构建；本地 `.env` 中的开发地址不能用于线上下载包。上线后应实际访问下载接口并下载 ZIP，核对 `manifest.json` 版本及扩展连接地址，不能只以 `/health` 成功判断下载功能可用。
|
||||
发布顺序为质量检查 → 上传扩展 ZIP 到 OSS → 部署 server。上传失败时不会部署新服务，避免下载接口指向尚不存在的新版本包。扩展使用 `production` Environment 的 `VITE_BRIGHT_WEBSOCKET_URL` 构建；本地 `.env` 中的开发地址不能用于线上下载包。上线后应实际访问下载接口并下载 ZIP，核对 `manifest.json` 版本及扩展连接地址，不能只以 `/health` 成功判断下载功能可用。
|
||||
|
||||
## Server 数据库迁移
|
||||
|
||||
@@ -336,16 +336,6 @@ pnpm build
|
||||
git diff --check
|
||||
```
|
||||
|
||||
需要执行真实 PostgreSQL 集成测试时，将 `TEST_DATABASE_URL` 放在根目录 `.env.local` 中（或显式导出），运行专用测试命令：
|
||||
|
||||
```bash
|
||||
pnpm --filter @trade-message-center/server test:integration
|
||||
```
|
||||
|
||||
该命令会按 server 迁移命令相同的顺序加载根目录 `.env`、`.env.local` 和 `.env.development`。普通 `pnpm test` 会排除 `*.integration.test.*`，不会自动连接 PostgreSQL；未设置 `TEST_DATABASE_URL` 时，单独运行集成测试会明确标记为 skip，不代表真实数据库迁移和事务边界已经验证。
|
||||
|
||||
`test:integration` 串行执行全部 `test/*.integration.test.ts`，覆盖写入幂等、联系人资料和会话读取；串行执行避免测试文件在同一个数据库上并发初始化 migration。发布流水线为该命令设置 60 秒硬超时。
|
||||
|
||||
## VSCode 开发环境
|
||||
|
||||
本项目统一使用 Oxfmt 格式化代码，缩进为 4 个空格。请在 VSCode 中安装 Oxc 插件 `oxc.oxc-vscode`，否则保存文件时可能使用其它 formatter，导致代码在提交钩子中再次变化。
|
||||
|
||||
@@ -65,7 +65,7 @@ test("names the quality archive from the validated root version", async () => {
|
||||
test("publishes the SHA artifact under the quality job version path", async () => {
|
||||
const publish = blockFromHeading(await readWorkflow(), " publish-extension:");
|
||||
|
||||
assert.match(publish, /needs: \[quality, postgres-integration\]/u);
|
||||
assert.match(publish, /needs: quality/u);
|
||||
assert.match(
|
||||
publish,
|
||||
/PACKAGE_VERSION:\s+\$\{\{ needs\.quality\.outputs\.extension_version \}\}/u,
|
||||
@@ -116,28 +116,3 @@ test("waits for the extension archive to be published before deploying its downl
|
||||
"Server deployment must wait for publish-extension so the advertised ZIP exists",
|
||||
);
|
||||
});
|
||||
|
||||
test("runs every PostgreSQL integration suite serially in the release database gate", async () => {
|
||||
const databaseGate = blockFromHeading(await readWorkflow(), " postgres-integration:");
|
||||
assert.match(
|
||||
databaseGate,
|
||||
/^\s+run: timeout 60s pnpm --filter @trade-message-center\/server (?:run )?test:integration\s*$/mu,
|
||||
"The database gate must invoke the integration script with a 60-second hard timeout",
|
||||
);
|
||||
|
||||
const serverPackage = JSON.parse(
|
||||
await readFile(new URL("../../server/package.json", import.meta.url), "utf8"),
|
||||
);
|
||||
const command = serverPackage.scripts["test:integration"];
|
||||
assert.equal(typeof command, "string");
|
||||
const arguments_ = command.split(/\s+/u);
|
||||
assert.ok(arguments_.includes("--test"), "Integration suites must use the Node test runner");
|
||||
assert.ok(
|
||||
arguments_.includes("test/*.integration.test.ts"),
|
||||
"The integration script must include both write and read PostgreSQL suites",
|
||||
);
|
||||
assert.ok(
|
||||
arguments_.includes("--test-concurrency=1"),
|
||||
"Integration suites share migrations and must not race against the same database",
|
||||
);
|
||||
});
|
||||
|
||||
@@ -20,9 +20,7 @@
|
||||
"typecheck": "tsc --noEmit -p tsconfig.json",
|
||||
"pretypecheck": "pnpm --filter @trade-message-center/onetalk-contract build",
|
||||
"test": "node --experimental-strip-types --test $(find test -type f -name '*.test.ts' -not -name '*.integration.test.ts' -print) && pnpm run build && node --test $(find dist/test -type f -name '*.test.js' -not -name '*.integration.test.js' -print)",
|
||||
"pretest": "pnpm --filter @trade-message-center/onetalk-contract build",
|
||||
"test:integration": "node --env-file-if-exists=../../.env --env-file-if-exists=../../.env.local --env-file-if-exists=../../.env.development --env-file-if-exists=../../.env.development.local --experimental-strip-types --test --test-concurrency=1 test/*.integration.test.ts",
|
||||
"pretest:integration": "pnpm run pretest"
|
||||
"pretest": "pnpm --filter @trade-message-center/onetalk-contract build"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fastify/websocket": "^11.3.0",
|
||||
|
||||
@@ -1,417 +0,0 @@
|
||||
// 验证买家事实 PostgreSQL 的复合键和来源替换审计
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import test from "node:test";
|
||||
|
||||
import postgres from "postgres";
|
||||
import {
|
||||
createOneTalkBuyerFactFingerprint,
|
||||
type OneTalkBuyerFact,
|
||||
} from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import { createDatabase, type DatabaseConnection } from "../src/database/index.ts";
|
||||
import { runMigrations } from "../src/database/migrate.ts";
|
||||
import {
|
||||
createOneTalkBuyerFactRepository,
|
||||
createOneTalkReadRepository,
|
||||
createOneTalkReadService,
|
||||
} from "../src/onetalk/index.ts";
|
||||
|
||||
const databaseUrl = process.env.TEST_DATABASE_URL?.trim();
|
||||
|
||||
const confirmedFact = (
|
||||
conversationId: string,
|
||||
buyerTags: string[],
|
||||
buyerFeatures: string[],
|
||||
observedAtMs: number,
|
||||
): OneTalkBuyerFact => ({
|
||||
conversationId,
|
||||
buyerTags,
|
||||
buyerFeatures,
|
||||
tags: {
|
||||
state: "confirmed",
|
||||
errorCode: null,
|
||||
attemptedAtMs: observedAtMs,
|
||||
confirmedAtMs: observedAtMs,
|
||||
},
|
||||
features: {
|
||||
state: "confirmed",
|
||||
errorCode: null,
|
||||
attemptedAtMs: observedAtMs,
|
||||
confirmedAtMs: observedAtMs,
|
||||
},
|
||||
observedAtMs,
|
||||
factFingerprint: createOneTalkBuyerFactFingerprint(buyerTags, buyerFeatures),
|
||||
});
|
||||
|
||||
const tagsFailureWithConfirmedFeatures = (
|
||||
conversationId: string,
|
||||
buyerFeatures: string[],
|
||||
attemptedAtMs: number,
|
||||
): OneTalkBuyerFact => ({
|
||||
conversationId,
|
||||
buyerTags: null,
|
||||
buyerFeatures,
|
||||
tags: {
|
||||
state: "failed",
|
||||
errorCode: "context_changed",
|
||||
attemptedAtMs,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
features: {
|
||||
state: "confirmed",
|
||||
errorCode: null,
|
||||
attemptedAtMs,
|
||||
confirmedAtMs: attemptedAtMs,
|
||||
},
|
||||
observedAtMs: attemptedAtMs,
|
||||
factFingerprint: createOneTalkBuyerFactFingerprint(null, buyerFeatures),
|
||||
});
|
||||
|
||||
const unconfirmedFact = (conversationId: string, attemptedAtMs: number): OneTalkBuyerFact => ({
|
||||
conversationId,
|
||||
buyerTags: null,
|
||||
buyerFeatures: null,
|
||||
tags: {
|
||||
state: "failed",
|
||||
errorCode: "crm_contract_unverified",
|
||||
attemptedAtMs,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
features: {
|
||||
state: "pending",
|
||||
errorCode: null,
|
||||
attemptedAtMs,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
observedAtMs: attemptedAtMs,
|
||||
factFingerprint: createOneTalkBuyerFactFingerprint(null, null),
|
||||
});
|
||||
|
||||
test(
|
||||
"stores buyer sources by account/conversation and preserves confirmed tags across later failures",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL buyer fact integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const client = postgres(databaseUrl, { max: 1 });
|
||||
const channelAccountId = `buyer-integration-${randomUUID()}`;
|
||||
const otherChannelAccountId = `buyer-integration-other-${randomUUID()}`;
|
||||
const conversationId = "shared-conversation";
|
||||
const repository = createOneTalkBuyerFactRepository(database.db);
|
||||
const initialAtMs = 1_700_000_000_000;
|
||||
const failureAtMs = initialAtMs + 1;
|
||||
const duplicateAttemptAtMs = failureAtMs + 1;
|
||||
const firstReceivedAt = new Date("2026-09-09T00:00:00.000Z");
|
||||
const secondReceivedAt = new Date("2026-09-09T00:01:00.000Z");
|
||||
|
||||
try {
|
||||
await repository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [confirmedFact(conversationId, ["will-loss"], ["priority"], initialAtMs)],
|
||||
receivedAt: firstReceivedAt,
|
||||
});
|
||||
await repository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [tagsFailureWithConfirmedFeatures(conversationId, [], failureAtMs)],
|
||||
receivedAt: secondReceivedAt,
|
||||
});
|
||||
await repository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [tagsFailureWithConfirmedFeatures(conversationId, [], duplicateAttemptAtMs)],
|
||||
receivedAt: secondReceivedAt,
|
||||
});
|
||||
await repository.storeFacts({
|
||||
channelAccountId: otherChannelAccountId,
|
||||
facts: [
|
||||
confirmedFact(
|
||||
conversationId,
|
||||
["other-account"],
|
||||
["other-feature"],
|
||||
initialAtMs,
|
||||
),
|
||||
],
|
||||
receivedAt: firstReceivedAt,
|
||||
});
|
||||
|
||||
const rows = await client<
|
||||
{
|
||||
channel_account_id: string;
|
||||
buyer_tags: string[] | null;
|
||||
buyer_features: string[] | null;
|
||||
tags_state: string;
|
||||
features_state: string;
|
||||
tags_last_attempted_at: Date;
|
||||
tags_confirmed_at: Date | null;
|
||||
tags_error_code: string | null;
|
||||
fact_fingerprint: string;
|
||||
received_at: Date;
|
||||
}[]
|
||||
>`
|
||||
select
|
||||
channel_account_id,
|
||||
buyer_tags,
|
||||
buyer_features,
|
||||
tags_state,
|
||||
features_state,
|
||||
tags_last_attempted_at,
|
||||
tags_confirmed_at,
|
||||
tags_error_code,
|
||||
fact_fingerprint,
|
||||
received_at
|
||||
from onetalk_buyer_fact
|
||||
where conversation_id = ${conversationId}
|
||||
and channel_account_id in (${channelAccountId}, ${otherChannelAccountId})
|
||||
order by channel_account_id
|
||||
`;
|
||||
const persisted = rows.find((row) => row.channel_account_id === channelAccountId);
|
||||
const isolated = rows.find((row) => row.channel_account_id === otherChannelAccountId);
|
||||
|
||||
assert.equal(rows.length, 2);
|
||||
assert.deepEqual(persisted?.buyer_tags, ["will-loss"]);
|
||||
assert.deepEqual(persisted?.buyer_features, []);
|
||||
assert.equal(persisted?.tags_state, "failed");
|
||||
assert.equal(persisted?.features_state, "confirmed");
|
||||
assert.equal(persisted?.tags_error_code, "context_changed");
|
||||
assert.equal(persisted?.tags_last_attempted_at.getTime(), duplicateAttemptAtMs);
|
||||
assert.equal(persisted?.tags_confirmed_at?.getTime(), initialAtMs);
|
||||
assert.equal(
|
||||
persisted?.fact_fingerprint,
|
||||
createOneTalkBuyerFactFingerprint(["will-loss"], []),
|
||||
);
|
||||
assert.equal(persisted?.received_at.getTime(), secondReceivedAt.getTime());
|
||||
assert.deepEqual(isolated?.buyer_tags, ["other-account"]);
|
||||
assert.deepEqual(isolated?.buyer_features, ["other-feature"]);
|
||||
} finally {
|
||||
await client`
|
||||
delete from onetalk_buyer_fact
|
||||
where channel_account_id in (${channelAccountId}, ${otherChannelAccountId})
|
||||
`;
|
||||
await client.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"projects buyer facts only for the same account/conversation key without internal source state",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL buyer read integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const client = postgres(databaseUrl, { max: 1 });
|
||||
const channelAccountId = `buyer-read-${randomUUID()}`;
|
||||
const otherChannelAccountId = `buyer-read-other-${randomUUID()}`;
|
||||
const conversationId = "same-conversation";
|
||||
const unconfirmedConversationId = "unconfirmed-conversation";
|
||||
const asOf = new Date("2026-09-09T00:00:00.000Z");
|
||||
const buyerRepository = createOneTalkBuyerFactRepository(database.db);
|
||||
const service = createOneTalkReadService(createOneTalkReadRepository(database.db), {
|
||||
now: () => asOf,
|
||||
});
|
||||
|
||||
try {
|
||||
for (const accountId of [channelAccountId, otherChannelAccountId]) {
|
||||
await client`
|
||||
insert into onetalk_conversation (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
conversation_kind,
|
||||
first_discovered_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${accountId},
|
||||
${conversationId},
|
||||
'direct',
|
||||
${asOf},
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
}
|
||||
await client`
|
||||
insert into onetalk_conversation (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
conversation_kind,
|
||||
first_discovered_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
${unconfirmedConversationId},
|
||||
'direct',
|
||||
${asOf},
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
await buyerRepository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [
|
||||
{
|
||||
...confirmedFact(conversationId, ["same-account"], [], 1_700_000_000_000),
|
||||
email: "buyer@example.test",
|
||||
registrationDate: "2025-03-12",
|
||||
companyWebsite: "https://example.test/",
|
||||
contactDetails: {
|
||||
state: "confirmed",
|
||||
errorCode: null,
|
||||
attemptedAtMs: 1_700_000_000_000,
|
||||
confirmedAtMs: 1_700_000_000_000,
|
||||
},
|
||||
factFingerprint: createOneTalkBuyerFactFingerprint(["same-account"], [], {
|
||||
email: "buyer@example.test",
|
||||
registrationDate: "2025-03-12",
|
||||
companyWebsite: "https://example.test/",
|
||||
}),
|
||||
},
|
||||
],
|
||||
receivedAt: asOf,
|
||||
});
|
||||
await buyerRepository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [
|
||||
{
|
||||
...confirmedFact(conversationId, ["same-account"], [], 1_700_000_000_001),
|
||||
email: null,
|
||||
registrationDate: null,
|
||||
companyWebsite: null,
|
||||
contactDetails: {
|
||||
email: {
|
||||
state: "pending",
|
||||
errorCode: null,
|
||||
attemptedAtMs: 1_700_000_000_001,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
registrationDate: {
|
||||
state: "pending",
|
||||
errorCode: null,
|
||||
attemptedAtMs: 1_700_000_000_001,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
companyWebsite: {
|
||||
state: "pending",
|
||||
errorCode: null,
|
||||
attemptedAtMs: 1_700_000_000_001,
|
||||
confirmedAtMs: null,
|
||||
},
|
||||
},
|
||||
factFingerprint: createOneTalkBuyerFactFingerprint(["same-account"], [], {
|
||||
email: null,
|
||||
registrationDate: null,
|
||||
companyWebsite: null,
|
||||
}),
|
||||
},
|
||||
],
|
||||
receivedAt: asOf,
|
||||
});
|
||||
await buyerRepository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [
|
||||
confirmedFact(
|
||||
conversationId,
|
||||
["same-account-after-legacy"],
|
||||
[],
|
||||
1_700_000_000_002,
|
||||
),
|
||||
],
|
||||
receivedAt: asOf,
|
||||
});
|
||||
await buyerRepository.storeFacts({
|
||||
channelAccountId: otherChannelAccountId,
|
||||
facts: [
|
||||
confirmedFact(
|
||||
conversationId,
|
||||
["other-account"],
|
||||
["other-feature"],
|
||||
1_700_000_000_000,
|
||||
),
|
||||
],
|
||||
receivedAt: asOf,
|
||||
});
|
||||
await buyerRepository.storeFacts({
|
||||
channelAccountId,
|
||||
facts: [unconfirmedFact(unconfirmedConversationId, 1_700_000_000_000)],
|
||||
receivedAt: asOf,
|
||||
});
|
||||
|
||||
const result = await service.readConversation({
|
||||
scope: {
|
||||
mindUserId: "mind-user-1",
|
||||
workspaceId: "workspace-1",
|
||||
channelAccountId,
|
||||
},
|
||||
conversationId,
|
||||
});
|
||||
assert.equal(result.status, "accepted");
|
||||
if (result.status !== "accepted") return;
|
||||
|
||||
assert.deepEqual(Object.keys(result.conversation).sort(), [
|
||||
"avatarUrl",
|
||||
"conversationId",
|
||||
"conversationType",
|
||||
"customerProfile",
|
||||
"lastContactTimeLong",
|
||||
"messagePreview",
|
||||
"name",
|
||||
]);
|
||||
assert.deepEqual(result.conversation.customerProfile, {
|
||||
name: null,
|
||||
avatarUrl: null,
|
||||
buyerTags: ["same-account-after-legacy"],
|
||||
buyerFeatures: [],
|
||||
email: "buyer@example.test",
|
||||
registrationDate: "2025-03-12",
|
||||
companyWebsite: "https://example.test/",
|
||||
countryCode: null,
|
||||
companyName: null,
|
||||
});
|
||||
|
||||
const unconfirmed = await service.readConversation({
|
||||
scope: {
|
||||
mindUserId: "mind-user-1",
|
||||
workspaceId: "workspace-1",
|
||||
channelAccountId,
|
||||
},
|
||||
conversationId: unconfirmedConversationId,
|
||||
});
|
||||
assert.equal(unconfirmed.status, "accepted");
|
||||
if (unconfirmed.status !== "accepted") return;
|
||||
assert.deepEqual(unconfirmed.conversation.customerProfile, {
|
||||
name: null,
|
||||
avatarUrl: null,
|
||||
buyerTags: null,
|
||||
buyerFeatures: null,
|
||||
email: null,
|
||||
registrationDate: null,
|
||||
companyWebsite: null,
|
||||
countryCode: null,
|
||||
companyName: null,
|
||||
});
|
||||
} finally {
|
||||
await client`
|
||||
delete from onetalk_buyer_fact
|
||||
where channel_account_id in (${channelAccountId}, ${otherChannelAccountId})
|
||||
`;
|
||||
await client`
|
||||
delete from onetalk_conversation
|
||||
where channel_account_id in (${channelAccountId}, ${otherChannelAccountId})
|
||||
`;
|
||||
await client.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
@@ -1,601 +0,0 @@
|
||||
// 验证 OneTalk PostgreSQL 迁移与事务
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import test from "node:test";
|
||||
|
||||
import postgres from "postgres";
|
||||
import type {
|
||||
OneTalkContactProfile,
|
||||
OneTalkObservedMessage,
|
||||
} from "@trade-message-center/onetalk-contract";
|
||||
|
||||
import { createDatabase, type DatabaseConnection } from "../src/database/index.ts";
|
||||
import { runMigrations } from "../src/database/migrate.ts";
|
||||
import {
|
||||
createOneTalkProfileRepository,
|
||||
createOneTalkProfileService,
|
||||
createOneTalkRepository,
|
||||
createOneTalkService,
|
||||
OneTalkDatabaseError,
|
||||
} from "../src/onetalk/index.ts";
|
||||
|
||||
const databaseUrl = process.env.TEST_DATABASE_URL?.trim();
|
||||
|
||||
const integrationMessage = (
|
||||
messageId: string,
|
||||
sentAtMs = 1_700_000_000_000,
|
||||
): OneTalkObservedMessage => {
|
||||
return {
|
||||
messageId,
|
||||
conversationId: "integration-conversation",
|
||||
senderId: "sender-1",
|
||||
direction: "received",
|
||||
sentAtMs,
|
||||
content: { version: 1, kind: "text", text: "integration" },
|
||||
participantIds: ["sender-1", "login-user-1"],
|
||||
readStatus: 1,
|
||||
messageStatus: 2,
|
||||
unreadCount: 0,
|
||||
};
|
||||
};
|
||||
|
||||
const integrationProfile = (
|
||||
overrides: Partial<OneTalkContactProfile> = {},
|
||||
): OneTalkContactProfile => {
|
||||
return {
|
||||
conversationId: "profile-conversation-1",
|
||||
aliId: "profile-ali-1",
|
||||
accountId: "profile-account-id-1",
|
||||
loginId: "profile-login-1",
|
||||
name: "First profile name",
|
||||
companyName: "First profile company",
|
||||
countryCode: "CN",
|
||||
currentTimeZone: 8,
|
||||
serviceType: "standard",
|
||||
avatarUrl: "https://cdn.example.com/profile-1.jpg",
|
||||
observedAtMs: 1_700_000_000_000,
|
||||
profileFingerprint: "profile-fingerprint-1",
|
||||
observationStatus: "confirmed",
|
||||
...overrides,
|
||||
};
|
||||
};
|
||||
|
||||
test(
|
||||
"uses the real migration and transaction boundary for message idempotency",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const sql = postgres(databaseUrl, { max: 1 });
|
||||
const accountId = `integration-${randomUUID()}`;
|
||||
const service = createOneTalkService(createOneTalkRepository(database.db));
|
||||
const context = {
|
||||
binding: "integration-binding",
|
||||
mindUserId: "integration-mind-user",
|
||||
workspaceId: "integration-workspace",
|
||||
channelAccountId: accountId,
|
||||
deviceId: "integration-device",
|
||||
};
|
||||
|
||||
try {
|
||||
const discovered = await service.discoverConversation(
|
||||
context,
|
||||
"integration-conversation",
|
||||
undefined,
|
||||
"direct",
|
||||
);
|
||||
const repeatedDiscovery = await service.discoverConversation(
|
||||
context,
|
||||
"integration-conversation",
|
||||
undefined,
|
||||
"direct",
|
||||
);
|
||||
assert.equal(discovered.conversationKind, "direct");
|
||||
assert.equal(repeatedDiscovery.conversationKind, "direct");
|
||||
const first = await service.observeMessage(
|
||||
context,
|
||||
"history",
|
||||
integrationMessage("integration-message"),
|
||||
);
|
||||
const duplicate = await service.observeMessage(
|
||||
context,
|
||||
"live",
|
||||
integrationMessage("integration-message"),
|
||||
);
|
||||
const second = await service.observeMessage(
|
||||
context,
|
||||
"live",
|
||||
integrationMessage("integration-message-2", 1_700_000_000_100),
|
||||
);
|
||||
const anomaly = await service.observeMessage(context, "live", {
|
||||
messageId: "integration-invalid-message",
|
||||
} as unknown as OneTalkObservedMessage);
|
||||
|
||||
assert.equal(first.status, "accepted");
|
||||
assert.equal(duplicate.status, "duplicate");
|
||||
assert.equal(second.status, "accepted");
|
||||
assert.equal(anomaly.status, "anomaly");
|
||||
|
||||
const readScope = {
|
||||
mindUserId: context.mindUserId,
|
||||
workspaceId: context.workspaceId,
|
||||
channelAccountId: context.channelAccountId,
|
||||
};
|
||||
const firstPage = await service.readHistory(
|
||||
readScope,
|
||||
"integration-conversation",
|
||||
null,
|
||||
1,
|
||||
);
|
||||
assert.ok(firstPage);
|
||||
assert.deepEqual(
|
||||
firstPage.messages.map(({ messageId }) => messageId),
|
||||
["integration-message"],
|
||||
);
|
||||
assert.equal(firstPage.hasMore, true);
|
||||
|
||||
const completion = await service.completeSync(context, {
|
||||
conversationId: "integration-conversation",
|
||||
mode: "full",
|
||||
historyComplete: true,
|
||||
result: "succeeded",
|
||||
latestMessageId: "integration-message-2",
|
||||
latestMessageAtMs: 1_700_000_000_300,
|
||||
});
|
||||
assert.equal(completion.status, "accepted");
|
||||
const staleCompletion = await service.completeSync(context, {
|
||||
conversationId: "integration-conversation",
|
||||
mode: "incremental",
|
||||
historyComplete: true,
|
||||
result: "succeeded",
|
||||
latestMessageId: "integration-message-2",
|
||||
latestMessageAtMs: 1_700_000_000_200,
|
||||
});
|
||||
assert.equal(staleCompletion.status, "accepted");
|
||||
|
||||
const secondPage = await service.readHistory(
|
||||
readScope,
|
||||
"integration-conversation",
|
||||
firstPage.nextCursor,
|
||||
1,
|
||||
);
|
||||
assert.ok(secondPage);
|
||||
assert.deepEqual(
|
||||
secondPage.messages.map(({ messageId }) => messageId),
|
||||
["integration-message-2"],
|
||||
);
|
||||
assert.equal(secondPage.hasMore, false);
|
||||
|
||||
const lateHistory = await service.observeMessage(
|
||||
context,
|
||||
"history",
|
||||
integrationMessage("integration-message-3", 1_700_000_000_050),
|
||||
);
|
||||
assert.equal(lateHistory.status, "accepted");
|
||||
|
||||
const messageRows = await sql`
|
||||
select channel_account_id, conversation_id, message_id, binding, observation_type
|
||||
from onetalk_message
|
||||
where channel_account_id = ${accountId}
|
||||
order by message_id
|
||||
`;
|
||||
assert.deepEqual(
|
||||
[...messageRows],
|
||||
[
|
||||
{
|
||||
channel_account_id: accountId,
|
||||
conversation_id: "integration-conversation",
|
||||
message_id: "integration-message",
|
||||
binding: "integration-binding",
|
||||
observation_type: "history",
|
||||
},
|
||||
{
|
||||
channel_account_id: accountId,
|
||||
conversation_id: "integration-conversation",
|
||||
message_id: "integration-message-2",
|
||||
binding: "integration-binding",
|
||||
observation_type: "new",
|
||||
},
|
||||
{
|
||||
channel_account_id: accountId,
|
||||
conversation_id: "integration-conversation",
|
||||
message_id: "integration-message-3",
|
||||
binding: "integration-binding",
|
||||
observation_type: "history",
|
||||
},
|
||||
],
|
||||
);
|
||||
|
||||
const conversationRows = await sql`
|
||||
select message_count, conversation_kind, last_message_at_ms
|
||||
from onetalk_conversation
|
||||
where channel_account_id = ${accountId}
|
||||
and conversation_id = 'integration-conversation'
|
||||
`;
|
||||
assert.deepEqual(
|
||||
[...conversationRows],
|
||||
[
|
||||
{
|
||||
message_count: 3,
|
||||
conversation_kind: "direct",
|
||||
last_message_at_ms: "1700000000300",
|
||||
},
|
||||
],
|
||||
);
|
||||
|
||||
const conversationKindCommentRows = await sql`
|
||||
select col_description('onetalk_conversation'::regclass, attnum) as comment
|
||||
from pg_attribute
|
||||
where attrelid = 'onetalk_conversation'::regclass
|
||||
and attname = 'conversation_kind'
|
||||
and not attisdropped
|
||||
`;
|
||||
assert.match(String(conversationKindCommentRows[0]?.comment), /显式 direct 发现/u);
|
||||
|
||||
const anomalyRows = await sql`
|
||||
select binding
|
||||
from onetalk_message_anomaly
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
assert.deepEqual([...anomalyRows], [{ binding: "integration-binding" }]);
|
||||
} finally {
|
||||
await sql`
|
||||
delete from onetalk_message_anomaly
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql`
|
||||
delete from onetalk_message
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql`
|
||||
delete from onetalk_conversation
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"persists profile facts without a conversation FK and only replaces them with a newer snapshot",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const sql = postgres(databaseUrl, { max: 1 });
|
||||
const accountId = `profile-integration-${randomUUID()}`;
|
||||
const receivedAt = new Date("2026-09-03T00:00:00.000Z");
|
||||
const service = createOneTalkProfileService(createOneTalkProfileRepository(database.db), {
|
||||
now: () => receivedAt,
|
||||
});
|
||||
|
||||
try {
|
||||
const first = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [integrationProfile()],
|
||||
});
|
||||
assert.deepEqual(first, {
|
||||
status: "accepted",
|
||||
inputProfileCount: 1,
|
||||
profileCount: 1,
|
||||
writtenProfileCount: 1,
|
||||
staleProfileCount: 0,
|
||||
writtenConversationIds: ["profile-conversation-1"],
|
||||
});
|
||||
|
||||
const newer = integrationProfile({
|
||||
aliId: "profile-ali-newer",
|
||||
accountId: null,
|
||||
loginId: null,
|
||||
name: null,
|
||||
companyName: null,
|
||||
countryCode: null,
|
||||
currentTimeZone: null,
|
||||
serviceType: null,
|
||||
avatarUrl: null,
|
||||
observedAtMs: 1_700_000_000_002,
|
||||
profileFingerprint: "profile-fingerprint-newer",
|
||||
observationStatus: "partial",
|
||||
});
|
||||
const newerResult = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [newer],
|
||||
});
|
||||
assert.equal(newerResult.status, "accepted");
|
||||
if (newerResult.status === "accepted") {
|
||||
assert.equal(newerResult.writtenProfileCount, 1);
|
||||
assert.equal(newerResult.staleProfileCount, 0);
|
||||
}
|
||||
|
||||
const equalResult = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [
|
||||
integrationProfile({
|
||||
aliId: "profile-ali-equal",
|
||||
observedAtMs: 1_700_000_000_002,
|
||||
profileFingerprint: "profile-fingerprint-equal",
|
||||
}),
|
||||
],
|
||||
});
|
||||
assert.equal(equalResult.status, "accepted");
|
||||
if (equalResult.status === "accepted") {
|
||||
assert.equal(equalResult.writtenProfileCount, 0);
|
||||
assert.equal(equalResult.staleProfileCount, 1);
|
||||
}
|
||||
|
||||
const staleResult = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [
|
||||
integrationProfile({
|
||||
aliId: "profile-ali-stale",
|
||||
accountId: "stale-account",
|
||||
loginId: "stale-login",
|
||||
name: "Stale name",
|
||||
companyName: "Stale company",
|
||||
countryCode: "US",
|
||||
currentTimeZone: -5,
|
||||
serviceType: "stale",
|
||||
avatarUrl: "https://cdn.example.com/stale.jpg",
|
||||
observedAtMs: 1_700_000_000_001,
|
||||
profileFingerprint: "profile-fingerprint-stale",
|
||||
}),
|
||||
],
|
||||
});
|
||||
assert.equal(staleResult.status, "accepted");
|
||||
if (staleResult.status === "accepted") {
|
||||
assert.equal(staleResult.writtenProfileCount, 0);
|
||||
assert.equal(staleResult.staleProfileCount, 1);
|
||||
}
|
||||
|
||||
const anotherConversation = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [
|
||||
integrationProfile({
|
||||
conversationId: "profile-conversation-2",
|
||||
aliId: "profile-ali-newer",
|
||||
accountId: null,
|
||||
loginId: null,
|
||||
name: null,
|
||||
companyName: null,
|
||||
countryCode: null,
|
||||
currentTimeZone: 5.5,
|
||||
serviceType: null,
|
||||
avatarUrl: null,
|
||||
profileFingerprint: "profile-fingerprint-2",
|
||||
}),
|
||||
],
|
||||
});
|
||||
assert.equal(anotherConversation.status, "accepted");
|
||||
|
||||
const profileRows = await sql`
|
||||
select
|
||||
conversation_id,
|
||||
ali_id,
|
||||
account_id,
|
||||
login_id,
|
||||
name,
|
||||
company_name,
|
||||
country_code,
|
||||
current_time_zone,
|
||||
service_type,
|
||||
avatar_url,
|
||||
observed_at_ms,
|
||||
profile_fingerprint,
|
||||
observation_status,
|
||||
received_at = ${receivedAt} as received_at_matches
|
||||
from onetalk_contact_profile
|
||||
where channel_account_id = ${accountId}
|
||||
order by conversation_id asc
|
||||
`;
|
||||
assert.equal(profileRows.length, 2);
|
||||
assert.deepEqual(profileRows[0], {
|
||||
conversation_id: "profile-conversation-1",
|
||||
ali_id: "profile-ali-newer",
|
||||
account_id: null,
|
||||
login_id: null,
|
||||
name: null,
|
||||
company_name: null,
|
||||
country_code: null,
|
||||
current_time_zone: null,
|
||||
service_type: null,
|
||||
avatar_url: null,
|
||||
observed_at_ms: "1700000000002",
|
||||
profile_fingerprint: "profile-fingerprint-newer",
|
||||
observation_status: "partial",
|
||||
received_at_matches: true,
|
||||
});
|
||||
assert.equal(profileRows[1]?.conversation_id, "profile-conversation-2");
|
||||
assert.equal(profileRows[1]?.ali_id, "profile-ali-newer");
|
||||
assert.equal(profileRows[1]?.current_time_zone, 5.5);
|
||||
|
||||
const futureResult = await service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [
|
||||
integrationProfile({
|
||||
conversationId: "future-profile-conversation",
|
||||
observedAtMs: receivedAt.getTime() + 5 * 60 * 1000 + 1,
|
||||
}),
|
||||
],
|
||||
});
|
||||
assert.deepEqual(futureResult, {
|
||||
status: "rejected",
|
||||
reason: "profile_observed_at_future",
|
||||
});
|
||||
|
||||
await assert.rejects(
|
||||
service.ingestProfiles({
|
||||
channelAccountId: accountId,
|
||||
profiles: [
|
||||
integrationProfile({ conversationId: "database-failure-valid" }),
|
||||
integrationProfile({
|
||||
conversationId: "database-failure-invalid",
|
||||
observationStatus:
|
||||
"invalid" as OneTalkContactProfile["observationStatus"],
|
||||
}),
|
||||
],
|
||||
}),
|
||||
OneTalkDatabaseError,
|
||||
);
|
||||
|
||||
const failedBatchRows = await sql`
|
||||
select conversation_id
|
||||
from onetalk_contact_profile
|
||||
where channel_account_id = ${accountId}
|
||||
and conversation_id like 'database-failure-%'
|
||||
`;
|
||||
assert.deepEqual([...failedBatchRows], []);
|
||||
|
||||
const foreignKeyRows = await sql`
|
||||
select count(*)::integer as count
|
||||
from information_schema.table_constraints
|
||||
where table_schema = 'public'
|
||||
and table_name = 'onetalk_contact_profile'
|
||||
and constraint_type = 'FOREIGN KEY'
|
||||
`;
|
||||
assert.equal(foreignKeyRows[0]?.count, 0);
|
||||
|
||||
const tableCommentRows = await sql`
|
||||
select obj_description('onetalk_contact_profile'::regclass) as comment
|
||||
`;
|
||||
assert.match(String(tableCommentRows[0]?.comment), /联系人当前事实/u);
|
||||
|
||||
const relationRows = await sql`
|
||||
select 'onetalk_contact_profile'::regclass::oid as oid
|
||||
`;
|
||||
const relationOid = relationRows[0]?.oid;
|
||||
assert.notEqual(relationOid, undefined);
|
||||
|
||||
const columnCommentRows = await sql`
|
||||
select attname, col_description(${relationOid}::oid, attnum) as comment
|
||||
from pg_attribute
|
||||
where attrelid = ${relationOid}::oid
|
||||
and attnum > 0
|
||||
and not attisdropped
|
||||
`;
|
||||
assert.equal(columnCommentRows.length, 15);
|
||||
assert.ok(
|
||||
columnCommentRows.every((row) => /[\u4e00-\u9fff]/u.test(String(row.comment))),
|
||||
);
|
||||
} finally {
|
||||
await sql`
|
||||
delete from onetalk_contact_profile
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
test(
|
||||
"commits mixed batch observations once with idempotent results and aggregates",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const sql = postgres(databaseUrl, { max: 1 });
|
||||
const accountId = `batch-integration-${randomUUID()}`;
|
||||
const service = createOneTalkService(createOneTalkRepository(database.db));
|
||||
const context = {
|
||||
binding: "batch-integration-binding",
|
||||
mindUserId: "batch-integration-mind-user",
|
||||
workspaceId: "batch-integration-workspace",
|
||||
channelAccountId: accountId,
|
||||
deviceId: "batch-integration-device",
|
||||
};
|
||||
const commitGuard = { assertValid: () => {} };
|
||||
|
||||
try {
|
||||
await service.discoverConversation(
|
||||
context,
|
||||
"integration-conversation",
|
||||
undefined,
|
||||
"direct",
|
||||
);
|
||||
const results = await service.observeMessages(
|
||||
context,
|
||||
[
|
||||
{
|
||||
observationSource: "history",
|
||||
message: integrationMessage("batch-message-1", 1_700_000_000_100),
|
||||
},
|
||||
{
|
||||
observationSource: "live",
|
||||
message: integrationMessage("batch-message-2", 1_700_000_000_200),
|
||||
},
|
||||
{
|
||||
observationSource: "history",
|
||||
message: integrationMessage("batch-message-1", 1_700_000_000_100),
|
||||
},
|
||||
],
|
||||
commitGuard,
|
||||
);
|
||||
|
||||
assert.deepEqual(
|
||||
results.map((result) => result.status),
|
||||
["accepted", "accepted", "duplicate"],
|
||||
);
|
||||
const messageRows = await sql`
|
||||
select message_id, observation_type
|
||||
from onetalk_message
|
||||
where channel_account_id = ${accountId}
|
||||
order by message_id
|
||||
`;
|
||||
assert.deepEqual(
|
||||
[...messageRows],
|
||||
[
|
||||
{ message_id: "batch-message-1", observation_type: "history" },
|
||||
{ message_id: "batch-message-2", observation_type: "new" },
|
||||
],
|
||||
);
|
||||
const conversationRows = await sql`
|
||||
select message_count, last_message_at_ms
|
||||
from onetalk_conversation
|
||||
where channel_account_id = ${accountId}
|
||||
and conversation_id = 'integration-conversation'
|
||||
`;
|
||||
assert.deepEqual(
|
||||
[...conversationRows],
|
||||
[{ message_count: 2, last_message_at_ms: "1700000000200" }],
|
||||
);
|
||||
} finally {
|
||||
await sql`
|
||||
delete from onetalk_message_anomaly
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql`
|
||||
delete from onetalk_message
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql`
|
||||
delete from onetalk_conversation
|
||||
where channel_account_id = ${accountId}
|
||||
`;
|
||||
await sql.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
@@ -1,490 +0,0 @@
|
||||
// 验证 OneTalk 读取 SQL 的快照与 direct 边界
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import test from "node:test";
|
||||
|
||||
import postgres from "postgres";
|
||||
|
||||
import { createDatabase, type DatabaseConnection } from "../src/database/index.ts";
|
||||
import { runMigrations } from "../src/database/migrate.ts";
|
||||
import { createOneTalkReadRepository, createOneTalkReadService } from "../src/onetalk/index.ts";
|
||||
|
||||
const databaseUrl = process.env.TEST_DATABASE_URL?.trim();
|
||||
const imageDimensionMigrationTimestamp = 1_789_102_776_000;
|
||||
|
||||
test(
|
||||
"reads only direct snapshot facts with real latest-message and profile semantics",
|
||||
{
|
||||
skip: databaseUrl
|
||||
? false
|
||||
: "TEST_DATABASE_URL is not set; PostgreSQL read integration test not run",
|
||||
timeout: 60_000,
|
||||
},
|
||||
async () => {
|
||||
if (!databaseUrl) return;
|
||||
|
||||
await runMigrations(databaseUrl);
|
||||
const database: DatabaseConnection = createDatabase(databaseUrl);
|
||||
const client = postgres(databaseUrl, { max: 1 });
|
||||
const channelAccountId = `read-integration-${randomUUID()}`;
|
||||
const asOf = new Date("2026-09-03T00:00:00.000Z");
|
||||
const afterAsOf = new Date("2026-09-04T00:00:00.000Z");
|
||||
const scope = {
|
||||
mindUserId: "integration-mind-user",
|
||||
workspaceId: "integration-workspace",
|
||||
channelAccountId,
|
||||
};
|
||||
const service = createOneTalkReadService(createOneTalkReadRepository(database.db), {
|
||||
now: () => asOf,
|
||||
});
|
||||
|
||||
try {
|
||||
await client`
|
||||
delete from drizzle.__drizzle_migrations
|
||||
where created_at = ${imageDimensionMigrationTimestamp}
|
||||
`;
|
||||
await client`
|
||||
insert into onetalk_conversation (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
conversation_kind,
|
||||
history_complete,
|
||||
first_discovered_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'migration-image',
|
||||
'direct',
|
||||
true,
|
||||
${asOf},
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
const legacyImage = {
|
||||
version: 1,
|
||||
kind: "image",
|
||||
fileId: "migration-image-file",
|
||||
extension: "jpg",
|
||||
sizeBytes: 42_000,
|
||||
width: 1280,
|
||||
height: 720,
|
||||
isOriginal: true,
|
||||
md5: "a".repeat(32),
|
||||
previewUrl: null,
|
||||
urlScope: "onetalk_session",
|
||||
};
|
||||
const canonicalImage = {
|
||||
version: 1,
|
||||
kind: "image",
|
||||
fileId: "migration-image-file",
|
||||
extension: "jpg",
|
||||
sizeBytes: 42_000,
|
||||
isOriginal: true,
|
||||
md5: "a".repeat(32),
|
||||
previewUrl: null,
|
||||
urlScope: "onetalk_session",
|
||||
};
|
||||
const imageWithOnlyWidth = {
|
||||
...canonicalImage,
|
||||
fileId: "image-with-only-width",
|
||||
width: 800,
|
||||
};
|
||||
const imageWithOnlyHeight = {
|
||||
...canonicalImage,
|
||||
fileId: "image-with-only-height",
|
||||
height: 600,
|
||||
};
|
||||
const imageWithoutDimensions = {
|
||||
...canonicalImage,
|
||||
fileId: "image-with-only-width",
|
||||
};
|
||||
const imageWithoutHeight = {
|
||||
...canonicalImage,
|
||||
fileId: "image-with-only-height",
|
||||
};
|
||||
const imageWithPreservedKey = {
|
||||
...legacyImage,
|
||||
fileId: "image-with-preserved-key",
|
||||
retainedMigrationProbe: { source: "pre-v6" },
|
||||
};
|
||||
const preservedImage = {
|
||||
...canonicalImage,
|
||||
fileId: "image-with-preserved-key",
|
||||
retainedMigrationProbe: { source: "pre-v6" },
|
||||
};
|
||||
const textSentinel = {
|
||||
version: 1,
|
||||
kind: "text",
|
||||
text: "unmodified text",
|
||||
width: 900,
|
||||
height: 901,
|
||||
};
|
||||
for (const [conversationId, messageId, content] of [
|
||||
["migration-image", "legacy-image", legacyImage],
|
||||
["migration-preservation", "image-only-width", imageWithOnlyWidth],
|
||||
["migration-preservation", "image-only-height", imageWithOnlyHeight],
|
||||
["migration-preservation", "image-preserved-key", imageWithPreservedKey],
|
||||
["migration-preservation", "text-sentinel", textSentinel],
|
||||
] as const) {
|
||||
await client`
|
||||
insert into onetalk_message (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
message_id,
|
||||
sender_id,
|
||||
binding,
|
||||
mind_user_id,
|
||||
workspace_id,
|
||||
device_id,
|
||||
direction,
|
||||
observation_type,
|
||||
sent_at_ms,
|
||||
content,
|
||||
participant_ids,
|
||||
read_status,
|
||||
message_status,
|
||||
unread_count,
|
||||
first_observed_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
${conversationId},
|
||||
${messageId},
|
||||
'sender-1',
|
||||
'binding-1',
|
||||
'mind-user-1',
|
||||
'workspace-1',
|
||||
'device-1',
|
||||
'received',
|
||||
'history',
|
||||
1,
|
||||
${client.json(content)},
|
||||
${["sender-1", channelAccountId]},
|
||||
1,
|
||||
2,
|
||||
0,
|
||||
${asOf},
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
}
|
||||
await runMigrations(databaseUrl);
|
||||
const migratedRows = await client<
|
||||
{
|
||||
content: Record<string, unknown>;
|
||||
message_id: string;
|
||||
}[]
|
||||
>`
|
||||
select message_id, content
|
||||
from onetalk_message
|
||||
where channel_account_id = ${channelAccountId}
|
||||
order by message_id
|
||||
`;
|
||||
assert.deepEqual(migratedRows, [
|
||||
{ message_id: "image-only-height", content: imageWithoutHeight },
|
||||
{ message_id: "image-only-width", content: imageWithoutDimensions },
|
||||
{ message_id: "image-preserved-key", content: preservedImage },
|
||||
{ message_id: "legacy-image", content: canonicalImage },
|
||||
{ message_id: "text-sentinel", content: textSentinel },
|
||||
]);
|
||||
const migratedHistory = await service.readHistory({
|
||||
scope,
|
||||
conversationId: "migration-image",
|
||||
});
|
||||
assert.equal(migratedHistory.status, "accepted");
|
||||
if (migratedHistory.status !== "accepted") return;
|
||||
assert.deepEqual(migratedHistory.messages[0]?.content, canonicalImage);
|
||||
await runMigrations(databaseUrl);
|
||||
const migrationRecords = await client<{ count: string }[]>`
|
||||
select count(*) as count
|
||||
from drizzle.__drizzle_migrations
|
||||
where created_at = ${imageDimensionMigrationTimestamp}
|
||||
`;
|
||||
assert.equal(migrationRecords[0]?.count, "1");
|
||||
|
||||
for (const [conversationId, conversationKind, discoveredAt, lastMessageAtMs] of [
|
||||
["direct-a", "direct", asOf, 300],
|
||||
["direct-b", "direct", asOf, 250],
|
||||
["direct-c", "direct", asOf, 200],
|
||||
["group", "group", asOf, 400],
|
||||
["unknown", null, asOf, 500],
|
||||
["future-direct", "direct", afterAsOf, 600],
|
||||
] as const) {
|
||||
await client`
|
||||
insert into onetalk_conversation (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
conversation_kind,
|
||||
last_message_at_ms,
|
||||
first_discovered_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
${conversationId},
|
||||
${conversationKind},
|
||||
${lastMessageAtMs},
|
||||
${discoveredAt},
|
||||
${discoveredAt}
|
||||
)
|
||||
`;
|
||||
}
|
||||
await client`
|
||||
insert into onetalk_contact_profile (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
ali_id,
|
||||
name,
|
||||
avatar_url,
|
||||
observed_at_ms,
|
||||
profile_fingerprint,
|
||||
observation_status,
|
||||
received_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'direct-a',
|
||||
'ali-a',
|
||||
'Alpha Profile',
|
||||
'https://cdn.example.com/alpha.jpg',
|
||||
100,
|
||||
'profile-a',
|
||||
'confirmed',
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
await client`
|
||||
insert into onetalk_contact_profile (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
ali_id,
|
||||
name,
|
||||
observed_at_ms,
|
||||
profile_fingerprint,
|
||||
observation_status,
|
||||
received_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'direct-c',
|
||||
'ali-c',
|
||||
'Later Profile',
|
||||
100,
|
||||
'profile-c',
|
||||
'confirmed',
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
await client`
|
||||
insert into onetalk_contact_profile (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
ali_id,
|
||||
name,
|
||||
observed_at_ms,
|
||||
profile_fingerprint,
|
||||
observation_status,
|
||||
received_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'profile-only',
|
||||
'ali-only',
|
||||
'Profile without conversation',
|
||||
100,
|
||||
'profile-only',
|
||||
'confirmed',
|
||||
${asOf}
|
||||
)
|
||||
`;
|
||||
for (const [messageId, sentAtMs, firstObservedAt] of [
|
||||
["anchor-only", 100, asOf],
|
||||
["latest-a", 200, asOf],
|
||||
["latest-z", 200, asOf],
|
||||
["after-snapshot", 900, afterAsOf],
|
||||
] as const) {
|
||||
await client`
|
||||
insert into onetalk_message (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
message_id,
|
||||
sender_id,
|
||||
binding,
|
||||
mind_user_id,
|
||||
workspace_id,
|
||||
device_id,
|
||||
direction,
|
||||
observation_type,
|
||||
sent_at_ms,
|
||||
content,
|
||||
participant_ids,
|
||||
read_status,
|
||||
message_status,
|
||||
unread_count,
|
||||
first_observed_at,
|
||||
last_observed_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'direct-a',
|
||||
${messageId},
|
||||
'sender-1',
|
||||
'binding-1',
|
||||
'mind-user-1',
|
||||
'workspace-1',
|
||||
'device-1',
|
||||
'received',
|
||||
'history',
|
||||
${sentAtMs},
|
||||
${client.json({ version: 1, kind: "text", text: "visible text" })},
|
||||
${["sender-1", channelAccountId]},
|
||||
1,
|
||||
2,
|
||||
0,
|
||||
${firstObservedAt},
|
||||
${firstObservedAt}
|
||||
)
|
||||
`;
|
||||
}
|
||||
await client`
|
||||
update onetalk_conversation
|
||||
set message_count = 4,
|
||||
latest_message_id = 'anchor-only'
|
||||
where channel_account_id = ${channelAccountId}
|
||||
and conversation_id = 'direct-a'
|
||||
`;
|
||||
|
||||
const firstPage = await service.listConversations({ scope, limit: 1 });
|
||||
assert.equal(firstPage.status, "accepted");
|
||||
if (firstPage.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
firstPage.conversations.map(({ conversationId }) => conversationId),
|
||||
["direct-a"],
|
||||
);
|
||||
assert.equal(firstPage.conversations[0]?.name, "Alpha Profile");
|
||||
assert.equal(firstPage.page.hasMore, true);
|
||||
assert.ok(firstPage.page.nextCursor);
|
||||
|
||||
await client`
|
||||
insert into onetalk_contact_profile (
|
||||
channel_account_id,
|
||||
conversation_id,
|
||||
ali_id,
|
||||
name,
|
||||
observed_at_ms,
|
||||
profile_fingerprint,
|
||||
observation_status,
|
||||
received_at
|
||||
) values (
|
||||
${channelAccountId},
|
||||
'direct-b',
|
||||
'ali-b',
|
||||
'Current second profile',
|
||||
101,
|
||||
'profile-b',
|
||||
'confirmed',
|
||||
${afterAsOf}
|
||||
)
|
||||
`;
|
||||
const secondPage = await service.listConversations({
|
||||
scope,
|
||||
cursor: firstPage.page.nextCursor,
|
||||
limit: 1,
|
||||
});
|
||||
assert.equal(secondPage.status, "accepted");
|
||||
if (secondPage.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
secondPage.conversations.map(({ conversationId }) => conversationId),
|
||||
["direct-b"],
|
||||
);
|
||||
assert.equal(secondPage.conversations[0]?.name, "Current second profile");
|
||||
assert.equal(secondPage.page.hasMore, true);
|
||||
assert.ok(secondPage.page.nextCursor);
|
||||
|
||||
const filtered = await service.listConversations({
|
||||
scope,
|
||||
query: " alpha ",
|
||||
limit: 10,
|
||||
});
|
||||
assert.equal(filtered.status, "accepted");
|
||||
if (filtered.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
filtered.conversations.map(({ conversationId }) => conversationId),
|
||||
["direct-a"],
|
||||
);
|
||||
|
||||
const profileFiltered = await service.listConversations({
|
||||
scope,
|
||||
query: "later",
|
||||
limit: 1,
|
||||
});
|
||||
assert.equal(profileFiltered.status, "accepted");
|
||||
if (profileFiltered.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
profileFiltered.conversations.map(({ conversationId }) => conversationId),
|
||||
["direct-c"],
|
||||
);
|
||||
assert.equal(profileFiltered.page.hasMore, false);
|
||||
assert.equal(profileFiltered.page.nextCursor, null);
|
||||
|
||||
const detailWithoutProfile = await service.readConversation({
|
||||
scope,
|
||||
conversationId: "direct-b",
|
||||
});
|
||||
assert.equal(detailWithoutProfile.status, "accepted");
|
||||
if (detailWithoutProfile.status !== "accepted") return;
|
||||
assert.equal(detailWithoutProfile.conversation.avatarUrl, null);
|
||||
|
||||
const ordinaryHistory = await service.readHistory({
|
||||
scope,
|
||||
conversationId: "direct-a",
|
||||
fromSentAtMs: 100,
|
||||
toSentAtMs: 900,
|
||||
limit: 2,
|
||||
});
|
||||
assert.equal(ordinaryHistory.status, "accepted");
|
||||
if (ordinaryHistory.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
ordinaryHistory.messages.map(({ messageId }) => messageId),
|
||||
["latest-a", "latest-z"],
|
||||
);
|
||||
assert.equal(ordinaryHistory.page.hasMore, true);
|
||||
assert.ok(ordinaryHistory.page.nextCursor);
|
||||
const olderHistory = await service.readHistory({
|
||||
scope,
|
||||
conversationId: "direct-a",
|
||||
fromSentAtMs: 100,
|
||||
toSentAtMs: 900,
|
||||
cursor: ordinaryHistory.page.nextCursor,
|
||||
limit: 2,
|
||||
});
|
||||
assert.equal(olderHistory.status, "accepted");
|
||||
if (olderHistory.status !== "accepted") return;
|
||||
assert.deepEqual(
|
||||
olderHistory.messages.map(({ messageId }) => messageId),
|
||||
["anchor-only"],
|
||||
);
|
||||
assert.equal(olderHistory.page.hasMore, false);
|
||||
assert.deepEqual(
|
||||
await service.readHistory({
|
||||
scope,
|
||||
conversationId: "direct-a",
|
||||
purpose: "communication_summary_read",
|
||||
}),
|
||||
{ status: "rejected", reason: "history_incomplete" },
|
||||
);
|
||||
} finally {
|
||||
await runMigrations(databaseUrl);
|
||||
await client`
|
||||
delete from onetalk_message
|
||||
where channel_account_id = ${channelAccountId}
|
||||
`;
|
||||
await client`
|
||||
delete from onetalk_contact_profile
|
||||
where channel_account_id = ${channelAccountId}
|
||||
`;
|
||||
await client`
|
||||
delete from onetalk_conversation
|
||||
where channel_account_id = ${channelAccountId}
|
||||
`;
|
||||
await client.end({ timeout: 5 });
|
||||
await database.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
Reference in New Issue
Block a user