feat: add crm dead-letter operations and benchmark summary

2026-06-30 23:07:43 +08:00
parent 90ad9e90a3
commit 9891fe9ef3
22 changed files with 1284 additions and 110 deletions
--- a/apps/api/src/features/referral/index.ts
+++ b/apps/api/src/features/referral/index.ts
@@ -15,6 +15,9 @@ import {
 } from './commission.js';
 import {
  crmConfigRoute,
+  crmDeadLettersRoute,
+  crmQueueActionRoute,
+  crmQueueLogsRoute,
  crmQueueRoute,
  referralBindRoute,
  referralInviteCodeRoute,
@@ -45,6 +48,9 @@ export const referralRoutes: RouteDefinition[] = [
  ['GET', '/api/crm/config', crmConfigRoute],
  ['PUT', '/api/crm/config', upsertCrmConfigRoute],
  ['GET', '/api/crm/queue', crmQueueRoute],
+  ['GET', '/api/crm/dead-letters', crmDeadLettersRoute],
+  ['GET', '/api/crm/queue/logs', crmQueueLogsRoute],
+  ['POST', '/api/crm/queue/action', crmQueueActionRoute],
  ['GET', '/api/commission/settings', commissionSettingsRoute],
  ['PUT', '/api/commission/settings', updateCommissionSettingsRoute],
  ['PUT', '/api/commission/member-rate', updateMemberCommissionRateRoute],
--- a/apps/api/src/features/referral/routes.ts
+++ b/apps/api/src/features/referral/routes.ts
@@ -13,6 +13,10 @@ const EVENT_TYPES = ['enter', 'register', 'purchase', 'share', 'scan', 'manual_b
 const TRACK_SOURCES = ['share', 'qrcode', 'timeline', 'miniapp', 'h5', 'manual', 'unknown'];
 const CRM_ASSIGNMENT_MODES = ['none', 'direct', 'round_robin', 'referrer'];
 const CRM_ASSIGNABLE_ROLES = ['tenant_owner', 'tenant_admin', 'tenant_operator', 'teacher', 'sales', 'agent'];
+const CRM_QUEUE_STATUSES = ['pending', 'processing', 'retrying', 'sent', 'failed', 'discarded'];
+const CRM_DEAD_LETTER_STATUSES = ['failed', 'discarded'];
+const CRM_QUEUE_ACTIONS = ['retry', 'ignore'];
+const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;

 function nullableString(value: unknown) {
  return typeof value === 'string' && value.trim() ? value.trim() : null;
@@ -30,6 +34,97 @@ function optionalChoice(value: unknown, allowed: string[], fallback: string) {
  return candidate;
 }

+function optionalUuidString(value: unknown, key: string) {
+  const text = nullableString(value);
+  if (!text) return '';
+  if (!UUID_RE.test(text)) {
+    throw new HttpError(400, `${key} must be a UUID`, 'INVALID_UUID');
+  }
+  return text;
+}
+
+function requiredUuidString(body: JsonBody, key: string) {
+  const text = optionalUuidString(body[key], key);
+  if (!text) throw new HttpError(400, `${key} is required`, 'REQUIRED_FIELD');
+  return text;
+}
+
+function limitedOptionalString(value: unknown, key: string, maxLength: number) {
+  const text = nullableString(value);
+  if (!text) return null;
+  if (text.length > maxLength) {
+    throw new HttpError(400, `${key} is too long`, 'FIELD_TOO_LONG');
+  }
+  return text;
+}
+
+function redactSensitiveString(value: string) {
+  return value
+    .replace(/("(?:access[_-]?token|token|secret|sign|signature|key|password)"\s*:\s*")[^"]+(")/gi, '$1[redacted]$2')
+    .replace(/(access[_-]?token|token|secret|sign|signature|key|password)=([^&\s]+)/gi, '$1=[redacted]')
+    .replace(/(Bearer\s+)[A-Za-z0-9._~+/=-]+/gi, '$1[redacted]')
+    .replace(/(https?:\/\/[^/\s]+\/(?:robot\/send|bot|webhook|open-apis\/bot)\/)[^?\s/]+/gi, '$1[redacted]');
+}
+
+function isSensitiveFieldKey(key: string) {
+  const normalized = key.replace(/[^a-z0-9]/gi, '').toLowerCase();
+  return new Set([
+    'accesstoken',
+    'refreshtoken',
+    'token',
+    'secret',
+    'clientsecret',
+    'appsecret',
+    'password',
+    'credential',
+    'credentials',
+    'authorization',
+    'signature',
+    'sign',
+    'key',
+    'apikey',
+    'secretkey',
+    'accesskey',
+    'privatekey',
+  ]).has(normalized);
+}
+
+function redactSensitiveValue(value: unknown): unknown {
+  if (typeof value === 'string') return redactSensitiveString(value);
+  if (Array.isArray(value)) return value.map(item => redactSensitiveValue(item));
+  if (!value || typeof value !== 'object') return value;
+  const output: Record<string, unknown> = {};
+  for (const [key, raw] of Object.entries(value as Record<string, unknown>)) {
+    if (isSensitiveFieldKey(key)) {
+      output[key] = '[redacted]';
+    } else {
+      output[key] = redactSensitiveValue(raw);
+    }
+  }
+  return output;
+}
+
+function safeUrlForResponse(value: unknown) {
+  const text = nullableString(value);
+  if (!text) return null;
+  try {
+    const url = new URL(text);
+    url.username = '';
+    url.password = '';
+    for (const key of Array.from(url.searchParams.keys())) {
+      if (/token|secret|password|credential|authorization|signature|sign|key/i.test(key)) {
+        url.searchParams.set(key, '[redacted]');
+      }
+    }
+    if (/\/(robot\/send|bot|webhook|open-apis\/bot)\//i.test(url.pathname)) {
+      url.pathname = url.pathname.replace(/(\/(?:robot\/send|bot|webhook|open-apis\/bot)\/)[^/]+/i, '$1[redacted]');
+    }
+    return url.toString();
+  } catch {
+    return redactSensitiveString(text);
+  }
+}
+
 function stringArray(value: unknown) {
  if (!Array.isArray(value)) return [];
  const seen = new Set<string>();
@@ -543,6 +638,43 @@ function crmPermission(auth: TenantAdminAuth) {
  requireTenantPermission(auth, 'crm:read');
 }

+async function recordCrmAudit(
+  client: pg.PoolClient,
+  auth: TenantAdminAuth,
+  action: string,
+  targetId: string | null,
+  details: Record<string, unknown> = {},
+) {
+  await client.query(
+    `
+    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+    values ($1, $2, $3, 'crm_webhook_queue', $4, $5::jsonb)
+    `,
+    [auth.tenantId, auth.userId, action, targetId, JSON.stringify(redactSensitiveValue(details))],
+  );
+}
+
+function crmQueueResponseItem<T extends Record<string, unknown>>(item: T) {
+  return {
+    ...item,
+    targetUrl: safeUrlForResponse(item.targetUrl),
+    payload: redactSensitiveValue(item.payload || {}) as Record<string, unknown>,
+    operatorMetadata: redactSensitiveValue(item.operatorMetadata || {}) as Record<string, unknown>,
+    lastError: item.lastError ? redactSensitiveString(String(item.lastError)) : item.lastError,
+    lastResponseSummary: item.lastResponseSummary ? redactSensitiveString(String(item.lastResponseSummary)) : item.lastResponseSummary,
+  };
+}
+
+function crmLogResponseItem<T extends Record<string, unknown>>(item: T) {
+  return {
+    ...item,
+    requestBody: item.requestBody ? redactSensitiveString(String(item.requestBody)) : item.requestBody,
+    requestPayload: redactSensitiveValue(item.requestPayload || {}) as Record<string, unknown>,
+    responseSummary: item.responseSummary ? redactSensitiveString(String(item.responseSummary)) : item.responseSummary,
+    errorMessage: item.errorMessage ? redactSensitiveString(String(item.errorMessage)) : item.errorMessage,
+  };
+}
+
 export async function referralInviteCodeRoute(ctx: RequestContext) {
  const tenantId = await tenantIdFrom(ctx);
  const userId = await userIdFrom(ctx);
@@ -1146,12 +1278,20 @@ export async function crmQueueRoute(ctx: RequestContext) {
  crmPermission(auth);
  const limit = intParam(ctx, 'limit', 100, 500);
  const status = stringParam(ctx, 'status');
+  const queueId = optionalUuidString(stringParam(ctx, 'queueId') || stringParam(ctx, 'id'), 'queueId');
  const params: unknown[] = [auth.tenantId];
  const filters = ['tenant_id = $1'];
  if (status) {
+    if (!CRM_QUEUE_STATUSES.includes(status)) {
+      throw new HttpError(400, `Invalid CRM queue status: ${status}`, 'INVALID_CRM_QUEUE_STATUS');
+    }
    params.push(status);
    filters.push(`status = $${params.length}`);
  }
+  if (queueId) {
+    params.push(queueId);
+    filters.push(`id = $${params.length}::uuid`);
+  }
  params.push(limit);

  const items = await query(
@@ -1160,6 +1300,11 @@ export async function crmQueueRoute(ctx: RequestContext) {
           attempts, next_attempt_at as "nextAttemptAt", last_error as "lastError",
           last_http_code as "lastHttpCode", lead_id as "leadId", sent_at as "sentAt",
           source, idempotency_key as "idempotencyKey", target_url as "targetUrl",
+           provider, last_attempt_at as "lastAttemptAt", last_response_summary as "lastResponseSummary",
+           dead_lettered_at as "deadLetteredAt", ignored_at as "ignoredAt",
+           ignored_by as "ignoredBy", last_operator_user_id as "lastOperatorUserId",
+           last_operator_action as "lastOperatorAction", last_operator_note as "lastOperatorNote",
+           last_operator_at as "lastOperatorAt", operator_metadata as "operatorMetadata",
           payload, created_at as "createdAt", updated_at as "updatedAt"
    from public.crm_webhook_queue
    where ${filters.join(' and ')}
@@ -1168,5 +1313,297 @@ export async function crmQueueRoute(ctx: RequestContext) {
    `,
    params,
  );
-  return { items };
+  return { items: items.map(item => crmQueueResponseItem(item as Record<string, unknown>)) };
+}
+
+export async function crmDeadLettersRoute(ctx: RequestContext) {
+  const auth = await requireTenantAdmin(ctx);
+  crmPermission(auth);
+  const limit = intParam(ctx, 'limit', 100, 500);
+  const status = stringParam(ctx, 'status');
+  const source = stringParam(ctx, 'source');
+  const params: unknown[] = [auth.tenantId];
+  const filters = [`status = any($2::text[])`];
+  params.push(CRM_DEAD_LETTER_STATUSES);
+  if (status) {
+    if (!CRM_DEAD_LETTER_STATUSES.includes(status)) {
+      throw new HttpError(400, `Invalid CRM dead-letter status: ${status}`, 'INVALID_CRM_DEAD_LETTER_STATUS');
+    }
+    params.push(status);
+    filters.push(`status = $${params.length}`);
+  }
+  if (source) {
+    params.push(source);
+    filters.push(`source = $${params.length}`);
+  }
+  params.push(limit);
+
+  const items = await query(
+    `
+    select id, record_id as "recordId", status, scheduled_at as "scheduledAt",
+           attempts, next_attempt_at as "nextAttemptAt", last_error as "lastError",
+           last_http_code as "lastHttpCode", lead_id as "leadId", sent_at as "sentAt",
+           source, idempotency_key as "idempotencyKey", target_url as "targetUrl",
+           provider, last_attempt_at as "lastAttemptAt", last_response_summary as "lastResponseSummary",
+           dead_lettered_at as "deadLetteredAt", ignored_at as "ignoredAt",
+           ignored_by as "ignoredBy", last_operator_user_id as "lastOperatorUserId",
+           last_operator_action as "lastOperatorAction", last_operator_note as "lastOperatorNote",
+           last_operator_at as "lastOperatorAt", operator_metadata as "operatorMetadata",
+           payload, created_at as "createdAt", updated_at as "updatedAt"
+    from public.crm_webhook_queue
+    where tenant_id = $1 and ${filters.join(' and ')}
+    order by coalesce(dead_lettered_at, updated_at, created_at) desc
+    limit $${params.length}
+    `,
+    params,
+  );
+  const summary = await queryOne<{
+    failed: string;
+    discarded: string;
+    ignored: string;
+    total: string;
+    oldestOpenAt: string | null;
+  }>(
+    `
+    select count(*) filter (where status = 'failed')::text as failed,
+           count(*) filter (where status = 'discarded')::text as discarded,
+           count(*) filter (where ignored_at is not null)::text as ignored,
+           count(*)::text as total,
+           min(coalesce(dead_lettered_at, updated_at, created_at)) filter (where ignored_at is null) as "oldestOpenAt"
+    from public.crm_webhook_queue
+    where tenant_id = $1 and status in ('failed', 'discarded')
+    `,
+    [auth.tenantId],
+  );
+  return {
+    summary: {
+      failed: Number(summary?.failed || 0),
+      discarded: Number(summary?.discarded || 0),
+      ignored: Number(summary?.ignored || 0),
+      total: Number(summary?.total || 0),
+      oldestOpenAt: summary?.oldestOpenAt || null,
+    },
+    items: items.map(item => crmQueueResponseItem(item as Record<string, unknown>)),
+  };
+}
+
+export async function crmQueueLogsRoute(ctx: RequestContext) {
+  const auth = await requireTenantAdmin(ctx);
+  crmPermission(auth);
+  const queueId = optionalUuidString(stringParam(ctx, 'queueId') || stringParam(ctx, 'id'), 'queueId');
+  if (!queueId) throw new HttpError(400, 'queueId is required', 'REQUIRED_FIELD');
+  const limit = intParam(ctx, 'limit', 50, 200);
+
+  const task = await queryOne<{
+    id: string;
+    recordId: string | null;
+    leadId: string | null;
+    idempotencyKey: string | null;
+  }>(
+    `
+    select id, record_id as "recordId", lead_id as "leadId", idempotency_key as "idempotencyKey"
+    from public.crm_webhook_queue
+    where tenant_id = $1 and id = $2::uuid
+    limit 1
+    `,
+    [auth.tenantId, queueId],
+  );
+  if (!task) throw new HttpError(404, 'CRM queue task not found', 'CRM_QUEUE_TASK_NOT_FOUND');
+
+  const items = await query(
+    `
+    select id, queue_id as "queueId", record_id as "recordId", http_code as "httpCode",
+           outcome, error_message as "errorMessage", lead_id as "leadId",
+           request_body as "requestBody", request_payload as "requestPayload",
+           response_summary as "responseSummary", signed_at as "signedAt",
+           attempt, operator_user_id as "operatorUserId", operation,
+           created_at as "createdAt", updated_at as "updatedAt"
+    from public.crm_webhook_log
+    where tenant_id = $1
+      and (
+        queue_id = $2::uuid
+        or (
+          queue_id is null
+          and (
+            ($3::text is not null and record_id = $3::text)
+            or ($4::text is not null and lead_id = $4::text)
+          )
+        )
+      )
+    order by created_at desc
+    limit $5
+    `,
+    [auth.tenantId, queueId, task.recordId, task.leadId, limit],
+  );
+  return { item: task, items: items.map(item => crmLogResponseItem(item as Record<string, unknown>)) };
+}
+
+export async function crmQueueActionRoute(ctx: RequestContext) {
+  const auth = await requireTenantAdmin(ctx);
+  requireTenantPermission(auth, 'crm:write');
+  const body = await readJsonBody(ctx);
+  const queueId = requiredUuidString(body, 'queueId');
+  const action = optionalChoice(body.action, CRM_QUEUE_ACTIONS, 'retry');
+  const note = limitedOptionalString(body.note ?? body.reason, 'note', 500);
+  const metadata = objectValue(body.metadata);
+  const safeMetadata = redactSensitiveValue(metadata) as Record<string, unknown>;
+
+  const item = await transaction(async client => {
+    const current = await client.query<{
+      id: string;
+      status: string;
+      attempts: number;
+      recordId: string | null;
+      leadId: string | null;
+      source: string | null;
+      payload: Record<string, unknown>;
+    }>(
+      `
+      select id, status, attempts, record_id as "recordId", lead_id as "leadId", source, payload
+      from public.crm_webhook_queue
+      where tenant_id = $1 and id = $2::uuid
+      limit 1
+      for update
+      `,
+      [auth.tenantId, queueId],
+    );
+    const task = current.rows[0];
+    if (!task) throw new HttpError(404, 'CRM queue task not found', 'CRM_QUEUE_TASK_NOT_FOUND');
+
+    if (action === 'retry') {
+      if (!['failed', 'discarded', 'retrying', 'pending'].includes(task.status)) {
+        throw new HttpError(409, `CRM task cannot be retried from status ${task.status}`, 'CRM_QUEUE_STATUS_NOT_RETRYABLE');
+      }
+      const result = await client.query(
+        `
+        update public.crm_webhook_queue
+        set status = 'pending',
+            next_attempt_at = now(),
+            scheduled_at = coalesce(scheduled_at, now()),
+            dead_lettered_at = null,
+            ignored_at = null,
+            ignored_by = null,
+            last_error = null,
+            last_http_code = null,
+            last_response_summary = null,
+            last_operator_user_id = $3::uuid,
+            last_operator_action = 'retry',
+            last_operator_note = $4,
+            last_operator_at = now(),
+            operator_metadata = coalesce(operator_metadata, '{}'::jsonb) || $5::jsonb,
+            updated_at = now()
+        where tenant_id = $1 and id = $2::uuid
+        returning id, record_id as "recordId", status, scheduled_at as "scheduledAt",
+                  attempts, next_attempt_at as "nextAttemptAt", last_error as "lastError",
+                  last_http_code as "lastHttpCode", lead_id as "leadId", sent_at as "sentAt",
+                  source, idempotency_key as "idempotencyKey", target_url as "targetUrl",
+                  provider, last_attempt_at as "lastAttemptAt", last_response_summary as "lastResponseSummary",
+                  dead_lettered_at as "deadLetteredAt", ignored_at as "ignoredAt",
+                  ignored_by as "ignoredBy", last_operator_user_id as "lastOperatorUserId",
+                  last_operator_action as "lastOperatorAction", last_operator_note as "lastOperatorNote",
+                  last_operator_at as "lastOperatorAt", operator_metadata as "operatorMetadata",
+                  payload, created_at as "createdAt", updated_at as "updatedAt"
+        `,
+        [
+          auth.tenantId,
+          queueId,
+          auth.userId,
+          note,
+          JSON.stringify({ lastRetry: { by: auth.userId, at: new Date().toISOString(), note, ...safeMetadata } }),
+        ],
+      );
+      await client.query(
+        `
+        insert into public.crm_webhook_log (
+          tenant_id, queue_id, record_id, http_code, outcome, error_message, lead_id,
+          request_payload, response_summary, signed_at, attempt, operator_user_id, operation
+        )
+        values ($1, $2::uuid, $3, null, 'operator_retry', $4, $5, $6::jsonb, null, now(), $7, $8::uuid, 'retry')
+        `,
+        [
+          auth.tenantId,
+          queueId,
+          task.recordId,
+          note || 'Queued for manual retry',
+          task.leadId,
+          JSON.stringify({ source: 'tenant_operator', metadata: safeMetadata }),
+          task.attempts,
+          auth.userId,
+        ],
+      );
+      await recordCrmAudit(client, auth, 'crm.queue.retried', queueId, {
+        previousStatus: task.status,
+        attempts: task.attempts,
+        source: task.source,
+        note,
+      });
+      return result.rows[0];
+    }
+
+    if (!CRM_DEAD_LETTER_STATUSES.includes(task.status)) {
+      throw new HttpError(409, `CRM task cannot be ignored from status ${task.status}`, 'CRM_QUEUE_STATUS_NOT_IGNORABLE');
+    }
+    const result = await client.query(
+      `
+      update public.crm_webhook_queue
+      set status = 'discarded',
+          next_attempt_at = null,
+          dead_lettered_at = coalesce(dead_lettered_at, now()),
+          ignored_at = now(),
+          ignored_by = $3::uuid,
+          last_operator_user_id = $3::uuid,
+          last_operator_action = 'ignore',
+          last_operator_note = $4,
+          last_operator_at = now(),
+          operator_metadata = coalesce(operator_metadata, '{}'::jsonb) || $5::jsonb,
+          updated_at = now()
+      where tenant_id = $1 and id = $2::uuid
+      returning id, record_id as "recordId", status, scheduled_at as "scheduledAt",
+                attempts, next_attempt_at as "nextAttemptAt", last_error as "lastError",
+                last_http_code as "lastHttpCode", lead_id as "leadId", sent_at as "sentAt",
+                source, idempotency_key as "idempotencyKey", target_url as "targetUrl",
+                provider, last_attempt_at as "lastAttemptAt", last_response_summary as "lastResponseSummary",
+                dead_lettered_at as "deadLetteredAt", ignored_at as "ignoredAt",
+                ignored_by as "ignoredBy", last_operator_user_id as "lastOperatorUserId",
+                last_operator_action as "lastOperatorAction", last_operator_note as "lastOperatorNote",
+                last_operator_at as "lastOperatorAt", operator_metadata as "operatorMetadata",
+                payload, created_at as "createdAt", updated_at as "updatedAt"
+      `,
+      [
+        auth.tenantId,
+        queueId,
+        auth.userId,
+        note,
+        JSON.stringify({ lastIgnore: { by: auth.userId, at: new Date().toISOString(), note, ...safeMetadata } }),
+      ],
+    );
+    await client.query(
+      `
+      insert into public.crm_webhook_log (
+        tenant_id, queue_id, record_id, http_code, outcome, error_message, lead_id,
+        request_payload, response_summary, signed_at, attempt, operator_user_id, operation
+      )
+      values ($1, $2::uuid, $3, null, 'operator_ignore', $4, $5, $6::jsonb, null, now(), $7, $8::uuid, 'ignore')
+      `,
+      [
+        auth.tenantId,
+        queueId,
+        task.recordId,
+        note || 'Ignored by tenant operator',
+        task.leadId,
+        JSON.stringify({ source: 'tenant_operator', metadata: safeMetadata }),
+        task.attempts,
+        auth.userId,
+      ],
+    );
+    await recordCrmAudit(client, auth, 'crm.queue.ignored', queueId, {
+      previousStatus: task.status,
+      attempts: task.attempts,
+      source: task.source,
+      note,
+    });
+    return result.rows[0];
+  });
+
+  return { item: crmQueueResponseItem(item as Record<string, unknown>) };
 }