Files

T

浅念 7f4f3be8d8 插件

2026-08-27 16:12:37 +08:00

46 KiB

Raw Blame History

阿里询盘后台 - 后端 MVP 实施计划 (Plan 1, v2)

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 搭建询盘后台的 Node.js + TypeScript + Express + Prisma + MySQL 后端,实现 /api/ingest(插件推送幂等入库) + 管理员 JWT 登录 + 询盘列表/详情/消息/单询盘导出/统计/列表导出 API,可用 supertest / curl 完整验证。

Architecture: 分层(routes / services / middleware / utils)。幂等靠唯一键 + 早返回(messagesCount 存总量 + has_failed_batches 防失败批次被跳过);消息批量用 createMany({ skipDuplicates: true }) 分批 500、失败批次隔离。PrismaClient 惰性初始化(适配 testcontainers)。ESM("type":"module")。所有时间存 UTC。

Tech Stack: Node 18, TypeScript 5(ESM), Express 4, Prisma 5, MySQL 8, zod, jsonwebtoken, bcryptjs, express-rate-limit, vitest, supertest, testcontainers(mysql)。

关联 spec: docs/superpowers/specs/2026-07-13-alibaba-inquiry-backend-design.md(v3)

v2 修订说明: 修正 plan review 发现的 4 致命(ESM type / messagesCount 用总量 / 增量回归测试 / 惰性 Prisma)+ 关键重要项(.env 不被覆盖、Task9 单一实现、vitest singleFork、单询盘导出 API、customerUpdate 不覆盖 null、统一测试脚手架、游标 zod 校验、显式 equals、docker exec -i)+ 高价值建议(SIGTERM、buildWhere 复用、e2e 补全、DoD 增量测试)。

前提: 工作目录 d:\alibb-server\backend\,在该目录 git init。Docker Desktop 已装。

文件结构

d:\alibb-server\backend\
  prisma\schema.prisma              # 4 表 + 索引(手写,不用 prisma init)
  prisma\migrations\                # prisma migrate 生成
  src\
    config\env.ts                   # loadEnv() 函数(延迟校验,适配测试覆写)
    lib\prisma.ts                   # getPrisma() 惰性单例
    middleware\{jwtAuth,ingestAuth,rateLimit,error}.ts
    routes\{auth,ingest,inquiries,stats,export}.ts
    services\{auth,ingest,inquiry,stats}.service.ts
    utils\{batchInsert,time}.ts
    app.ts                          # createApp() 装配
    server.ts                       # listen + bootstrapAdmin + SIGTERM
  tests\
    setup.ts                        # setupApp(): 起 testcontainer + migrate + 返回 {app, prisma, resetDb}
    fixtures.ts                     # makePayload / seedInquiries
    {health,auth,ingest,batchInsert,inquiries,stats,export}.test.ts
  Dockerfile  docker-compose.yml  scripts\e2e.sh
  package.json(type:module)  tsconfig.json  .env.example  .env  vitest.config.ts(singleFork)

Task 1: 项目骨架(ESM + 依赖 + vitest singleFork)

Files: package.json, tsconfig.json, .env.example, .gitignore, vitest.config.ts

Step 1: 初始化

mkdir -p d:/alibb-server/backend && cd d:/alibb-server/backend
git init && npm init -y

Step 2: 装依赖

npm install express @prisma/client prisma zod jsonwebtoken bcryptjs express-rate-limit dotenv cors morgan mysql2
npm install -D typescript tsx @types/express @types/node @types/jsonwebtoken @types/bcryptjs @types/cors @types/morgan vitest supertest @types/supertest testcontainers @testcontainers/mysql

Step 3: package.json 关键字段(必须含 "type":"module",否则 ESM build 跑不起来 —— 修正 F1)

{
  "name": "inquiry-backend",
  "type": "module",
  "scripts": {
    "dev": "tsx watch src/server.ts",
    "build": "tsc",
    "start": "node dist/server.js",
    "test": "vitest run",
    "db:migrate": "prisma migrate dev",
    "db:deploy": "prisma migrate deploy",
    "db:generate": "prisma generate"
  }
}

Step 4: tsconfig.json

{ "compilerOptions": { "target": "ES2022", "module": "NodeNext", "moduleResolution": "NodeNext",
  "strict": true, "esModuleInterop": true, "outDir": "dist", "rootDir": "src",
  "resolveJsonModule": true, "skipLibCheck": true, "types": ["vitest/globals"] },
  "include": ["src/**/*"] }

NodeNext 要求 import 路径带 .js 后缀(写 .js,tsc 输出 .js)。测试文件由 vitest(tsx)跑,不受 tsc 管。

Step 5: .gitignore

node_modules/ dist/ .env *.log

Step 6: .env.example(并 cp .env.example .env —— 修正 I2,避免后续被覆盖)

DATABASE_URL="mysql://root:root@localhost:3306/inquiry"
INGEST_TOKEN="dev-ingest-token-1234"
JWT_SECRET="dev-jwt-secret-1234"
JWT_EXPIRES_IN="7d"
ADMIN_USERNAME="admin"
ADMIN_PASSWORD="admin-password-1234"
PORT=8080
NODE_ENV=development

cp .env.example .env

注:不再有 JWT_REFRESH_SECRET(MVP 单密钥,refresh 用 JWT_SECRET;修正 I7)。TEST_DATABASE_URL 由测试运行时注入,不入 env。

Step 7: vitest.config.ts(singleFork 串行,避免并行起多个 mysql 容器 —— 修正 I4)

import { defineConfig } from 'vitest/config';
export default defineConfig({
  test: { globals: true, testTimeout: 90_000, hookTimeout: 90_000,
    poolOptions: { forks: { singleFork: true } } }
});

Step 8: 临时 src/server.ts = console.log('placeholder');
Step 9: Commit

git add -A && git commit -m "chore: scaffold ESM backend project"

Task 2: Prisma Schema(手写,跳过 prisma init)

Files: prisma/schema.prisma(手写);起本地 mysql 容器;首次迁移

Step 1: 起本地 MySQL 容器(开发用)

docker run -d --name inquiry-mysql -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=inquiry -p 3306:3306 mysql:8 --max_allowed_packet=128M --character-set-server=utf8mb4

