feat: add crm webhook worker

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"
+  }
+}

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),
+};

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();
+}

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();
+}

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<string, unknown>;
+  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<string, unknown> | null;
+}
+
+interface PreparedRequest {
+  provider: CrmProvider;
+  url: string;
+  body: Record<string, unknown>;
+  headers: Record<string, string>;
+}
+
+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<string, unknown> {
+  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
+}
+
+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<string, string> {
+  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<string, string> {
+  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<string, string>) {
+  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<CrmConfigRow>(
+    `
+    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<SecretRow>(
+    `
+    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<CrmQueueRow>(
+    `
+    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<SendResult> {
+  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<ProcessResult> {
+  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;
+}

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"]
+}