From ead1296f80de3cff81eff4d0fe40e883ff378b06 Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 29 Jun 2026 04:35:19 +0800 Subject: [PATCH] feat: add crm webhook worker --- .env.example | 9 + README.md | 12 +- apps/worker/package.json | 22 + apps/worker/src/config.ts | 25 + apps/worker/src/db.ts | 11 + apps/worker/src/index.ts | 50 ++ apps/worker/src/jobs/crm.ts | 528 ++++++++++++++++++ apps/worker/tsconfig.json | 14 + docs/refactor/README.md | 2 +- docs/refactor/architecture.md | 4 +- docs/refactor/backend-capability-status.md | 3 +- docs/refactor/backend-handoff-roadmap.md | 2 +- docs/refactor/backend-progress.md | 6 +- docs/refactor/blueprint-coverage.md | 2 +- docs/refactor/crm-worker.md | 107 ++++ docs/refactor/implementation-status.md | 2 +- docs/refactor/next-development-todo.md | 5 +- docs/refactor/project-structure.md | 10 + package-lock.json | 34 ++ package.json | 5 +- scripts/crm-worker-integration-test.js | 230 ++++++++ scripts/smoke-seed.js | 13 + .../202606290010_crm_worker_hardening.sql | 16 + 23 files changed, 1098 insertions(+), 14 deletions(-) create mode 100644 apps/worker/package.json create mode 100644 apps/worker/src/config.ts create mode 100644 apps/worker/src/db.ts create mode 100644 apps/worker/src/index.ts create mode 100644 apps/worker/src/jobs/crm.ts create mode 100644 apps/worker/tsconfig.json create mode 100644 docs/refactor/crm-worker.md create mode 100644 scripts/crm-worker-integration-test.js create mode 100644 supabase/migrations/202606290010_crm_worker_hardening.sql diff --git a/.env.example b/.env.example index 2b67cec7..17680e9b 100644 --- a/.env.example +++ b/.env.example @@ -46,3 +46,12 @@ ALLOW_PLATFORM_ADMIN_KEY=false # 迁移期平台管理 API Key。生产环境必须使用平台管理员 JWT/服务端会话，不能开启 ALLOW_PLATFORM_ADMIN_KEY。 PLATFORM_ADMIN_API_KEY=replace_with_platform_admin_key + +# Worker 配置：CRM webhook、支付补偿、导入复检等后台任务使用 +WORKER_CRM_BATCH_SIZE=20 +WORKER_CRM_POLL_INTERVAL_MS=10000 +WORKER_CRM_MAX_ATTEMPTS=5 +WORKER_CRM_BACKOFF_SECONDS=5,30,120,600,1800 +WORKER_CRM_REQUEST_TIMEOUT_MS=10000 +# 仅本地 fake webhook 测试允许 http://127.0.0.1；生产建议 false +WORKER_CRM_ALLOW_INSECURE_LOCALHOST=false diff --git a/README.md b/README.md index a670d49d..fb177fae 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,7 @@ - 学生端能力：题库入口、分类树、题目集合、顺序/随机/模考 session 组卷快照、答题、错题本、收藏夹、背单词进度、个人中心、考试倒计时、签到积分、题目反馈、排行榜、分数线、题目视频、订单详情/状态轮询、优惠券领取/抵扣、权益、激活码预检查/兑换、资料下载。 - 平台后台能力：租户管理、SaaS 套餐、订阅、账单、服务费收款、用量记录。 - 销售/代理/CRM 增长链路：邀请码、扫码/分享事件、首绑客资保护、销售统计、团队关系、CRM 配置和队列。 +- `apps/worker` 后台任务进程：CRM webhook 队列消费、generic/钉钉/飞书/企微机器人发送、签名、失败重试和日志。 - 销售/代理分佣结算基础闭环：租户默认比例、成员比例、激活码批次比例、订单/激活码归因、结算单生成、审核、线下打款状态和权限隔离。 - PocketBase schema/数据导入器雏形和导入后校验脚本。 - 本地 Supabase reset、烟测 seed、API 集成测试、完整重构检查命令。 @@ -27,7 +28,7 @@ - 阿里云/腾讯云短信、微信小程序登录、微信支付、支付宝主链路已完成本地适配；微信网页登录、QQ 登录、手机号换绑、退款/对账、支付补偿和真实生产账号联调还没接完。 - OSS/COS/Supabase Storage 上传下载签名 provider 已接入；上传后校验、PDF 预览、防盗链和视频水印还没完成。 - Excel/CSV 导入、分数线/视频批量导入和异步 worker 还没完成。 -- 分佣真实打款、结算导出、发票/凭证、CRM webhook worker 和销售转化看板还没完成。 +- 分佣真实打款、结算导出、发票/凭证、CRM 轮询/定向分配、富卡片模板、失败告警和销售转化看板还没完成。 - Taro 跨端前端还没开始 scaffold。 - 根目录已清理为新 Supabase SaaS monorepo 编排层；旧 PocketBase/React 项目和旧构建产物仅保留在 `参考/` 目录作为迁移参考，不进入 Git 提交。 @@ -54,6 +55,7 @@ ```text apps/api/ Node.js 业务 API +apps/worker/ 后台异步任务：CRM webhook、后续支付补偿/导入复检等 packages/config/ 共享配置 packages/db/ PostgreSQL 连接池和查询封装 packages/domain/ 领域常量和共享类型 @@ -86,6 +88,12 @@ npm run db:smoke-seed npm run dev:api ``` +单次运行 CRM worker： + +```bash +npm --workspace @tiku-saas/worker run crm:once +``` + 默认本地数据库： ```text @@ -119,9 +127,11 @@ npm run check:refactor ```bash npm run check:api +npm run check:worker npm run check:importer npm run pb:import:validate npm run test:api +npm run test:worker:crm ``` ## API 模块 diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 00000000..206d4914 --- /dev/null +++ b/apps/worker/package.json @@ -0,0 +1,22 @@ +{ + "name": "@tiku-saas/worker", + "version": "0.1.0", + "private": true, + "type": "module", + "scripts": { + "dev": "tsx watch src/index.ts --loop", + "start": "node dist/apps/worker/src/index.js --loop", + "build": "node -e \"fs.rmSync('dist',{recursive:true,force:true})\" && tsc -p tsconfig.json", + "check": "tsc -p tsconfig.json --noEmit", + "crm:once": "tsx src/index.ts --once --job crm" + }, + "dependencies": { + "pg": "^8.16.3" + }, + "devDependencies": { + "@types/node": "^24.0.4", + "@types/pg": "^8.15.4", + "tsx": "^4.20.3", + "typescript": "^5.8.3" + } +} diff --git a/apps/worker/src/config.ts b/apps/worker/src/config.ts new file mode 100644 index 00000000..b2bba513 --- /dev/null +++ b/apps/worker/src/config.ts @@ -0,0 +1,25 @@ +import { DEFAULT_DATABASE_URL, envBoolean, envList, envNumber, envString, loadDotenv } from '../../../packages/config/src/index.js'; + +loadDotenv(); + +export interface WorkerConfig { + databaseUrl: string; + crmBatchSize: number; + crmPollIntervalMs: number; + crmMaxAttempts: number; + crmBackoffSeconds: number[]; + crmRequestTimeoutMs: number; + crmAllowInsecureLocalhost: boolean; +} + +export const config: WorkerConfig = { + databaseUrl: envString('DATABASE_URL', DEFAULT_DATABASE_URL), + crmBatchSize: envNumber('WORKER_CRM_BATCH_SIZE', 20), + crmPollIntervalMs: envNumber('WORKER_CRM_POLL_INTERVAL_MS', 10_000), + crmMaxAttempts: envNumber('WORKER_CRM_MAX_ATTEMPTS', 5), + crmBackoffSeconds: envList('WORKER_CRM_BACKOFF_SECONDS', '5,30,120,600,1800') + .map((value: string) => Number(value)) + .filter((value: number) => Number.isFinite(value) && value > 0), + crmRequestTimeoutMs: envNumber('WORKER_CRM_REQUEST_TIMEOUT_MS', 10_000), + crmAllowInsecureLocalhost: envBoolean('WORKER_CRM_ALLOW_INSECURE_LOCALHOST', false), +}; diff --git a/apps/worker/src/db.ts b/apps/worker/src/db.ts new file mode 100644 index 00000000..882bdb59 --- /dev/null +++ b/apps/worker/src/db.ts @@ -0,0 +1,11 @@ +import { createPool } from '../../../packages/db/src/index.js'; +import { config } from './config.js'; + +export const pool = createPool({ + connectionString: config.databaseUrl, + max: 5, +}); + +export async function closePool() { + await pool.end(); +} diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts new file mode 100644 index 00000000..2e3228a9 --- /dev/null +++ b/apps/worker/src/index.ts @@ -0,0 +1,50 @@ +import { closePool } from './db.js'; +import { config } from './config.js'; +import { processCrmBatch } from './jobs/crm.js'; + +function hasArg(name: string) { + return process.argv.includes(name); +} + +function argValue(name: string, fallback = '') { + const index = process.argv.indexOf(name); + return index >= 0 ? process.argv[index + 1] || fallback : fallback; +} + +async function runOnce() { + const job = argValue('--job', 'crm'); + if (job !== 'crm') { + throw new Error(`Unsupported worker job: ${job}`); + } + const result = await processCrmBatch(); + console.log(`[worker] crm batch processed=${result.processed} sent=${result.sent} failed=${result.failed} retrying=${result.retrying} discarded=${result.discarded}`); +} + +async function runLoop() { + console.log('[worker] started'); + let stopped = false; + const stop = () => { + stopped = true; + }; + process.once('SIGINT', stop); + process.once('SIGTERM', stop); + + while (!stopped) { + try { + await runOnce(); + } catch (error) { + console.error('[worker] job failed', error); + } + await new Promise(resolve => setTimeout(resolve, config.crmPollIntervalMs)); + } +} + +try { + if (hasArg('--loop')) { + await runLoop(); + } else { + await runOnce(); + } +} finally { + await closePool(); +} diff --git a/apps/worker/src/jobs/crm.ts b/apps/worker/src/jobs/crm.ts new file mode 100644 index 00000000..d321befe --- /dev/null +++ b/apps/worker/src/jobs/crm.ts @@ -0,0 +1,528 @@ +import crypto from 'node:crypto'; +import type pg from 'pg'; +import { pool } from '../db.js'; +import { config } from '../config.js'; + +const PROVIDERS = ['generic', 'dingtalk', 'feishu', 'wecom'] as const; +type CrmProvider = typeof PROVIDERS[number]; + +interface CrmQueueRow { + id: string; + tenantId: string; + recordId: string | null; + status: string; + attempts: number; + leadId: string | null; + source: string | null; + payload: Record; + targetUrl: string | null; +} + +interface CrmConfigRow { + tenantId: string; + enabled: boolean; + url: string | null; + secretRef: string | null; + formName: string | null; + examType: string | null; + timeoutSec: number | null; +} + +interface SecretRow { + secretValue: string | null; + secretJson: Record | null; +} + +interface PreparedRequest { + provider: CrmProvider; + url: string; + body: Record; + headers: Record; +} + +interface SendResult { + ok: boolean; + httpCode: number; + responseSummary: string; + errorMessage: string; +} + +interface ProcessResult { + processed: number; + sent: number; + failed: number; + retrying: number; + discarded: number; +} + +function objectValue(value: unknown): Record { + return value && typeof value === 'object' && !Array.isArray(value) ? value as Record : {}; +} + +function stringValue(value: unknown, fallback = '') { + return typeof value === 'string' && value.trim() ? value.trim() : fallback; +} + +function numberValue(value: unknown, fallback = 0) { + const parsed = Number(value ?? fallback); + return Number.isFinite(parsed) ? parsed : fallback; +} + +function providerFrom(configRow: CrmConfigRow | null, queueRow: CrmQueueRow): CrmProvider { + const payload = objectValue(queueRow.payload); + const payloadProvider = stringValue(payload.provider); + if ((PROVIDERS as readonly string[]).includes(payloadProvider)) return payloadProvider as CrmProvider; + return providerFromUrl(queueRow.targetUrl || configRow?.url || ''); +} + +function providerFromUrl(url: string) { + const lower = url.toLowerCase(); + if (lower.includes('dingtalk.com')) return 'dingtalk'; + if (lower.includes('feishu.cn') || lower.includes('larksuite.com')) return 'feishu'; + if (lower.includes('qyapi.weixin.qq.com') || lower.includes('work.weixin.qq.com')) return 'wecom'; + return 'generic'; +} + +function parseSecretRef(ref: string | null) { + if (!ref) return null; + const parts = ref.split(':'); + if (parts.length !== 3 || parts[0] !== 'app_private.tenant_secrets') return null; + return { scope: parts[1], key: parts[2] }; +} + +function secretText(secret: SecretRow | null, keys: string[]) { + if (!secret) return ''; + if (secret.secretValue?.trim()) return secret.secretValue.trim(); + const json = objectValue(secret.secretJson); + for (const key of keys) { + const value = json[key]; + if (typeof value === 'string' && value.trim()) return value.trim(); + } + return ''; +} + +function truncate(value: unknown, max = 1900) { + return String(value ?? '').slice(0, max); +} + +function validateWebhookUrl(rawUrl: string) { + let url: URL; + try { + url = new URL(rawUrl); + } catch { + throw new Error('CRM webhook URL is invalid'); + } + const isLocalhost = ['127.0.0.1', 'localhost', '::1'].includes(url.hostname); + if (url.protocol !== 'https:' && !(config.crmAllowInsecureLocalhost && isLocalhost)) { + throw new Error('CRM webhook URL must use HTTPS outside local development'); + } + url.username = ''; + url.password = ''; + return url; +} + +function compactLeadPayload(task: CrmQueueRow, configRow: CrmConfigRow | null) { + const payload = objectValue(task.payload); + const student = objectValue(payload.student); + const referral = objectValue(payload.referral); + return { + formName: stringValue(payload.formName, configRow?.formName || '刷题题库'), + examType: stringValue(payload.examType, configRow?.examType || '成人本科'), + source: stringValue(payload.source, task.source || 'unknown'), + leadId: stringValue(payload.leadId, task.leadId || ''), + student: { + id: stringValue(student.id, task.recordId || ''), + name: stringValue(student.name) || stringValue(student.username) || '未填写', + phone: stringValue(student.phone), + email: stringValue(student.email), + }, + referral: { + refCode: stringValue(referral.refCode), + referrerUserId: stringValue(referral.referrerUserId), + boundAt: stringValue(referral.boundAt), + }, + metadata: objectValue(payload.metadata), + }; +} + +function leadMarkdown(task: CrmQueueRow, configRow: CrmConfigRow | null) { + const payload = compactLeadPayload(task, configRow); + const student = payload.student; + const lines = [ + `### ${payload.formName}新客资`, + `- 考试类型：${payload.examType}`, + `- 来源：${payload.source}`, + `- 学生：${student.name}`, + `- 手机：${student.phone || '未填写'}`, + `- 推广码：${payload.referral.refCode || '无'}`, + `- 归属人ID：${payload.referral.referrerUserId || '无'}`, + `- 绑定时间：${payload.referral.boundAt || '无'}`, + `- 线索ID：${payload.leadId}`, + ]; + return lines.join('\n'); +} + +function dingtalkSign(secret: string): Record { + if (!secret) return {}; + const timestamp = String(Date.now()); + const sign = crypto + .createHmac('sha256', secret) + .update(`${timestamp}\n${secret}`) + .digest('base64'); + return { timestamp, sign }; +} + +function feishuSign(secret: string): Record { + if (!secret) return {}; + const timestamp = String(Math.floor(Date.now() / 1000)); + const sign = crypto + .createHmac('sha256', `${timestamp}\n${secret}`) + .update('') + .digest('base64'); + return { timestamp, sign }; +} + +function appendQuery(url: URL, params: Record) { + for (const [key, value] of Object.entries(params)) { + if (value) url.searchParams.set(key, value); + } + return url.toString(); +} + +function prepareRequest(task: CrmQueueRow, configRow: CrmConfigRow | null, secret: SecretRow | null): PreparedRequest { + const target = validateWebhookUrl(task.targetUrl || configRow?.url || ''); + const provider = providerFrom(configRow, task); + const secretValue = secretText(secret, ['secret', 'signSecret', 'webhookSecret']); + const markdown = leadMarkdown(task, configRow); + const headers = { 'content-type': 'application/json' }; + + if (provider === 'dingtalk') { + return { + provider, + url: appendQuery(target, dingtalkSign(secretValue)), + headers, + body: { + msgtype: 'markdown', + markdown: { + title: '新客资提醒', + text: markdown, + }, + }, + }; + } + + if (provider === 'feishu') { + return { + provider, + url: target.toString(), + headers, + body: { + msg_type: 'interactive', + ...feishuSign(secretValue), + card: { + config: { wide_screen_mode: true }, + header: { title: { tag: 'plain_text', content: '新客资提醒' }, template: 'blue' }, + elements: [{ tag: 'markdown', content: markdown }], + }, + }, + }; + } + + if (provider === 'wecom') { + return { + provider, + url: target.toString(), + headers, + body: { + msgtype: 'markdown', + markdown: { content: markdown }, + }, + }; + } + + return { + provider, + url: target.toString(), + headers, + body: { + event: 'lead.created', + data: compactLeadPayload(task, configRow), + }, + }; +} + +async function loadCrmConfig(client: pg.PoolClient, tenantId: string) { + const result = await client.query( + ` + select tenant_id as "tenantId", enabled, url, secret_ref as "secretRef", + form_name as "formName", exam_type as "examType", timeout_sec as "timeoutSec" + from public.crm_config + where tenant_id = $1 + limit 1 + `, + [tenantId], + ); + return result.rows[0] || null; +} + +async function loadSecret(client: pg.PoolClient, tenantId: string, secretRef: string | null) { + const parsed = parseSecretRef(secretRef); + if (!parsed) return null; + const result = await client.query( + ` + select secret_value as "secretValue", secret_json as "secretJson" + from app_private.tenant_secrets + where tenant_id = $1 + and secret_scope = $2 + and secret_key = $3 + limit 1 + `, + [tenantId, parsed.scope, parsed.key], + ); + return result.rows[0] || null; +} + +async function claimDueTasks(client: pg.PoolClient, limit: number) { + const result = await client.query( + ` + with due as ( + select id + from public.crm_webhook_queue + where status in ('pending', 'retrying') + and coalesce(next_attempt_at, scheduled_at, created_at) <= now() + order by coalesce(next_attempt_at, scheduled_at, created_at) asc, created_at asc + limit $1 + for update skip locked + ) + update public.crm_webhook_queue q + set status = 'processing', + last_attempt_at = now(), + updated_at = now() + from due + where q.id = due.id + returning q.id, + q.tenant_id as "tenantId", + q.record_id as "recordId", + q.status, + q.attempts, + q.lead_id as "leadId", + q.source, + q.payload, + q.target_url as "targetUrl" + `, + [limit], + ); + return result.rows; +} + +async function recoverStaleProcessingTasks(client: pg.PoolClient) { + const staleMs = Math.max(config.crmRequestTimeoutMs * 3, 60_000); + await client.query( + ` + update public.crm_webhook_queue + set status = 'retrying', + next_attempt_at = now(), + last_error = coalesce(last_error, 'Recovered stale processing task'), + updated_at = now() + where status = 'processing' + and coalesce(last_attempt_at, updated_at, created_at) < now() - ($1::int * interval '1 millisecond') + `, + [staleMs], + ); +} + +async function sendWebhook(request: PreparedRequest, timeoutMs: number): Promise { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), timeoutMs); + try { + const response = await fetch(request.url, { + method: 'POST', + headers: request.headers, + body: JSON.stringify(request.body), + signal: controller.signal, + }); + const text = await response.text().catch(() => ''); + return { + ok: response.ok, + httpCode: response.status, + responseSummary: truncate(text, 1900), + errorMessage: response.ok ? '' : `CRM webhook returned HTTP ${response.status}`, + }; + } catch (error) { + return { + ok: false, + httpCode: 0, + responseSummary: '', + errorMessage: error instanceof Error ? error.message : String(error), + }; + } finally { + clearTimeout(timeout); + } +} + +async function appendLog( + client: pg.PoolClient, + task: CrmQueueRow, + request: PreparedRequest | null, + result: SendResult, + attempt: number, +) { + await client.query( + ` + insert into public.crm_webhook_log ( + tenant_id, record_id, http_code, outcome, error_message, lead_id, + request_body, request_payload, response_summary, signed_at, attempt + ) + values ($1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9, now(), $10) + `, + [ + task.tenantId, + task.recordId, + result.httpCode, + result.ok ? 'sent' : 'failed', + truncate(result.errorMessage), + task.leadId, + request ? JSON.stringify(request.body).slice(0, 19_000) : '', + JSON.stringify({ + provider: request?.provider || null, + urlHost: request ? new URL(request.url).hostname : null, + body: request?.body || null, + }), + truncate(result.responseSummary), + attempt, + ], + ); +} + +async function markTaskResult( + client: pg.PoolClient, + task: CrmQueueRow, + request: PreparedRequest | null, + result: SendResult, + attempt: number, +) { + const backoff = config.crmBackoffSeconds[Math.min(attempt - 1, config.crmBackoffSeconds.length - 1)] || 60; + if (result.ok) { + await client.query( + ` + update public.crm_webhook_queue + set status = 'sent', + attempts = $2, + provider = $3, + next_attempt_at = null, + last_error = null, + last_http_code = $4, + last_response_summary = $5, + sent_at = now(), + updated_at = now() + where id = $1 + `, + [task.id, attempt, request?.provider || null, result.httpCode, truncate(result.responseSummary)], + ); + return 'sent'; + } + + const terminal = attempt >= config.crmMaxAttempts; + await client.query( + ` + update public.crm_webhook_queue + set status = $2, + attempts = $3, + provider = $4, + next_attempt_at = case when $2 = 'retrying' then now() + ($5::int * interval '1 second') else null end, + last_error = $6, + last_http_code = $7, + last_response_summary = $8, + updated_at = now() + where id = $1 + `, + [ + task.id, + terminal ? 'failed' : 'retrying', + attempt, + request?.provider || null, + backoff, + truncate(result.errorMessage), + result.httpCode, + truncate(result.responseSummary), + ], + ); + return terminal ? 'failed' : 'retrying'; +} + +async function discardTask(client: pg.PoolClient, task: CrmQueueRow, message: string) { + await client.query( + ` + update public.crm_webhook_queue + set status = 'discarded', + attempts = attempts + 1, + last_attempt_at = now(), + last_error = $2, + updated_at = now() + where id = $1 + `, + [task.id, truncate(message)], + ); + await appendLog(client, task, null, { + ok: false, + httpCode: 0, + responseSummary: '', + errorMessage: message, + }, task.attempts + 1); +} + +async function processTask(task: CrmQueueRow) { + const client = await pool.connect(); + try { + const crmConfig = await loadCrmConfig(client, task.tenantId); + if (!crmConfig?.enabled || !(task.targetUrl || crmConfig.url)) { + await discardTask(client, task, 'CRM config is disabled or missing URL'); + return 'discarded'; + } + const secret = await loadSecret(client, task.tenantId, crmConfig.secretRef); + const request = prepareRequest(task, crmConfig, secret); + const timeoutMs = Math.max(1000, (crmConfig.timeoutSec || 0) * 1000 || config.crmRequestTimeoutMs); + const attempt = task.attempts + 1; + const result = await sendWebhook(request, timeoutMs); + await appendLog(client, task, request, result, attempt); + return markTaskResult(client, task, request, result, attempt); + } catch (error) { + const attempt = task.attempts + 1; + const result: SendResult = { + ok: false, + httpCode: 0, + responseSummary: '', + errorMessage: error instanceof Error ? error.message : String(error), + }; + await appendLog(client, task, null, result, attempt); + return markTaskResult(client, task, null, result, attempt); + } finally { + client.release(); + } +} + +export async function processCrmBatch(limit = config.crmBatchSize): Promise { + const client = await pool.connect(); + let tasks: CrmQueueRow[] = []; + try { + await client.query('begin'); + await recoverStaleProcessingTasks(client); + tasks = await claimDueTasks(client, limit); + await client.query('commit'); + } catch (error) { + await client.query('rollback'); + throw error; + } finally { + client.release(); + } + + const result: ProcessResult = { processed: tasks.length, sent: 0, failed: 0, retrying: 0, discarded: 0 }; + for (const task of tasks) { + const status = await processTask(task); + if (status === 'sent') result.sent += 1; + if (status === 'failed') result.failed += 1; + if (status === 'retrying') result.retrying += 1; + if (status === 'discarded') result.discarded += 1; + } + return result; +} diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json new file mode 100644 index 00000000..d8eb659e --- /dev/null +++ b/apps/worker/tsconfig.json @@ -0,0 +1,14 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "NodeNext", + "moduleResolution": "NodeNext", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "outDir": "dist", + "rootDir": "../..", + "types": ["node"] + }, + "include": ["src/**/*.ts", "../../packages/**/*.ts"] +} diff --git a/docs/refactor/README.md b/docs/refactor/README.md index bc0e2c32..317dca41 100644 --- a/docs/refactor/README.md +++ b/docs/refactor/README.md @@ -35,4 +35,4 @@ 2. 新建 `apps/taro`，按 `docs/refactor/taro-frontend-integration.md` 优先接租户解析、首页、题库练习、背单词、知识手册、个人中心。 3. 导出 PocketBase 真实数据到 `pb_export/*.json`，执行 `npm run pb:import:json` 和 `npm run pb:import:validate`。 4. 为对象存储、分数线、视频、Excel/CSV 补齐 provider/导入能力，并复用 `content_import_jobs` 管线。 -5. 接真实短信、微信/QQ 登录、微信支付/支付宝和 CRM worker，进入商用验收。 +5. 接真实短信、微信/QQ 登录、微信支付/支付宝，继续增强 CRM worker 和其它异步任务，进入商用验收。 diff --git a/docs/refactor/architecture.md b/docs/refactor/architecture.md index 27564999..3a8ca9a6 100644 --- a/docs/refactor/architecture.md +++ b/docs/refactor/architecture.md @@ -92,7 +92,7 @@ npm run pb:import:validate ## 下一阶段拆分 -- `apps/api/src/features` 继续按业务域扩展：真实支付 provider、真实 OAuth provider、平台审计和 worker。 +- `apps/api/src/features` 继续按业务域扩展：退款对账、真实 OAuth provider、平台审计和更多后台任务。 - `src/services/supabaseApi.ts` 逐页替换旧 PB 只读接口，优先学生端和小程序共用页面。 -- 新增 `apps/worker` 承接 CRM webhook、支付补偿、日报统计、导入后异步检查。 +- 扩展 `apps/worker`：CRM webhook 已落地，后续继续承接支付补偿、日报统计、导入后异步检查和公共题库同步。 - 新增 `apps/taro` 后，Auth/JWT 优先复用 Supabase client；复杂业务命令复用 `apps/api`/RPC/Edge Functions，不单独维护另一套后端逻辑。 diff --git a/docs/refactor/backend-capability-status.md b/docs/refactor/backend-capability-status.md index 05b79f91..17519454 100644 --- a/docs/refactor/backend-capability-status.md +++ b/docs/refactor/backend-capability-status.md @@ -140,7 +140,8 @@ | 手工补绑 | 可联调 | 需要 `referral:write` | | 销售统计/客户列表/团队 | 可联调 | `/api/referral/sales-*`、`team` | | CRM 配置/队列 | 可联调 | `/api/crm/config`、`/api/crm/queue` | -| 真实 CRM webhook worker | 待补齐 | 钉钉/飞书/企微发送、签名、重试、死信 | +| CRM webhook worker | 可联调 | `apps/worker` 已支持 generic webhook、钉钉、飞书、企微群机器人消息体/签名、到期任务消费、失败退避重试、最终失败、discarded 和 `crm_webhook_log` | +| CRM 增强 | 待补齐 | 轮询/定向分配策略、富卡片模板、失败告警、死信运营后台和批量 CRM 推送 | | 分佣结算基础闭环 | 可联调 | `/api/commission/settings`、`member-rate`、`summary`、`orders`、`settlements`、`settlements/generate`、`settlements/status`；支持订单/激活码归因、批次/成员/默认比例优先级、北京时间账期、结算单生成、审核、打款状态、已打款锁定、销售/代理本人范围和租户隔离 | | 分佣打款增强 | 待补齐 | 银行/微信/支付宝真实打款、结算导出、发票/凭证、财务复核和分佣看板 | diff --git a/docs/refactor/backend-handoff-roadmap.md b/docs/refactor/backend-handoff-roadmap.md index 939bfb3f..eb9bac6d 100644 --- a/docs/refactor/backend-handoff-roadmap.md +++ b/docs/refactor/backend-handoff-roadmap.md @@ -95,7 +95,7 @@ - 租户自定义角色模板基础 API 已完成；继续补权限配置 UI、班级/教师/学生范围权限和平台级审计报表。 - 三套默认主题、租户主题预览、Logo/图标/分享图配置。 -- CRM worker：钉钉、飞书、企微机器人，轮询/定向分配，失败重试。 +- CRM worker：钉钉、飞书、企微机器人发送、签名、失败重试已落地；继续补轮询/定向分配、富卡片、失败告警和死信运营台。 - 销售/代理分佣结算、销售团队看板、客资跟进效果。 - AI 择校推荐：地区考试数据、学生输入 schema、AI JSON 输出、PDF 报告生成。 - 性能压测、慢 SQL 审查、备份恢复演练、灰度发布和回滚预案。 diff --git a/docs/refactor/backend-progress.md b/docs/refactor/backend-progress.md index cefc38d6..c1d04214 100644 --- a/docs/refactor/backend-progress.md +++ b/docs/refactor/backend-progress.md @@ -28,6 +28,7 @@ - 已新增 `npm run db:smoke-seed`，用于 `supabase:reset` 后恢复最小烟测数据。 - 已新增 `npm run smoke:core-api`，用于验证个人中心、分数线、题目视频、背单词进度/收藏等学生端核心 API。 - 已新增 `npm run test:api`，自动 seed、构建、启动临时 API，并断言核心学生端接口、内容导航/组卷、租户隔离、资源权限和题目导入。 +- 已新增 `apps/worker` 和 `npm run test:worker:crm`，用于消费 CRM webhook 队列，验证本地 fake webhook、队列状态、日志和密钥不泄露。 ## 已验证接口 @@ -233,7 +234,7 @@ GET /api/tenant-admin/audit-logs - 班级学生 API 会按 `tenant_memberships.role_template_id -> tenant_role_templates.data_scope`、成员显式权限和 `tenant_class_members` 共同确定可见范围；非全局权限教师只能查看自己负责班级的学生。 - 学生批量导入、批量分班、学生状态、备注和跟进任务都使用独立权限点；教师默认可为范围内学生写备注和跟进任务，但不能批量导入、禁用学生或放大可见班级。 - 销售/代理客资采用首绑保护：普通扫码/分享事件不会覆盖已有归属，只有具备 `referral:write` 的租户成员可手动强制补绑。 -- CRM 当前完成配置、密钥入私密表、客资入队和队列查询；真实 webhook 发送、重试、签名在后续 `apps/worker` 中实现。 +- CRM 当前完成配置、密钥入私密表、客资入队、队列查询和 `apps/worker` 消费；worker 支持 generic webhook、钉钉、飞书、企业微信机器人消息体/签名、失败重试和日志。 - 内容资源当前完成台账、租户后台维护、学生端 SVIP 下载权限，以及 `local_dev`、阿里云 OSS、腾讯 COS、Supabase Storage 的上传/下载签名 provider。真实对象存在性校验、PDF 预览渲染、防盗链、水印和大文件上传后 worker 校验仍需继续补。 - 题库内容导航当前以 `content_entries/content_nodes` 为主模型，可表达“入口 -> 多级分类 -> 院校/专业/学科/销售意向标记”；题目集合和练习方式由 `question_collections/practice_blueprints` 管理，练习 session 会保存当次题目 ID 快照。 - 练习访问控制由 `content_entries/content_nodes/question_collections/practice_blueprints` 的 `accessRules` 合并决定；普通用户消耗 `practice_daily_usage`，事件写入 `practice_access_events`，SVIP/staff 不消耗免费额度。 @@ -245,12 +246,13 @@ GET /api/tenant-admin/audit-logs 2. 接入真实短信 provider：阿里云/腾讯云，密钥放 `app_private.tenant_secrets` 或生产 Vault。 3. 接入真实 OAuth provider：微信网页、微信小程序、QQ，并处理旧 PocketBase 身份映射。 4. 补退款、支付补偿任务、对账、异常订单处理和优惠券核销报表。 -5. 增加 `apps/worker`：支付补偿、CRM webhook、日报统计、导入后检查。 +5. 扩展 `apps/worker`：支付补偿、日报统计、导入后检查、CRM 死信告警和公共题库同步。 6. 开始 Taro scaffold，把 `supabaseApi` 抽到跨端包或适配层。 ## 测试命令 ```text npm run test:api +npm run test:worker:crm npm run check:refactor ``` diff --git a/docs/refactor/blueprint-coverage.md b/docs/refactor/blueprint-coverage.md index b3ca98d6..e8d1ee0c 100644 --- a/docs/refactor/blueprint-coverage.md +++ b/docs/refactor/blueprint-coverage.md @@ -41,4 +41,4 @@ 3. 学习统计增强：排行榜防刷/预聚合、断点续练、专项练习策略和更细题型分析。 4. 视频会员控制：视频资源签名 URL、防盗链、水印、播放次数和会员权益。 5. 数据看板 API：把旧 dashboard/revenue 统计迁到新 API。 -6. 真实 provider：短信、微信/QQ 登录、微信支付/支付宝、CRM worker。 +6. 真实 provider：短信、微信/QQ 登录、微信支付/支付宝；CRM worker 基础已落地，继续补分配策略和告警。 diff --git a/docs/refactor/crm-worker.md b/docs/refactor/crm-worker.md new file mode 100644 index 00000000..259076c2 --- /dev/null +++ b/docs/refactor/crm-worker.md @@ -0,0 +1,107 @@ +# CRM Webhook Worker + +更新时间：2026-06-29 + +`apps/worker` 是后端异步任务进程，当前首个落地任务是 CRM 客资 webhook 推送。API 只负责在客资首绑时写入 `crm_webhook_queue`；worker 负责消费到期任务、签名、发送、失败重试和写入 `crm_webhook_log`。 + +## 运行命令 + +本地单次消费一批任务： + +```bash +npm --workspace @tiku-saas/worker run crm:once +``` + +或使用根脚本测试： + +```bash +npm run test:worker:crm +``` + +生产常驻： + +```bash +npm --workspace @tiku-saas/worker run start +``` + +建议 API 和 worker 作为两个独立进程部署，共用同一个 `DATABASE_URL`。 + +## 配置 + +环境变量： + +```text +WORKER_CRM_BATCH_SIZE=20 +WORKER_CRM_POLL_INTERVAL_MS=10000 +WORKER_CRM_MAX_ATTEMPTS=5 +WORKER_CRM_BACKOFF_SECONDS=5,30,120,600,1800 +WORKER_CRM_REQUEST_TIMEOUT_MS=10000 +WORKER_CRM_ALLOW_INSECURE_LOCALHOST=false +``` + +租户后台配置仍走： + +```text +PUT /api/crm/config +GET /api/crm/config +GET /api/crm/queue +``` + +密钥必须写入 `app_private.tenant_secrets`，公共表 `crm_config.secret_ref` 只保留引用，例如： + +```text +app_private.tenant_secrets:crm:webhook +``` + +## Provider + +worker 根据队列 payload 的 `provider` 或 webhook URL 自动判断： + +| provider | URL 特征 | 行为 | +| --- | --- | --- | +| `generic` | 默认 | POST `{ event: "lead.created", data: ... }` | +| `dingtalk` | `dingtalk.com` | markdown 消息；按钉钉自定义机器人安全设置做 timestamp/sign | +| `feishu` | `feishu.cn` / `larksuite.com` | interactive card；按飞书自定义机器人签名做 timestamp/sign | +| `wecom` | `qyapi.weixin.qq.com` / `work.weixin.qq.com` | 企业微信群机器人 markdown 消息 | + +官方参考： + +- 钉钉开放平台：[自定义机器人安全设置](https://open.dingtalk.com/document/robots/customize-robot-security-settings)，`timestamp + "\n" + secret` 做 HMAC-SHA256 后 Base64。 +- 飞书开放平台：[自定义机器人签名校验](https://open.feishu.cn/document/client-docs/bot-v3/add-custom-bot)，使用 timestamp 和 secret 生成签名。 +- 企业微信开发者文档：[群机器人配置说明](https://developer.work.weixin.qq.com/document/path/91770)，群机器人 webhook 使用 `msgtype` 消息体。 + +## 队列状态 + +`crm_webhook_queue.status`： + +```text +pending -> processing -> sent +pending -> processing -> retrying -> processing -> sent +pending -> processing -> retrying -> failed +pending -> discarded +``` + +说明： + +- `pending` / `retrying`：等待 worker 消费。 +- `processing`：worker 已抢占任务。 +- `sent`：目标 webhook 返回 2xx。 +- `failed`：超过最大重试次数。 +- `discarded`：租户 CRM 未启用或 URL 缺失，任务不会继续重试。 + +每次尝试都会写入 `crm_webhook_log`，日志只记录 provider、目标 host、请求体和响应摘要，不写入 webhook secret。 + +## 安全边界 + +- 前端不能直接写 `crm_webhook_queue`。 +- 前端不能持有 CRM webhook secret。 +- 生产环境 webhook 必须使用 HTTPS。 +- `WORKER_CRM_ALLOW_INSECURE_LOCALHOST=true` 只用于本地 fake webhook 测试。 +- worker 使用数据库服务端连接读取 `app_private.tenant_secrets`，不要把 secret 复制到公共表。 + +## 后续增强 + +- 轮询/定向分配销售。 +- 钉钉/飞书/企微富卡片模板。 +- 失败告警和死信运营后台。 +- 批量 CRM 推送、跟进效果统计。 diff --git a/docs/refactor/implementation-status.md b/docs/refactor/implementation-status.md index ec7af04f..9bfaadba 100644 --- a/docs/refactor/implementation-status.md +++ b/docs/refactor/implementation-status.md @@ -35,7 +35,7 @@ | 资料下载/PDF | 已扩展 `content_assets`，新增资源台账和导入任务表 | 旧 `app_assets/images` 兼容导入 | 租户后台资源管理、上传/下载签名占位、学生端资料列表/下载权限已实现 | 核心 API 集成测试含 SVIP 资料下载 | 资料资源基础闭环可跑，真实 OSS/COS 签名、PDF 预览渲染、资料下载前端待补 | | 个人中心 | 已建 `student_profiles`、会员权益、订单、练习记录 | 已支持部分用户资料导入 | 个人资料、目标院校/专业、会员状态、最近练习、统计聚合 API 已实现 | 核心 API 烟测 | 学生端基础个人中心已实现，签到/任务/更细统计待补 | | 活动/优惠 | 已建优惠券、激活码、激活码批次、banner、FAQ、公告等基础表 | 部分支持 | banner/FAQ/公告只读与租户后台维护、激活码预检查/兑换、激活码批次、批量生成激活码、优惠券维护、前台领取/下单抵扣已实现 | 核心 API 集成测试 | 基础运营后台可用，复杂活动规则、营销自动化、核销报表待补 | -| 销售/代理客资追踪 | 已建推荐码、首绑客资、团队关系、小程序码缓存、CRM 队列 | 旧 `referral_tracks` 已有映射基础 | 邀请码、扫码/分享事件、首绑保护、销售统计、客资明细、手动补绑、团队关系、CRM 配置/队列已实现 | 核心 API 集成测试 | 增长链路基础可用，真实微信小程序码、分佣结算单、CRM worker 推送待补 | +| 销售/代理客资追踪 | 已建推荐码、首绑客资、团队关系、小程序码缓存、CRM 队列 | 旧 `referral_tracks` 已有映射基础 | 邀请码、扫码/分享事件、首绑保护、销售统计、客资明细、手动补绑、团队关系、CRM 配置/队列、CRM worker 推送已实现 | 核心 API 集成测试、CRM worker 集成测试 | 增长链路基础可用，真实微信小程序码、CRM 分配策略、富卡片和销售转化看板待补 | | 租户后台 | 已建品牌、域名、设置、支付账户、登录 provider、私密密钥表、成员、审计日志、资源台账、导入台账、内容导航台账 | 不适用 | 概览、品牌、设置、域名、支付账户、登录配置、密钥掩码、活动内容、兑换码/优惠券、成员管理、权限矩阵、审计查询、内容入口/分类树/题目集合/练习蓝图维护、资源管理、题目 JSON 导入已实现 | 核心 API 集成测试含角色/权限/租户隔离/密钥不泄露/导航/组卷/资源与导入断言 | 租户配置与运营闭环可用，前端权限 UI、Excel 导入、真实对象存储签名待补 | | 平台后台 | 已建 SaaS 套餐、订阅、账单、服务费、用量 | 不适用 | 租户管理、账单、收款确认、用量记录、平台管理员 Supabase JWT 鉴权已实现 | API 集成测试 | 平台收费链路骨架可用，平台审计报表/自动计费待补 | | 登录认证 | 已建短信验证码、会话、OAuth provider 配置表，并支持 `auth_user_id` 映射 | 旧用户映射已预留 | 短信 mock 登录、迁移期 session、Supabase JWT 验签映射、微信小程序登录主链路已实现 | API 集成测试 | H5 Supabase Auth 可联调；真实短信/微信网页/QQ 登录生产联调待补 | diff --git a/docs/refactor/next-development-todo.md b/docs/refactor/next-development-todo.md index 5300c93b..356ee800 100644 --- a/docs/refactor/next-development-todo.md +++ b/docs/refactor/next-development-todo.md @@ -133,9 +133,8 @@ - 主题预览和发布。 3. CRM worker - - 钉钉、飞书、企业微信机器人 adapter。 - - 轮询/定向分配。 - - 推送失败重试和签名。 + - 已完成 `apps/worker` CRM 队列消费、generic webhook、钉钉、飞书、企业微信机器人 adapter、签名、失败重试和日志。 + - 继续补轮询/定向分配、富卡片模板、失败告警、死信运营后台和批量 CRM 推送。 4. 运维 - 后台操作审计报表。 diff --git a/docs/refactor/project-structure.md b/docs/refactor/project-structure.md index 8f7bfdff..5d570e12 100644 --- a/docs/refactor/project-structure.md +++ b/docs/refactor/project-structure.md @@ -38,6 +38,15 @@ F:\project Dockerfile package.json tsconfig.json + worker/ 后台异步任务进程 + src/ + jobs/ + crm.ts CRM webhook 队列消费、签名、重试、日志 + config.ts worker 环境变量 + db.ts worker 数据库连接 + index.ts worker CLI/常驻循环入口 + package.json + tsconfig.json packages/ config/ 共享配置和 env 工具 @@ -117,6 +126,7 @@ git status --short --branch 正常情况下，后续提交应只包含这些路径： - `apps/api/**` +- `apps/worker/**` - `packages/**` - `supabase/**` - `scripts/import-pocketbase/**` diff --git a/package-lock.json b/package-lock.json index 66b33b8d..554793f4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -50,6 +50,36 @@ "dev": true, "license": "MIT" }, + "apps/worker": { + "name": "@tiku-saas/worker", + "version": "0.1.0", + "dependencies": { + "pg": "^8.16.3" + }, + "devDependencies": { + "@types/node": "^24.0.4", + "@types/pg": "^8.15.4", + "tsx": "^4.20.3", + "typescript": "^5.8.3" + } + }, + "apps/worker/node_modules/@types/node": { + "version": "24.13.2", + "resolved": "https://registry.npmjs.org/@types/node/-/node-24.13.2.tgz", + "integrity": "sha512-fRa09kZTgu8o71KFcDjUFuc7F+dEbZYZmkI0mg5YBTRs0yMKjYHsq/c0urDKeDb+D5qVgXOdFcuu+DZPKOITwA==", + "dev": true, + "license": "MIT", + "dependencies": { + "undici-types": "~7.18.0" + } + }, + "apps/worker/node_modules/undici-types": { + "version": "7.18.2", + "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-7.18.2.tgz", + "integrity": "sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==", + "dev": true, + "license": "MIT" + }, "node_modules/@esbuild/netbsd-arm64": { "version": "0.28.1", "resolved": "https://registry.npmjs.org/@esbuild/netbsd-arm64/-/netbsd-arm64-0.28.1.tgz", @@ -252,6 +282,10 @@ "resolved": "scripts/import-pocketbase", "link": true }, + "node_modules/@tiku-saas/worker": { + "resolved": "apps/worker", + "link": true + }, "node_modules/@types/node": { "version": "25.6.0", "resolved": "https://registry.npmjs.org/@types/node/-/node-25.6.0.tgz", diff --git a/package.json b/package.json index 87fd4f7f..ea87974e 100644 --- a/package.json +++ b/package.json @@ -18,9 +18,11 @@ "scripts": { "dev:api": "npm --workspace @tiku-saas/api run dev", "build:api": "npm --workspace @tiku-saas/api run build", + "build:worker": "npm --workspace @tiku-saas/worker run build", "check:api": "npm --workspace @tiku-saas/api run check", "check:importer": "npm --workspace @tiku-saas/import-pocketbase run check", - "check:refactor": "npm run check:api && npm run check:importer && npm run pb:import:validate && npm run test:api", + "check:worker": "npm --workspace @tiku-saas/worker run check", + "check:refactor": "npm run check:api && npm run check:worker && npm run check:importer && npm run pb:import:validate && npm run test:api", "docker:api:build": "docker compose -f docker-compose.api.yml build", "docker:api:up": "docker compose -f docker-compose.api.yml up api", "docker:api:down": "docker compose -f docker-compose.api.yml down", @@ -31,6 +33,7 @@ "db:smoke-seed": "node scripts/smoke-seed.js", "smoke:core-api": "node scripts/smoke-core-api.js", "test:api": "npm run db:smoke-seed && npm run build:api && node scripts/api-integration-test.js --start-server", + "test:worker:crm": "npm run db:smoke-seed && npm run build:worker && node scripts/crm-worker-integration-test.js", "test:api:remote": "node scripts/api-integration-test.js", "pb:schema:summary": "npm --workspace @tiku-saas/import-pocketbase run schema:summary", "pb:schema:risk": "npm --workspace @tiku-saas/import-pocketbase run schema:risk", diff --git a/scripts/crm-worker-integration-test.js b/scripts/crm-worker-integration-test.js new file mode 100644 index 00000000..7b67b9c3 --- /dev/null +++ b/scripts/crm-worker-integration-test.js @@ -0,0 +1,230 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import pg from 'pg'; +import { spawn } from 'node:child_process'; + +const databaseUrl = process.env.DATABASE_URL || 'postgresql://postgres:postgres@127.0.0.1:54322/postgres'; +const tenantId = '00000000-0000-0000-0000-000000000001'; +const adminUserId = '00000000-0000-0000-0000-000000000102'; +const studentUserId = '00000000-0000-0000-0000-000000000101'; +const salesUserId = '00000000-0000-0000-0000-000000000104'; + +function getFreePort() { + return new Promise((resolve, reject) => { + const server = http.createServer(); + server.listen(0, '127.0.0.1', () => { + const address = server.address(); + server.close(() => resolve(address.port)); + }); + server.on('error', reject); + }); +} + +async function startWebhookServer() { + const port = await getFreePort(); + const requests = []; + const server = http.createServer((req, res) => { + let raw = ''; + req.on('data', chunk => { + raw += chunk.toString(); + }); + req.on('end', () => { + requests.push({ + url: req.url, + headers: req.headers, + body: raw ? JSON.parse(raw) : {}, + }); + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true })); + }); + }); + await new Promise(resolve => server.listen(port, '127.0.0.1', resolve)); + return { + url: `http://127.0.0.1:${port}/crm`, + requests, + close: () => new Promise(resolve => server.close(resolve)), + }; +} + +async function runWorkerOnce() { + const child = spawn(process.execPath, ['apps/worker/dist/apps/worker/src/index.js', '--once', '--job', 'crm'], { + cwd: process.cwd(), + env: { + ...process.env, + DATABASE_URL: databaseUrl, + WORKER_CRM_ALLOW_INSECURE_LOCALHOST: 'true', + WORKER_CRM_REQUEST_TIMEOUT_MS: '5000', + }, + stdio: ['ignore', 'pipe', 'pipe'], + windowsHide: true, + }); + let output = ''; + child.stdout.on('data', chunk => { + output += chunk.toString(); + }); + child.stderr.on('data', chunk => { + output += chunk.toString(); + }); + const code = await new Promise(resolve => child.on('exit', resolve)); + assert.equal(code, 0, `worker should exit 0\n${output}`); + assert.match(output, /processed=\d+/, 'worker output should include processed count'); +} + +async function main() { + const webhook = await startWebhookServer(); + const pool = new pg.Pool({ connectionString: databaseUrl }); + try { + await pool.query('begin'); + await pool.query( + ` + delete from public.crm_webhook_log + where tenant_id = $1 and record_id = $2 + `, + [tenantId, studentUserId], + ); + await pool.query( + ` + delete from public.crm_webhook_queue + where tenant_id = $1 and idempotency_key = 'lead:worker-integration' + `, + [tenantId], + ); + await pool.query( + ` + insert into app_private.tenant_secrets (tenant_id, secret_scope, secret_key, secret_value, provider, last_rotated_at) + values ($1, 'crm', 'webhook', 'crm-secret-worker-test', 'webhook', now()) + on conflict (tenant_id, secret_scope, secret_key) + do update set secret_value = excluded.secret_value, + provider = excluded.provider, + last_rotated_at = now(), + updated_at = now() + `, + [tenantId], + ); + await pool.query( + ` + insert into public.crm_config ( + tenant_id, enabled, url, secret_ref, form_name, exam_type, timeout_sec, delay_sec + ) + values ($1, true, $2, 'app_private.tenant_secrets:crm:webhook', 'Worker集成测试客资', '专升本', 5, 0) + on conflict (tenant_id) + do update set enabled = excluded.enabled, + url = excluded.url, + secret_ref = excluded.secret_ref, + form_name = excluded.form_name, + exam_type = excluded.exam_type, + timeout_sec = excluded.timeout_sec, + delay_sec = excluded.delay_sec, + updated_at = now() + `, + [tenantId, webhook.url], + ); + await pool.query( + ` + insert into public.referral_leads ( + tenant_id, student_user_id, referrer_user_id, ref_code, source, bind_type, status, metadata + ) + values ($1, $2, $3, 'WORKER1'::citext, 'integration', 'manual', 'protected', '{"source":"crm-worker-test"}'::jsonb) + on conflict (tenant_id, student_user_id) + do update set referrer_user_id = excluded.referrer_user_id, + ref_code = excluded.ref_code, + source = excluded.source, + bind_type = excluded.bind_type, + status = excluded.status, + updated_at = now() + `, + [tenantId, studentUserId, salesUserId], + ); + const lead = await pool.query( + ` + select id, bound_at + from public.referral_leads + where tenant_id = $1 and student_user_id = $2 + limit 1 + `, + [tenantId, studentUserId], + ); + await pool.query( + ` + insert into public.crm_webhook_queue ( + tenant_id, record_id, status, scheduled_at, lead_id, source, + payload, idempotency_key, target_url + ) + values ( + $1, $2, 'pending', now() - interval '1 second', $3, 'integration', + $4::jsonb, 'lead:worker-integration', $5 + ) + on conflict (tenant_id, idempotency_key) where idempotency_key is not null + do update set status = 'pending', + scheduled_at = excluded.scheduled_at, + next_attempt_at = null, + attempts = 0, + last_error = null, + payload = excluded.payload, + target_url = excluded.target_url, + updated_at = now() + `, + [ + tenantId, + studentUserId, + lead.rows[0].id, + JSON.stringify({ + provider: 'generic', + formName: 'Worker集成测试客资', + examType: '专升本', + source: 'integration', + leadId: lead.rows[0].id, + student: { id: studentUserId, name: '烟测学生', phone: '13800000000' }, + referral: { refCode: 'WORKER1', referrerUserId: salesUserId, boundAt: lead.rows[0].bound_at }, + }), + webhook.url, + ], + ); + await pool.query('commit'); + + await runWorkerOnce(); + assert.equal(webhook.requests.length, 1, 'worker should send exactly one webhook request'); + assert.equal(webhook.requests[0].body.event, 'lead.created', 'generic webhook should send lead.created event'); + assert.equal(webhook.requests[0].body.data.student.phone, '13800000000', 'webhook body should include lead phone'); + assert.ok(!JSON.stringify(webhook.requests[0]).includes('crm-secret-worker-test'), 'webhook request should not leak CRM secret for generic provider'); + + const queue = await pool.query( + ` + select status, attempts, last_http_code, sent_at + from public.crm_webhook_queue + where tenant_id = $1 and idempotency_key = 'lead:worker-integration' + limit 1 + `, + [tenantId], + ); + assert.equal(queue.rows[0]?.status, 'sent', 'CRM queue item should be sent'); + assert.equal(queue.rows[0]?.attempts, 1, 'CRM queue item should record one attempt'); + assert.equal(queue.rows[0]?.last_http_code, 200, 'CRM queue item should record HTTP 200'); + assert.ok(queue.rows[0]?.sent_at, 'CRM queue item should record sent_at'); + + const logs = await pool.query( + ` + select outcome, http_code, request_payload, response_summary + from public.crm_webhook_log + where tenant_id = $1 and record_id = $2 + order by created_at desc + limit 1 + `, + [tenantId, studentUserId], + ); + assert.equal(logs.rows[0]?.outcome, 'sent', 'CRM log should record sent outcome'); + assert.equal(logs.rows[0]?.http_code, 200, 'CRM log should record HTTP 200'); + assert.equal(logs.rows[0]?.request_payload?.provider, 'generic', 'CRM log should record provider'); + assert.ok(!JSON.stringify(logs.rows[0]).includes('crm-secret-worker-test'), 'CRM log should not leak secret'); + + console.log('CRM worker integration test complete.'); + } finally { + await pool.end(); + await webhook.close(); + } +} + +main().catch(error => { + console.error(error); + process.exit(1); +}); diff --git a/scripts/smoke-seed.js b/scripts/smoke-seed.js index 64ab0c8b..0b440906 100644 --- a/scripts/smoke-seed.js +++ b/scripts/smoke-seed.js @@ -309,6 +309,7 @@ async function main() { where tenant_id = $1 and ( record_id in ($2::text, $3::text, $4::text, $5::text) + or idempotency_key = 'lead:worker-integration' or lead_id in ( select id::text from public.referral_leads @@ -320,6 +321,18 @@ async function main() { [tenantId, ids.user, ids.tenantOperatorUser, ids.tenantSalesUser, ids.tenantAgentUser], ); + await client.query( + ` + delete from public.crm_webhook_log + where tenant_id = $1 + and ( + record_id in ($2::text, $3::text, $4::text, $5::text) + or response_summary like '%worker%' + ) + `, + [tenantId, ids.user, ids.tenantOperatorUser, ids.tenantSalesUser, ids.tenantAgentUser], + ); + await client.query( ` with transient_users as ( diff --git a/supabase/migrations/202606290010_crm_worker_hardening.sql b/supabase/migrations/202606290010_crm_worker_hardening.sql new file mode 100644 index 00000000..ad6a4e2e --- /dev/null +++ b/supabase/migrations/202606290010_crm_worker_hardening.sql @@ -0,0 +1,16 @@ +alter table public.crm_webhook_queue + add column if not exists provider text, + add column if not exists last_attempt_at timestamptz, + add column if not exists last_response_summary text; + +do $$ +begin + if not exists (select 1 from pg_constraint where conname = 'crm_webhook_queue_status_check') then + alter table public.crm_webhook_queue + add constraint crm_webhook_queue_status_check + check (status in ('pending', 'processing', 'retrying', 'sent', 'failed', 'discarded')); + end if; +end $$; + +create index if not exists idx_crm_queue_due + on public.crm_webhook_queue(tenant_id, status, scheduled_at, next_attempt_at);