Step 2: 手写 prisma/schema.prisma(不用 prisma init,它会覆盖 .env —— 修正 I2)。完整对应 spec §6(含 v3 has_failed_batches、Direction.system、MEDIUMTEXT content、复合索引)

generator client { provider = "prisma-client-js" }
datasource db { provider = "mysql"; url = env("DATABASE_URL") }

enum Direction { incoming outgoing system }

model Customer {
  id              BigInt    @id @default(autoincrement())
  customerId      String    @unique @map("customer_id") @db.VarChar(64)
  name            String?   @db.VarChar(128)
  country         String?   @index @db.VarChar(64)
  email           String?   @db.VarChar(128)
  company         String?   @db.VarChar(255)
  registerTime    DateTime? @map("register_time")
  avatarUrl       String?   @map("avatar_url") @db.VarChar(512)
  buyerTags       String?   @map("buyer_tags") @db.VarChar(255)
  buyerFeatures   String?   @map("buyer_features") @db.VarChar(255)
  bProductViews   Int?      @map("behavior_product_views")
  bValidInquiries Int?      @map("behavior_valid_inquiries")
  bValidRfq       Int?      @map("behavior_valid_rfq")
  bLoginDays      Int?      @map("behavior_login_days")
  bSpamInquiries  Int?      @map("behavior_spam_inquiries")
  bTopIndustry    String?   @map("behavior_top_industry") @db.VarChar(255)
  inquiries       Inquiry[]
  createdAt       DateTime  @default(now()) @map("created_at")
  updatedAt       DateTime  @updatedAt @map("updated_at")
  @@map("customers")
}

model Inquiry {
  id               BigInt    @id @default(autoincrement())
  inquiryId        String    @unique @map("inquiry_id") @db.VarChar(32)
  customerId       String?   @index @map("customer_id") @db.VarChar(64)
  manager          String?   @index @db.VarChar(128)
  createTime       DateTime  @map("create_time")
  updateTime       DateTime  @map("update_time")
  status           String?   @db.VarChar(64)
  msgPreview       String?   @map("msg_preview") @db.VarChar(512)
  isFlagged        Boolean   @default(false) @map("is_flagged")
  messagesCount    Int       @default(0) @map("messages_count")
  lastSendTime     DateTime? @map("last_send_time")
  hasFailedBatches Boolean   @default(false) @map("has_failed_batches")
  customer         Customer? @relation(fields: [customerId], references: [customerId])
  messages         ChatMessage[]
  createdAt        DateTime  @default(now()) @map("created_at")
  updatedAt        DateTime  @updatedAt @map("updated_at")
  @@map("inquiries")
}

model ChatMessage {
  id          BigInt   @id @default(autoincrement())
  messageId   String   @unique @map("message_id") @db.VarChar(64)
  inquiryId   String   @index @map("inquiry_id") @db.VarChar(32)
  sender      String?  @db.VarChar(128)
  direction   Direction
  sendTime    DateTime @map("send_time")
  content     String   @db.MediumText
  translation String?  @db.Text
  inquiry     Inquiry  @relation(fields: [inquiryId], references: [inquiryId])
  createdAt   DateTime @default(now()) @map("created_at")
  @@index([inquiryId, sendTime, messageId])
  @@map("chat_messages")
}

model User {
  id           BigInt   @id @default(autoincrement())
  username     String   @unique
  passwordHash String   @map("password_hash")
  createdAt    DateTime @default(now()) @map("created_at")
  @@map("users")
}

Step 3: 迁移 + 生成 client

npx prisma migrate dev --name init
npx prisma generate

Step 4: 验证表结构(Git Bash 用 -i 不用 -it —— 修正 I9)

docker exec -i inquiry-mysql mysql -uroot -proot -e "USE inquiry; SHOW TABLES; DESC chat_messages;"

Expected:4 表;chat_messages.content 为 MEDIUMTEXT;message_id UNIQUE;索引含 inquiry_id_send_time_message_id_idx。

Step 5: Commit

git add -A && git commit -m "feat(db): prisma schema (4 tables, has_failed_batches, system enum, composite index)"

Task 3: 惰性 env + 惰性 Prisma(修正 F4)

Files: src/config/env.ts, src/lib/prisma.ts

Step 1: src/config/env.ts(函数化 loadEnv,延迟校验,让测试能先覆写 process.env)

import { z } from 'zod';

const schema = z.object({
  DATABASE_URL: z.string().regex(/^mysql:\/\//, 'must start with mysql://'),
  INGEST_TOKEN: z.string().min(8),
  JWT_SECRET: z.string().min(8),
  JWT_EXPIRES_IN: z.string().default('7d'),
  ADMIN_USERNAME: z.string(),
  ADMIN_PASSWORD: z.string().min(8),
  PORT: z.coerce.number().default(8080),
  NODE_ENV: z.enum(['development', 'production']).default('development'),
});

let _env: z.infer<typeof schema> | null = null;
export function loadEnv() {
  if (_env) return _env;
  // 首次调用时加载 .env(server 用);测试会在调用前手动设 process.env
  if (!process.env.DATABASE_URL) {
    const dotenv = await import('dotenv'); await dotenv.config();
  }
  ...
}

注:loadEnv 不能用顶层 import 'dotenv/config'(会立即固化),改用动态 import + 顶层 await 函数。简化写法:

import { z } from 'zod';
import { config as dotenvConfig } from 'dotenv';
dotenvConfig();   // 仅设置 process.env,不校验
const schema = z.object({ /* 同上,无 JWT_REFRESH_SECRET */ });
let _env: z.infer<typeof schema> | null = null;
export function loadEnv() {
  if (_env) return _env;
  const parsed = schema.safeParse(process.env);
  if (!parsed.success) {
    console.error('❌ invalid env:', parsed.error.flatten().fieldErrors);
    process.exit(1);
  }
  _env = parsed.data;
  return _env;
}
export const env = new Proxy({} as z.infer<typeof schema>, { get: (_, k) => (loadEnv() as any)[k] });

env 用 Proxy 延迟:模块加载时不校验,首次访问才 loadEnv()。测试在 import app 前覆写 process.env,Proxy 自然取到测试值。

Step 2: src/lib/prisma.ts(惰性单例,修正 F4)

import { PrismaClient } from '@prisma/client';
let _prisma: PrismaClient | null = null;
export function getPrisma(): PrismaClient {
  if (!_prisma) _prisma = new PrismaClient();
  return _prisma;
}
export async function disconnectPrisma() { if (_prisma) { await _prisma.$disconnect(); _prisma = null; } }

所有 service 用 getPrisma(),不在模块顶层实例化。测试在 setupApp 里先覆写 process.env.DATABASE_URL 再首次调 getPrisma()。

Step 3: Commit

git add -A && git commit -m "feat(config): lazy env proxy and prisma singleton (testcontainer-friendly)"

Task 4: Express app + 健康检查 + 错误中间件 + SIGTERM

Files: src/middleware/error.ts, src/app.ts, src/server.ts

Step 1: src/middleware/error.ts

import type { ErrorRequestHandler } from 'express';
export const errorHandler: ErrorRequestHandler = (err, _req, res, _next) => {
  console.error(err);
  res.status(err.status ?? 500).json({ error: err.message ?? 'internal error' });
};

Step 2: src/app.ts

import express from 'express';
import cors from 'cors';
import morgan from 'morgan';
import { errorHandler } from './middleware/error.js';

export function createApp() {
  const app = express();
  // MVP:CORS 全开(ingest 靠 token 鉴权;后台 Plan 3 联调时收窄到前端域名)
  app.use(cors({ origin: '*' }));
  app.use(express.json({ limit: '50mb' }));
  if (process.env.NODE_ENV !== 'test') app.use(morgan('dev'));
  app.get('/api/health', (_req, res) => res.json({ ok: true }));
  app.use(errorHandler);
  return app;
}

Step 3: tests/health.test.ts(纯单元,不起容器)

import { describe, it, expect } from 'vitest';
import request from 'supertest';
import { createApp } from '../src/app.js';
describe('health', () => {
  it('ok', async () => {
    const res = await request(createApp()).get('/api/health');
    expect(res.status).toBe(200);
    expect(res.body).toEqual({ ok: true });
  });
});

Step 4: 跑测试通过

npm test -- health   # 应 1 passed

Step 5: src/server.ts(含 SIGTERM 优雅关停 —— S9)

import { createApp } from './app.js';
import { loadEnv } from './config/env.js';
import { bootstrapAdmin } from './services/auth.service.js';
import { disconnectPrisma } from './lib/prisma.js';

async function main() {
  loadEnv();
  const { env } = await import('./config/env.js');
  await bootstrapAdmin();
  const app = createApp();
  const server = app.listen(env.PORT, () => console.log(`listening on ${env.PORT}`));
  process.on('SIGTERM', async () => { server.close(); await disconnectPrisma(); process.exit(0); });
}
main();

Step 6: Commit feat(app): express app, health, error handler, SIGTERM

Task 5: 管理员 bootstrap + JWT 登录 + 统一测试脚手架

Files: src/services/auth.service.ts, src/routes/auth.ts, tests/setup.ts, tests/auth.test.ts;Modify src/app.ts

Step 1: tests/setup.ts(统一脚手架 —— 修正 I8/F4。所有集成测试都用 setupApp())

import { MySqlContainer } from '@testcontainers/mysql';
import type { StartedTestContainer } from 'testcontainers';
import { execSync } from 'node:child_process';
import { getPrisma, disconnectPrisma } from '../src/lib/prisma.js';
import { createApp } from '../src/app.js';
// 注:MySqlContainer 默认 user=root password=test

let container: StartedTestContainer | null = null;

export async function setupApp() {
  if (!container) {
    container = await new MySqlContainer('mysql:8').withDatabase('t').start();
    process.env.DATABASE_URL = `mysql://root:test@${container.getHost()}:${container.getMappedPort(3306)}/t`;
    process.env.INGEST_TOKEN = 'dev-ingest-token-1234';
    process.env.JWT_SECRET = 'dev-jwt-secret-1234';
    process.env.ADMIN_USERNAME = 'admin';
    process.env.ADMIN_PASSWORD = 'admin-password-1234';
    execSync('npx prisma migrate deploy', { stdio: 'inherit', env: process.env });
  }
  const prisma = getPrisma();
  const app = createApp();
  return { app, prisma };
}

export async function resetDb() {
  const prisma = getPrisma();
  await prisma.$executeRawUnsafe('SET FOREIGN_KEY_CHECKS=0');
  for (const t of ['chat_messages', 'inquiries', 'customers', 'users']) {
    await prisma.$executeRawUnsafe(`DELETE FROM ${t}`);
  }
  await prisma.$executeRawUnsafe('SET FOREIGN_KEY_CHECKS=1');
}

export async function teardown() { await disconnectPrisma(); await container?.stop(); container = null; }

关键:先起容器 → 覆写 process.env.DATABASE_URL → 才首次 getPrisma()(惰性,取到容器地址)。vitest singleFork 保证一个进程一个容器。

Step 2: tests/auth.test.ts

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import request from 'supertest';
import { setupApp, resetDb, teardown } from './setup.js';

let app: any;
beforeAll(async () => { ({ app } = await setupApp()); }, 90000);
afterAll(async () => { await teardown(); });

describe('auth', () => {
  it('login with admin env creds returns token', async () => {
    const res = await request(app).post('/api/auth/login').send({ username: 'admin', password: 'admin-password-1234' });
    expect(res.status).toBe(200);
    expect(res.body.token).toMatch(/^[A-Za-z0-9._-]+$/);
  });
  it('wrong password 401', async () => {
    const res = await request(app).post('/api/auth/login').send({ username: 'admin', password: 'wrong' });
    expect(res.status).toBe(401);
  });
});

Step 3: src/services/auth.service.ts

import bcrypt from 'bcryptjs';
import jwt from 'jsonwebtoken';
import { getPrisma } from '../lib/prisma.js';
import { env } from '../config/env.js';

export async function bootstrapAdmin() {
  const prisma = getPrisma();
  if (await prisma.user.count() === 0) {
    await prisma.user.create({ data: { username: env.ADMIN_USERNAME, passwordHash: await bcrypt.hash(env.ADMIN_PASSWORD, 10) } });
    console.log(`seeded admin "${env.ADMIN_USERNAME}"`);
  }
}

export async function login(username: string, password: string) {
  const prisma = getPrisma();
  const user = await prisma.user.findUnique({ where: { username } });
  if (!user || !(await bcrypt.compare(password, user.passwordHash)))
    throw Object.assign(new Error('invalid credentials'), { status: 401 });
  return { token: jwt.sign({ sub: user.id, u: user.username }, env.JWT_SECRET, { expiresIn: env.JWT_EXPIRES_IN }),
           expiresIn: env.JWT_EXPIRES_IN };
}

注:测试里 bootstrapAdmin 由 /api/auth/login 首次调用时触发?不——改为在 createApp 或 setupApp 末尾调一次。实现时在 setupApp() 末尾加 await bootstrapAdmin();(或在 server.ts 启动调;测试环境 setupApp 调)。

Step 4: src/routes/auth.ts

import { Router } from 'express';
import jwt from 'jsonwebtoken';
import { login } from '../services/auth.service.js';
import { env } from '../config/env.js';
import { jwtAuth } from '../middleware/jwtAuth.js';

export const authRouter = Router();
authRouter.post('/login', async (req, res, next) => {
  try { res.json(await login(req.body.username, req.body.password)); } catch (e) { next(e); }
});
// refresh 用同一个 JWT_SECRET(MVP 单密钥)
authRouter.post('/refresh', jwtAuth, (req, res) => {
  const u = (req as any).user;
  res.json({ token: jwt.sign({ sub: u.sub, u: u.u }, env.JWT_SECRET, { expiresIn: env.JWT_EXPIRES_IN }), expiresIn: env.JWT_EXPIRES_IN });
});

Step 5: src/middleware/jwtAuth.ts

import type { RequestHandler } from 'express';
import jwt from 'jsonwebtoken';
import { env } from '../config/env.js';
export const jwtAuth: RequestHandler = (req, res, next) => {
  const h = req.headers.authorization || '';
  const token = h.startsWith('Bearer ') ? h.slice(7) : '';
  try { (req as any).user = jwt.verify(token, env.JWT_SECRET); next(); }
  catch { res.status(401).json({ error: 'unauthorized' }); }
};

Step 6: src/app.ts 挂载

import { authRouter } from './routes/auth.js';
// health 之后、errorHandler 之前:
app.use('/api/auth', authRouter);

Step 7: 测试通过 + Commit feat(auth): admin bootstrap, login, refresh, jwt middleware

Task 6: /api/ingest — 鉴权 + 限流 + payload zod 校验

Files: src/middleware/{ingestAuth,rateLimit}.ts, src/routes/ingest.ts;Modify src/app.ts

Step 1: src/middleware/ingestAuth.ts(timingSafeEqual —— S1)

import type { RequestHandler } from 'express';
import { timingSafeEqual } from 'node:crypto';
import { env } from '../config/env.js';
export const ingestAuth: RequestHandler = (req, res, next) => {
  const h = req.headers.authorization || '';
  const token = h.startsWith('Bearer ') ? h.slice(7) : '';
  const want = env.INGEST_TOKEN;
  const ok = token.length === want.length && timingSafeEqual(Buffer.from(token), Buffer.from(want));
  if (!ok) return res.status(401).json({ error: 'invalid ingest token' });
  next();
};

Step 2: src/middleware/rateLimit.ts(MVP IP 维度 —— S3)

import rateLimit from 'express-rate-limit';
// 注:spec 写 60/min/token,MVP 用 IP 维度(单操作者够用);多操作者后切 key
export const ingestLimiter = rateLimit({ windowMs: 60_000, max: 60, standardHeaders: true, legacyHeaders: false });

Step 3: tests/fixtures.ts

export function makePayload(o: any = {}) {
  return {
    inquiry: { inquiryId: '13776617920', manager: 'cara Hu', createTime: '2026-03-13T10:11:50Z',
      updateTime: '2026-07-13T10:11:50Z', status: '洽谈中', msgPreview: 'hi', isFlagged: false, ...o.inquiry },
    customer: { customerId: 'cid-1', name: 'Carlos', country: 'PE', email: null, company: 'World Shop',
      registerTime: '2025-06-23T00:00:00Z', avatarUrl: '', buyerTags: 'TM沟通', buyerFeatures: '',
      behavior: { productViews: null, validInquiries: null, validRfq: 39 }, ...o.customer },
    messages: [
      { messageId: 'm1', sendTime: '2026-07-13T02:11:50Z', sender: 'cara Hu', direction: 'outgoing', content: 'hello', translation: '' },
      { messageId: 'm2', sendTime: '2026-07-13T02:12:00Z', sender: 'Carlos', direction: 'incoming', content: 'hi', translation: '' },
    ],
    ...o,
  };
}

Step 4: src/routes/ingest.ts(zod schema + 路由;messages.min(1) 防 S4 空数组边界)

import { Router } from 'express';
import { z } from 'zod';
import { ingestAuth } from '../middleware/ingestAuth.js';
import { ingestLimiter } from '../middleware/rateLimit.js';
import { ingest } from '../services/ingest.service.js';

const MessageSchema = z.object({
  messageId: z.string().min(1),
  sendTime: z.string().datetime(),
  sender: z.string().optional(),
  direction: z.enum(['incoming', 'outgoing', 'system']),
  content: z.string(),
  translation: z.string().optional().default(''),
});
const BehaviorSchema = z.object({
  productViews: z.number().nullable().optional(), validInquiries: z.number().nullable().optional(),
  validRfq: z.number().nullable().optional(), loginDays: z.number().nullable().optional(),
  spamInquiries: z.number().nullable().optional(), topIndustry: z.string().nullable().optional(),
}).optional();
const PayloadSchema = z.object({
  inquiry: z.object({
    inquiryId: z.string().min(1), manager: z.string().optional(),
    createTime: z.string().datetime(), updateTime: z.string().datetime(),
    status: z.string().optional(), msgPreview: z.string().optional(), isFlagged: z.boolean().optional().default(false),
  }),
  customer: z.object({
    customerId: z.string().min(1), name: z.string().optional(), country: z.string().nullable().optional(),
    email: z.string().nullable().optional(), company: z.string().nullable().optional(),
    registerTime: z.string().datetime().nullable().optional(), avatarUrl: z.string().optional(),
    buyerTags: z.string().optional(), buyerFeatures: z.string().optional(), behavior: BehaviorSchema,
  }).nullable(),
  messages: z.array(MessageSchema).min(1),
});

export const ingestRouter = Router();
ingestRouter.post('/', ingestLimiter, ingestAuth, async (req, res, next) => {
  const parsed = PayloadSchema.safeParse(req.body);
  if (!parsed.success) return res.status(400).json({ error: 'invalid payload', issues: parsed.error.flatten() });
  try { res.json(await ingest(parsed.data)); } catch (e) { next(e); }
});

Step 5: 占位 src/services/ingest.service.ts = export async function ingest(p:any){ return {ok:true,inserted:0,failedBatches:0,earlyReturn:false}; }
Step 6: src/app.ts 挂 app.use('/api/ingest', ingestRouter)
Step 7: 写 tests/ingest.test.ts 的鉴权/校验部分(用 setupApp 模板)

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import request from 'supertest';
import { setupApp, resetDb, teardown } from './setup.js';
import { makePayload } from './fixtures.js';
let app: any;
beforeAll(async () => { ({ app } = await setupApp()); }, 90000);
afterAll(teardown);

describe('ingest auth/validation', () => {
  it('401 without token', async () => {
    expect((await request(app).post('/api/ingest').send(makePayload())).status).toBe(401);
  });
  it('400 bad payload', async () => {
    expect((await request(app).post('/api/ingest').set('Authorization', 'Bearer dev-ingest-token-1234').send({ inquiry: {} })).status).toBe(400);
  });
});

Step 8: 测试通过 + Commit feat(ingest): auth, rate-limit, payload validation

Task 7: ingest 早返回 + customer/inquiry upsert(customerUpdate 不覆盖 null)

Files: src/services/ingest.service.ts, src/utils/time.ts

Step 1: src/utils/time.ts

export const maxSendTime = (msgs: { sendTime: string }[]) =>
  msgs.length ? new Date(Math.max(...msgs.map(m => Date.parse(m.sendTime)))) : null;

Step 2: src/services/ingest.service.ts(早返回 + 主表 upsert;customerUpdate 用 Prisma.skip 不覆盖 null —— 修正 I6)

import { Prisma } from '@prisma/client';
import { getPrisma } from '../lib/prisma.js';
import { batchInsertMessages } from '../utils/batchInsert.js';
import { maxSendTime } from '../utils/time.js';

type Payload = {
  inquiry: any; customer: any | null;
  messages: { messageId: string; sendTime: string; sender?: string; direction: string; content: string; translation?: string }[];
};

export async function ingest(p: Payload) {
  const prisma = getPrisma();
  const { inquiry: iq, customer: cu, messages } = p;
  const newCount = messages.length;
  const newLast = maxSendTime(messages);

  // 早返回:存在 + 无失败批次 + count + lastSendTime 均一致
  const existing = await prisma.inquiry.findUnique({ where: { inquiryId: iq.inquiryId } });
  if (existing && !existing.hasFailedBatches
      && existing.messagesCount === newCount
      && newLast && existing.lastSendTime
      && existing.lastSendTime.getTime() === newLast.getTime()) {
    return { ok: true, inserted: existing.messagesCount, failedBatches: 0, earlyReturn: true };
  }

  // 主表事务
  await prisma.$transaction(async (tx) => {
    if (cu) {
      const data = mapCustomer(cu);
      const update = skipNulls(data);   // 不用 null 覆盖已落库值 —— 修正 I6
      await tx.customer.upsert({
        where: { customerId: cu.customerId },
        create: data,
        update,
      });
    }
    await tx.inquiry.upsert({
      where: { inquiryId: iq.inquiryId },
      create: { inquiryId: iq.inquiryId, customerId: cu?.customerId ?? null, manager: iq.manager,
        createTime: new Date(iq.createTime), updateTime: new Date(iq.updateTime),
        status: iq.status, msgPreview: iq.msgPreview, isFlagged: iq.isFlagged },
      update: { customerId: cu?.customerId ?? null, manager: iq.manager,
        createTime: new Date(iq.createTime), updateTime: new Date(iq.updateTime),
        status: iq.status, msgPreview: iq.msgPreview, isFlagged: iq.isFlagged },
    });
  });

  // 批量消息(Task 8)
  const { failedBatches } = await batchInsertMessages(iq.inquiryId, messages);

  // 用"总量"(批量后该询盘的消息总数),不是 createMany 的本次 delta —— 修正 F2
  const total = await prisma.chatMessage.count({ where: { inquiryId: iq.inquiryId } });
  await prisma.inquiry.update({
    where: { inquiryId: iq.inquiryId },
    data: { messagesCount: total, lastSendTime: newLast, hasFailedBatches: failedBatches > 0 },
  });

  return { ok: true, inserted: total, failedBatches, earlyReturn: false };
}

function mapCustomer(c: any) {
  return { customerId: c.customerId, name: c.name, country: c.country, email: c.email, company: c.company,
    registerTime: c.registerTime ? new Date(c.registerTime) : null, avatarUrl: c.avatarUrl,
    buyerTags: c.buyerTags, buyerFeatures: c.buyerFeatures,
    bProductViews: c.behavior?.productViews ?? null, bValidInquiries: c.behavior?.validInquiries ?? null,
    bValidRfq: c.behavior?.validRfq ?? null, bLoginDays: c.behavior?.loginDays ?? null,
    bSpamInquiries: c.behavior?.spamInquiries ?? null, bTopIndustry: c.behavior?.topIndustry ?? null };
}
// update 时跳过值为 null 的字段(避免覆盖已落库的有效值)
function skipNulls(data: Record<string, any>) {
  const out: Record<string, any> = {};
  for (const [k, v] of Object.entries(data)) if (v != null || k === 'customerId') out[k] = v;
  return out;
}

Step 3: Commit feat(ingest): early-return + upsert (customerUpdate skips nulls)

Task 8: 批量消息写入(单一 createMany 实现 + 增量回归测试)

Files: src/utils/batchInsert.ts;tests/ingest.test.ts 加增量测试

Step 1: src/utils/batchInsert.ts(只留 createMany skipDuplicates,删掉 raw SQL 双实现 —— 修正 I1)

import { getPrisma } from '../lib/prisma.js';
import type { Direction } from '@prisma/client';

const BATCH = 500;

export async function batchInsertMessages(inquiryId: string, messages: any[]) {
  const prisma = getPrisma();
  let failedBatches = 0;
  for (let i = 0; i < messages.length; i += BATCH) {
    const batch = messages.slice(i, i + BATCH);
    try {
      // createMany + skipDuplicates:MySQL 下生成 INSERT IGNORE,重复 messageId 跳过。
      // MVP 接受"重推时已存在的消息不更新 content"(spec R6 背书);需要更新 content 再换 raw ON DUPLICATE KEY。
      await prisma.chatMessage.createMany({
        skipDuplicates: true,
        data: batch.map(m => ({
          messageId: m.messageId, inquiryId,
          sender: m.sender ?? null,
          direction: m.direction as Direction,
          sendTime: new Date(m.sendTime),
          content: m.content,
          translation: m.translation ?? '',
        })),
      });
    } catch (e) {
      console.error(`batch at offset ${i} failed:`, e);
      failedBatches++;
    }
  }
  return { failedBatches };
}

Step 2: tests/ingest.test.ts 加幂等 + 增量回归测试(修正 F3 —— 这个测试会逼出 F2 修复)

import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import request from 'supertest';
import { setupApp, resetDb, teardown } from './setup.js';
import { makePayload } from './fixtures.js';
let app: any; const TOKEN = 'Bearer dev-ingest-token-1234';
beforeAll(async () => { ({ app } = await setupApp()); }, 90000);
afterAll(teardown);

describe('ingest idempotency + increment', () => {
  it('push N → push N again early-returns', async () => {
    await resetDb();
    const p = makePayload();
    const r1 = await request(app).post('/api/ingest').set('Authorization', TOKEN).send(p);
    expect(r1.status).toBe(200);
    expect(r1.body.inserted).toBe(2);
    expect(r1.body.earlyReturn).toBe(false);
    const r2 = await request(app).post('/api/ingest').set('Authorization', TOKEN).send(p);
    expect(r2.body.earlyReturn).toBe(true);   // 同样 payload → 早返回
  });

  it('push N → push N+1 → push N+1 again early-returns (regression for messagesCount=total)', async () => {
    await resetDb();
    const base = makePayload();
    // 推 2 条
    await request(app).post('/api/ingest').set('Authorization', TOKEN).send(base);
    // 推 3 条(新增 m3)
    const withOneMore = makePayload({ messages: [
      ...base.messages,
      { messageId: 'm3', sendTime: '2026-07-13T02:13:00Z', sender: 'x', direction: 'incoming', content: 'c3', translation: '' },
    ]});
    const r3 = await request(app).post('/api/ingest').set('Authorization', TOKEN).send(withOneMore);
    expect(r3.body.inserted).toBe(3);          // 总量 3,不是 delta 1
    expect(r3.body.earlyReturn).toBe(false);
    // 再推同样 3 条 → 早返回(若 messagesCount 退化成 delta,这里会失败)
    const r4 = await request(app).post('/api/ingest').set('Authorization', TOKEN).send(withOneMore);
    expect(r4.body.earlyReturn).toBe(true);
    expect(r4.body.inserted).toBe(3);
  });
});

Step 3: 跑测试

npm test -- ingest

Expected:增量回归测试通过(证明 messagesCount 用 total)。

Step 4: Commit feat(ingest): batched createMany skipDuplicates + idempotency/increment regression tests

Task 9: 询盘列表(筛选/搜索/排序/分页)+ buildInquiryWhere 复用

Files: src/services/inquiry.service.ts, src/routes/inquiries.ts;Modify src/app.ts

Step 1: src/services/inquiry.service.ts(buildInquiryWhere 导出复用 —— S6;显式 equals —— I10)

import { Prisma } from '@prisma/client';
import { getPrisma } from '../lib/prisma.js';

export function buildInquiryWhere(q: any): Prisma.InquiryWhereInput {
  const where: Prisma.InquiryWhereInput = {};
  if (q.country) where.customer = { is: { country: { equals: q.country } } };
  if (q.manager) where.manager = { equals: q.manager };
  if (q.status) where.status = { equals: q.status };
  if (q.dateFrom || q.dateTo) {
    where.createTime = {};
    if (q.dateFrom) where.createTime.gte = new Date(q.dateFrom);
    if (q.dateTo) where.createTime.lte = new Date(q.dateTo);
  }
  if (q.q) where.OR = [
    { inquiryId: { contains: q.q } },
    { msgPreview: { contains: q.q } },
    { customer: { is: { name: { contains: q.q } } } },
    { customer: { is: { company: { contains: q.q } } } },
  ];
  return where;
}

export async function listInquiries(q: any) {
  const prisma = getPrisma();
  const size = Math.min(Number(q.size) || 20, 100);
  const page = Number(q.page) || 1;
  const where = buildInquiryWhere(q);
  const orderBy = q.sort === 'update_time_desc' ? { updateTime: 'desc' as const }
                : q.sort === 'create_time_asc' ? { createTime: 'asc' as const }
                : { createTime: 'desc' as const };
  const [items, total] = await Promise.all([
    prisma.inquiry.findMany({ where, orderBy, skip: (page - 1) * size, take: size, include: { customer: true } }),
    prisma.inquiry.count({ where }),
  ]);
  return { items, total, page, size };
}

Step 2: src/routes/inquiries.ts(骨架,本 Task 只挂 list)

import { Router } from 'express';
import { jwtAuth } from '../middleware/jwtAuth.js';
import { listInquiries, getInquiryDetail, listMessages } from '../services/inquiry.service.js';
export const inquiriesRouter = Router();
inquiriesRouter.get('/', jwtAuth, async (req, res, next) => {
  try { res.json(await listInquiries(req.query)); } catch (e) { next(e); }
});

Step 3: src/app.ts 挂 app.use('/api/inquiries', inquiriesRouter)
Step 4: 测试 tests/inquiries.test.ts(list 筛选;seedInquiries fixture 自行构造或复用 makePayload 推几条)

it('lists filtered by country', async () => {
  await resetDb();
  await pushInquiry(app, makePayload({ customer: { country: 'PE' } }));   // helper: POST /api/ingest
  await pushInquiry(app, makePayload({ inquiry: { inquiryId: 'X2' }, customer: { country: 'US' } }));
  const token = (await request(app).post('/api/auth/login').send({ username: 'admin', password: 'admin-password-1234' })).body.token;
  const res = await request(app).get('/api/inquiries?country=PE').set('Authorization', 'Bearer ' + token);
  expect(res.status).toBe(200);
  expect(res.body.items.length).toBe(1);
  expect(res.body.items[0].customer.country).toBe('PE');
});

Step 5: 测试通过 + Commit feat(api): inquiry list with filters/search/pagination, shared buildInquiryWhere

Task 10: 询盘详情 API

Files: Modify src/services/inquiry.service.ts, src/routes/inquiries.ts

Step 1: 服务方法

export async function getInquiryDetail(inquiryId: string) {
  return getPrisma().inquiry.findUnique({
    where: { inquiryId }, include: { customer: true, _count: { select: { messages: true } } },
  });
}

Step 2: 路由 inquiriesRouter.get('/:id', jwtAuth, ...) → res.json(await getInquiryDetail(req.params.id))
Step 3: 测试(推一条 → GET /:id → 200 含 customer + _count.messages) + Commit feat(api): inquiry detail

Task 11: 聊天游标分页(复合游标 + zod 校验)

Files: Modify src/services/inquiry.service.ts, src/routes/inquiries.ts

Step 1: 服务方法((sendTime, messageId) 复合游标)

export async function listMessages(inquiryId: string, q: { afterSendTime?: string; afterMessageId?: string; size?: string }) {
  const prisma = getPrisma();
  const size = Math.min(Number(q.size) || 50, 200);
  const where: Prisma.ChatMessageWhereInput = { inquiryId };
  if (q.afterSendTime && q.afterMessageId) {
    const after = new Date(q.afterSendTime);
    where.OR = [{ sendTime: { gt: after } }, { sendTime: after, messageId: { gt: q.afterMessageId } }];
  }
  const items = await prisma.chatMessage.findMany({
    where, orderBy: [{ sendTime: 'asc' }, { messageId: 'asc' }], take: size,
  });
  const last = items[items.length - 1];
  return { items, nextCursor: items.length === size && last
    ? { afterSendTime: last.sendTime.toISOString(), afterMessageId: last.messageId } : null };
}

Step 2: 路由 + zod query 校验(半残游标 400 —— 修正 I3)

const CursorQuery = z.object({
  afterSendTime: z.string().datetime().optional(),
  afterMessageId: z.string().optional(),
  size: z.string().optional(),
}).refine(d => (!!d.afterSendTime && !!d.afterMessageId) || (!d.afterSendTime && !d.afterMessageId),
  { message: 'cursor fields must be both present or both absent' });

inquiriesRouter.get('/:id/messages', jwtAuth, async (req, res, next) => {
  const parsed = CursorQuery.safeParse(req.query);
  if (!parsed.success) return res.status(400).json({ error: parsed.error.flatten() });
  try { res.json(await listMessages(req.params.id, parsed.data)); } catch (e) { next(e); }
});

Step 3: 测试(插 50 条 → 首页 size=20 → 带 nextCursor 取下页;另测只传 afterSendTime → 400) + Commit feat(api): chat cursor pagination (composite key, zod-validated)

Task 12: 统计 API

Files: src/services/stats.service.ts, src/routes/stats.ts;Modify src/app.ts

Step 1: 服务(_count 重命名为 count —— S5)

import { getPrisma } from '../lib/prisma.js';
export async function overview() {
  const prisma = getPrisma();
  const total = await prisma.inquiry.count();
  const countryTop10 = (await prisma.customer.groupBy({
    by: ['country'], _count: { id: true }, orderBy: { _count: { id: 'desc' } }, take: 10, where: { country: { not: null } },
  })).map(r => ({ country: r.country, count: r._count.id }));
  const byManager = (await prisma.inquiry.groupBy({
    by: ['manager'], _count: { id: true }, orderBy: { _count: { id: 'desc' } }, where: { manager: { not: null } },
  })).map(r => ({ manager: r.manager, count: r._count.id }));
  const monthly = await prisma.$queryRaw<{ ym: string; c: bigint }[]>`
    SELECT DATE_FORMAT(create_time, '%Y-%m') ym, COUNT(*) c FROM inquiries GROUP BY ym ORDER BY ym DESC LIMIT 12`;
  return { total, countryTop10, byManager, monthly: monthly.map(r => ({ ym: r.ym, count: Number(r.c) })) };
}

Step 2: 路由 /api/stats/overview(jwtAuth)+ 挂载
Step 3: 测试 + Commit feat(api): stats overview

Task 13: 导出(列表 CSV + 单询盘聊天 CSV)

Files: src/routes/export.ts, Modify src/routes/inquiries.ts(单询盘导出);src/app.ts

Step 1: src/routes/export.ts(列表 CSV stream,复用 buildInquiryWhere —— S6)

import { Router } from 'express';
import { getPrisma } from '../lib/prisma.js';
import { jwtAuth } from '../middleware/jwtAuth.js';
import { buildInquiryWhere } from '../services/inquiry.service.js';
export const exportRouter = Router();
exportRouter.get('/', jwtAuth, async (req, res, next) => {
  try {
    res.setHeader('Content-Type', 'text/csv; charset=utf-8');
    res.setHeader('Content-Disposition', 'attachment; filename="inquiries.csv"');
    res.write('询价单号,业务员,国家,买家,公司,状态,创建时间,消息数\n');
    const rows = await getPrisma().inquiry.findMany({
      where: buildInquiryWhere(req.query), include: { customer: true }, orderBy: { createTime: 'desc' } });
    for (const r of rows) {
      res.write([r.inquiryId, r.manager, r.customer?.country, r.customer?.name, r.customer?.company,
        r.status, r.createTime.toISOString(), r.messagesCount].map(csvEscape).join(',') + '\n');
    }
    res.end();
  } catch (e) { next(e); }
});
function csvEscape(v: any) { const s = v == null ? '' : String(v); return /[",\n]/.test(s) ? '"' + s.replace(/"/g, '""') + '"' : s; }

Step 2: 单询盘聊天导出(挂到 inquiriesRouter —— 修正 I5)

// src/routes/inquiries.ts
inquiriesRouter.get('/:id/messages/export', jwtAuth, async (req, res, next) => {
  try {
    res.setHeader('Content-Type', 'text/csv; charset=utf-8');
    res.setHeader('Content-Disposition', `attachment; filename="chat_${req.params.id}.csv"`);
    res.write('时间,方向,发送者,原文,翻译\n');
    const msgs = await getPrisma().chatMessage.findMany({
      where: { inquiryId: req.params.id }, orderBy: [{ sendTime: 'asc' }, { messageId: 'asc' }] });
    for (const m of msgs) {
      res.write([m.sendTime.toISOString(), m.direction, m.sender, m.content, m.translation ?? ''].map(csvEscape).join(',') + '\n');
    }
    res.end();
  } catch (e) { next(e); }
});

Step 3: 挂 /api/export(exportRouter)+ 测试(列表导出 + 单询盘导出,验证 CSV 含表头且不含/含聊天) + Commit feat(api): csv export (list + single-inquiry chat)

Task 14: Docker 开发环境 + 端到端 e2e

Files: Dockerfile, docker-compose.yml, scripts/e2e.sh

Step 1: Dockerfile(多阶段;prisma generate 在 build 阶段;运行时 migrate deploy)

FROM node:18-bookworm-slim AS build
WORKDIR /app
COPY package*.json ./ && RUN npm ci
COPY . . && RUN npx prisma generate && npm run build
FROM node:18-bookworm-slim
WORKDIR /app
COPY --from=build /app/node_modules ./node_modules
COPY --from=build /app/dist ./dist
COPY --from=build /app/prisma ./prisma
COPY --from=build /app/package.json .
EXPOSE 8080
CMD ["sh","-c","npx prisma migrate deploy && node dist/server.js"]

Step 2: docker-compose.yml(mysql --max_allowed_packet=128M —— spec v3)

services:
  mysql:
    image: mysql:8
    environment: { MYSQL_ROOT_PASSWORD: root, MYSQL_DATABASE: inquiry }
    command: --max_allowed_packet=128M --character-set-server=utf8mb4
    volumes: [mysql_data:/var/lib/mysql]
    ports: ["3306:3306"]
    healthcheck: { test: ["CMD","mysqladmin","ping","-h","localhost","-uroot","-proot"], interval: 5s, retries: 20 }
  backend:
    build: .
    environment:
      DATABASE_URL: mysql://root:root@mysql:3306/inquiry
      INGEST_TOKEN: dev-ingest-token-1234
      JWT_SECRET: dev-jwt-secret-1234
      ADMIN_USERNAME: admin
      ADMIN_PASSWORD: admin-password-1234
      NODE_ENV: production
      PORT: "8080"
    depends_on: { mysql: { condition: service_healthy } }
    ports: ["8080:8080"]
volumes: { mysql_data: {} }

Step 3: scripts/e2e.sh(补全 detail/messages/stats/export 断言 —— S7)

#!/usr/bin/env bash
set -e
BASE=http://localhost:8080; ITOK=dev-ingest-token-1234
JWT=$(curl -s -XPOST $BASE/api/auth/login -H 'Content-Type: application/json' \
  -d '{"username":"admin","password":"admin-password-1234"}' | node -e "let s='';process.stdin.on('data',d=>s+=d).on('end',()=>console.log(JSON.parse(s).token))")
echo "auth ok"
curl -sf -XPOST $BASE/api/ingest -H "Authorization: Bearer $ITOK" -H 'Content-Type: application/json' \
  -d '{"inquiry":{"inquiryId":"E2E-1","createTime":"2026-07-13T10:00:00Z","updateTime":"2026-07-13T10:00:00Z","manager":"tester","status":"new"},"customer":{"customerId":"c-e2e","name":"Test Buyer","country":"US"},"messages":[{"messageId":"e2e-m1","sendTime":"2026-07-13T10:00:00Z","direction":"outgoing","content":"hello e2e"}]}' >/dev/null
echo "ingest ok"
curl -sf "$BASE/api/inquiries/E2E-1" -H "Authorization: Bearer $JWT" | grep -q E2E-1 && echo "detail ok"
curl -sf "$BASE/api/inquiries/E2E-1/messages" -H "Authorization: Bearer $JWT" | grep -q e2e-m1 && echo "messages ok"
curl -sf "$BASE/api/stats/overview" -H "Authorization: Bearer $JWT" | grep -q total && echo "stats ok"
curl -sf "$BASE/api/export" -H "Authorization: Bearer $JWT" | grep -q E2E-1 && echo "export ok"

Step 4: 起服务跑 e2e

docker compose up -d --build
sleep 30; bash scripts/e2e.sh

Expected:auth/ingest/detail/messages/stats/export ok 全部打印。

Step 5: 跑全量测试

npm test

Expected:全部 pass。

Step 6: Commit feat: docker dev env + e2e; Plan 1 v2 complete

完成标准(DoD)

4 张表按 schema 建好(has_failed_batches、MEDIUMTEXT content、Direction 含 system、复合索引)
/api/ingest:zod 校验 + token 鉴权(timingSafeEqual)+ 限流 + 早返回(!hasFailedBatches + count + lastSendTime) + 主表 upsert(customerUpdate 不覆盖 null)+ 批量消息(createMany skipDuplicates 分批 500、失败隔离)+ messagesCount 用总量
/api/auth/login + /refresh:env bootstrap admin + JWT
/api/inquiries(列表筛选/搜索/分页)+ /:id(详情)+ /:id/messages(复合游标分页,zod 校验)+ /:id/messages/export(单询盘聊天 CSV)
/api/stats/overview(_count → count 重命名)
/api/export(列表 CSV stream,不含聊天)
增量幂等回归测试:推 N → 推 N+1 → 再推 N+1 必须早返回(S11)
docker compose up 一键起,e2e 六步全 ok
全量 vitest 通过(惰性 Prisma + testcontainers + singleFork)

后续 Plan(本 plan 不含)

Plan 2:插件推送改造(配置页 + IndexedDB 队列经 SW 落盘 + chrome.alarms worker + optional_host_permissions + apiMsgToChat 字段扩展)
Plan 3:管理后台前端(Vue3 + Element Plus:登录/列表/详情虚拟滚动聊天/统计/导出;CORS 收窄)
Plan 4:正式部署(域名 + Let's Encrypt + 备份 cron + 健康监控 + 限流切 token 维度)

46 KiB Raw Blame History