import { randomBytes } from 'node:crypto';
import type pg from 'pg';
import { HttpError, type RequestContext } from '../../core/http.js';
import { intParam, optionalString, optionalUserIdFrom, readJsonBody, requiredString, stringParam, tenantIdFrom, userIdFrom } from '../../core/request.js';
import { query, queryOne, transaction } from '../../core/db.js';
import { clientIpFrom, userAgentFrom } from '../auth/service.js';
import { hasTenantPermission, requireTenantAdmin, requireTenantPermission, type TenantAdminAuth } from '../tenant-admin/auth.js';

type JsonBody = Record<string, unknown>;

const REFERRAL_CODE_ALPHABET = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789';
const EVENT_TYPES = ['enter', 'register', 'purchase', 'share', 'scan', 'manual_bind'];
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;
}

function objectValue(value: unknown): Record<string, unknown> {
  return value && typeof value === 'object' && !Array.isArray(value) ? value as Record<string, unknown> : {};
}

function optionalChoice(value: unknown, allowed: string[], fallback: string) {
  const candidate = nullableString(value) || fallback;
  if (!allowed.includes(candidate)) {
    throw new HttpError(400, `Invalid value: ${candidate}`, 'INVALID_FIELD_VALUE');
  }
  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>();
  const items: string[] = [];
  for (const item of value) {
    const normalized = typeof item === 'string' ? item.trim() : '';
    if (!normalized || seen.has(normalized)) continue;
    seen.add(normalized);
    items.push(normalized);
  }
  return items;
}

function normalizeCode(code: string) {
  return code.replace(/\s+/g, '').toUpperCase();
}

function generateReferralCode() {
  let code = '';
  for (let index = 0; index < 6; index += 1) {
    code += REFERRAL_CODE_ALPHABET[randomBytes(1)[0] % REFERRAL_CODE_ALPHABET.length];
  }
  return code;
}

function centsToAmount(cents: number) {
  return Math.round(cents) / 100;
}

function intValue(value: unknown, fallback = 0) {
  const parsed = Number(value ?? fallback);
  return Number.isFinite(parsed) ? Math.trunc(parsed) : fallback;
}

function dateValue(value: string, fallback: string) {
  const candidate = value || fallback;
  if (!/^\d{4}-\d{2}-\d{2}$/.test(candidate)) {
    throw new HttpError(400, 'Date must use YYYY-MM-DD format', 'INVALID_DATE');
  }
  return candidate;
}

function shanghaiDateKey(date = new Date()) {
  const formatter = new Intl.DateTimeFormat('en-US', {
    timeZone: 'Asia/Shanghai',
    year: 'numeric',
    month: '2-digit',
    day: '2-digit',
  });
  const parts = Object.fromEntries(formatter.formatToParts(date).map(part => [part.type, part.value]));
  return `${parts.year}-${parts.month}-${parts.day}`;
}

function daysBetweenInclusive(startDate: string, endDate: string) {
  const start = new Date(`${startDate}T00:00:00+08:00`).getTime();
  const end = new Date(`${endDate}T00:00:00+08:00`).getTime();
  return Math.floor((end - start) / 86400000) + 1;
}

function conversionPeriodFromParams(ctx: RequestContext) {
  const today = shanghaiDateKey();
  const startDate = dateValue(stringParam(ctx, 'startDate'), today.slice(0, 8) + '01');
  const endDate = dateValue(stringParam(ctx, 'endDate'), today);
  if (startDate > endDate) {
    throw new HttpError(400, 'startDate must be before or equal to endDate', 'INVALID_DATE_RANGE');
  }
  const days = daysBetweenInclusive(startDate, endDate);
  if (days > 180) {
    throw new HttpError(400, 'Conversion report range cannot exceed 180 days', 'REPORT_RANGE_TOO_LARGE');
  }
  return { startDate, endDate, days };
}

function ratio(numerator: number, denominator: number) {
  if (!denominator) return 0;
  return Number((numerator / denominator).toFixed(4));
}

function canViewAllReferral(auth: TenantAdminAuth) {
  return hasTenantPermission(auth, 'referral:read');
}

function canViewSelfReferral(auth: TenantAdminAuth) {
  return hasTenantPermission(auth, 'referral:self') || canViewAllReferral(auth);
}

function restrictReferralScope(auth: TenantAdminAuth, requestedReferrerId: string | null) {
  if (canViewAllReferral(auth)) return requestedReferrerId;
  if (requestedReferrerId && requestedReferrerId !== auth.userId) {
    throw new HttpError(403, 'Only own referral conversion data can be viewed', 'REFERRAL_SCOPE_REQUIRED');
  }
  return auth.userId;
}

async function userHasTenantMembership(client: pg.PoolClient, tenantId: string, userId: string) {
  const result = await client.query(
    `
    select 1
    from public.tenant_memberships
    where tenant_id = $1 and user_id = $2 and status = 'active'
    limit 1
    `,
    [tenantId, userId],
  );
  return Boolean(result.rows[0]);
}

async function tenantMemberForCrmAssignment(client: pg.PoolClient, tenantId: string, userId: string) {
  const result = await client.query<{ userId: string; role: string; name: string | null; username: string | null }>(
    `
    select tm.user_id as "userId", tm.role, u.name, u.username
    from public.tenant_memberships tm
    join public.platform_users u on u.id = tm.user_id
    where tm.tenant_id = $1
      and tm.user_id = $2
      and tm.status = 'active'
      and tm.role = any($3::text[])
    limit 1
    `,
    [tenantId, userId, CRM_ASSIGNABLE_ROLES],
  );
  return result.rows[0] || null;
}

async function validateCrmAssignmentPool(client: pg.PoolClient, tenantId: string, pool: string[], mode: string) {
  if (!pool.length) {
    if (mode === 'direct' || mode === 'round_robin') {
      throw new HttpError(400, 'assignmentPool is required for this CRM assignment mode', 'CRM_ASSIGNMENT_POOL_REQUIRED');
    }
    return [];
  }
  if (pool.length > 100) {
    throw new HttpError(400, 'assignmentPool cannot contain more than 100 members', 'CRM_ASSIGNMENT_POOL_TOO_LARGE');
  }

  const rows = await client.query<{ userId: string }>(
    `
    select tm.user_id as "userId"
    from public.tenant_memberships tm
    where tm.tenant_id = $1
      and tm.user_id = any($2::uuid[])
      and tm.status = 'active'
      and tm.role = any($3::text[])
    `,
    [tenantId, pool, CRM_ASSIGNABLE_ROLES],
  );
  const valid = new Set(rows.rows.map(row => row.userId));
  const invalid = pool.filter(userId => !valid.has(userId));
  if (invalid.length) {
    throw new HttpError(400, 'CRM assignment pool contains users outside the tenant or unsupported roles', 'CRM_ASSIGNMENT_POOL_INVALID');
  }
  return pool;
}

async function crmAssignmentPoolMembers(client: pg.PoolClient, tenantId: string, pool: unknown) {
  const ids = stringArray(pool);
  if (!ids.length) return [];
  const rows = await client.query<{ userId: string; role: string; name: string | null; username: string | null }>(
    `
    with pool(user_id, ord) as (
      select value::uuid, ord
      from jsonb_array_elements_text($2::jsonb) with ordinality as item(value, ord)
    )
    select p.user_id as "userId", tm.role, u.name, u.username
    from pool p
    join public.tenant_memberships tm on tm.tenant_id = $1
      and tm.user_id = p.user_id
      and tm.status = 'active'
      and tm.role = any($3::text[])
    join public.platform_users u on u.id = tm.user_id
    order by p.ord asc
    `,
    [tenantId, JSON.stringify(ids), CRM_ASSIGNABLE_ROLES],
  );
  return rows.rows;
}

async function assignReferralLeadFromCrmConfig(
  client: pg.PoolClient,
  input: {
    tenantId: string;
    leadId: string;
    referrerUserId: string | null;
    actorUserId?: string | null;
    trigger: string;
  },
) {
  const cfg = await client.query<{
    assignmentMode: string;
    assignmentPool: unknown;
    assignmentCursor: number;
  }>(
    `
    select assignment_mode as "assignmentMode",
           assignment_pool as "assignmentPool",
           assignment_cursor as "assignmentCursor"
    from public.crm_config
    where tenant_id = $1
    limit 1
    for update
    `,
    [input.tenantId],
  );
  const config = cfg.rows[0];
  const mode = config?.assignmentMode || 'none';
  if (mode === 'none') return null;

  let assignee: { userId: string; role: string; name: string | null; username: string | null } | null = null;
  if (mode === 'referrer' && input.referrerUserId) {
    assignee = await tenantMemberForCrmAssignment(client, input.tenantId, input.referrerUserId);
  }
  if (mode === 'direct' || mode === 'round_robin') {
    const members = await crmAssignmentPoolMembers(client, input.tenantId, config.assignmentPool);
    if (members.length) {
      const cursor = Math.max(0, Number(config.assignmentCursor || 0));
      assignee = mode === 'round_robin' ? members[cursor % members.length] : members[0];
      if (mode === 'round_robin') {
        await client.query(
          'update public.crm_config set assignment_cursor = assignment_cursor + 1, updated_at = now() where tenant_id = $1',
          [input.tenantId],
        );
      }
    }
  }
  if (!assignee) return null;

  const result = await client.query(
    `
    update public.referral_leads
    set assigned_to_user_id = $3::uuid,
        assignment_mode = $4::text,
        assigned_at = now(),
        assigned_by = $5::uuid,
        metadata = metadata || jsonb_build_object(
          'crmAssignment',
          jsonb_build_object(
            'mode', $4::text,
            'trigger', $6::text,
            'assignedToUserId', $3::uuid,
            'assignedToRole', $7::text,
            'assignedAt', now()
          )
        ),
        updated_at = now()
    where tenant_id = $1 and id = $2
    returning id, student_user_id as "studentUserId", referrer_user_id as "referrerUserId",
              ref_code::text as "refCode", source, bind_type as "bindType", status,
              assigned_to_user_id as "assignedToUserId", assignment_mode as "assignmentMode",
              assigned_at as "assignedAt", assigned_by as "assignedBy",
              bound_at as "boundAt", created_at as "createdAt", updated_at as "updatedAt"
    `,
    [input.tenantId, input.leadId, assignee.userId, mode, input.actorUserId || null, input.trigger, assignee.role],
  );

  await client.query(
    `
    insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
    values ($1, $2, 'crm.lead.assigned', 'referral_leads', $3, $4::jsonb)
    `,
    [
      input.tenantId,
      input.actorUserId || null,
      input.leadId,
      JSON.stringify({ mode, trigger: input.trigger, assignedToUserId: assignee.userId, assignedToRole: assignee.role }),
    ],
  );

  return { item: result.rows[0], assignee: { ...assignee, mode } };
}

async function resolveReferralCode(tenantId: string, code: string) {
  return queryOne<{
    code: string;
    userId: string;
    status: string;
    username: string | null;
    name: string | null;
    phone: string | null;
    role: string | null;
  }>(
    `
    select rc.code::text as code, rc.user_id as "userId", rc.status,
           u.username, u.name, u.phone, tm.role
    from public.referral_codes rc
    join public.platform_users u on u.id = rc.user_id
    left join lateral (
      select role
      from public.tenant_memberships
      where tenant_id = rc.tenant_id and user_id = rc.user_id and status = 'active'
      order by case role
        when 'tenant_owner' then 1
        when 'tenant_admin' then 2
        when 'sales' then 3
        when 'agent' then 4
        when 'teacher' then 5
        else 9
      end
      limit 1
    ) tm on true
    where rc.tenant_id = $1
      and rc.code = $2::citext
      and rc.status = 'active'
    limit 1
    `,
    [tenantId, normalizeCode(code)],
  );
}

async function createOrGetReferralCode(client: pg.PoolClient, tenantId: string, userId: string, metadata: Record<string, unknown> = {}) {
  const existing = await client.query<{ code: string }>(
    `
    select code::text
    from public.referral_codes
    where tenant_id = $1 and user_id = $2
    limit 1
    `,
    [tenantId, userId],
  );
  if (existing.rows[0]?.code) return existing.rows[0].code;

  for (let attempt = 0; attempt < 30; attempt += 1) {
    const candidate = generateReferralCode();
    const inserted = await client.query<{ code: string }>(
      `
      insert into public.referral_codes (tenant_id, user_id, code, metadata)
      values ($1, $2, $3::citext, $4::jsonb)
      on conflict do nothing
      returning code::text
      `,
      [tenantId, userId, candidate, JSON.stringify(metadata)],
    );
    if (inserted.rows[0]?.code) return inserted.rows[0].code;
  }

  throw new HttpError(500, 'Failed to generate referral code', 'REFERRAL_CODE_GENERATION_FAILED');
}

async function enqueueCrmLead(
  client: pg.PoolClient,
  input: {
    tenantId: string;
    leadId: string;
    studentUserId: string;
    source: string;
    metadata?: Record<string, unknown>;
  },
) {
  const cfg = await client.query<{
    enabled: boolean;
    url: string | null;
    form_name: string | null;
    exam_type: string | null;
    delay_sec: number | null;
    timeout_sec: number | null;
  }>(
    `
    select enabled, url, form_name, exam_type, delay_sec, timeout_sec
    from public.crm_config
    where tenant_id = $1
    limit 1
    `,
    [input.tenantId],
  );
  const config = cfg.rows[0];
  if (!config?.enabled || !config.url) return null;

  const student = await client.query<{
    id: string;
    username: string | null;
    name: string | null;
    phone: string | null;
    email: string | null;
    raw_profile: Record<string, unknown>;
  }>(
    `
    select id, username, name, phone, email::text, raw_profile
    from public.platform_users
    where id = $1
    limit 1
    `,
    [input.studentUserId],
  );

  const lead = await client.query<{
    ref_code: string | null;
    referrer_user_id: string | null;
    assigned_to_user_id: string | null;
    assignment_mode: string | null;
    bound_at: string;
    assignee_name: string | null;
    assignee_username: string | null;
    assignee_role: string | null;
  }>(
    `
    select rl.ref_code::text, rl.referrer_user_id, rl.assigned_to_user_id,
           rl.assignment_mode, rl.bound_at,
           assignee.name as assignee_name, assignee.username as assignee_username,
           assignee_tm.role as assignee_role
    from public.referral_leads rl
    left join public.platform_users assignee on assignee.id = rl.assigned_to_user_id
    left join public.tenant_memberships assignee_tm on assignee_tm.tenant_id = rl.tenant_id
      and assignee_tm.user_id = rl.assigned_to_user_id
      and assignee_tm.status = 'active'
    where rl.tenant_id = $1 and rl.id = $2
    limit 1
    `,
    [input.tenantId, input.leadId],
  );

  const payload = {
    formName: config.form_name || '刷题题库',
    examType: config.exam_type || '成人本科',
    source: input.source,
    leadId: input.leadId,
    student: {
      id: input.studentUserId,
      username: student.rows[0]?.username || null,
      name: student.rows[0]?.name || null,
      phone: student.rows[0]?.phone || null,
      email: student.rows[0]?.email || null,
    },
    referral: {
      refCode: lead.rows[0]?.ref_code || null,
      referrerUserId: lead.rows[0]?.referrer_user_id || null,
      boundAt: lead.rows[0]?.bound_at || null,
    },
    assignee: lead.rows[0]?.assigned_to_user_id ? {
      id: lead.rows[0].assigned_to_user_id,
      name: lead.rows[0].assignee_name || lead.rows[0].assignee_username || null,
      role: lead.rows[0].assignee_role || null,
      assignmentMode: lead.rows[0].assignment_mode || null,
    } : null,
    metadata: input.metadata || {},
  };

  const delaySeconds = Math.max(0, Number(config.delay_sec || 60));
  const idempotencyKey = `lead:${input.leadId}`;
  const queued = await client.query(
    `
    insert into public.crm_webhook_queue (
      tenant_id, record_id, status, scheduled_at, next_attempt_at, lead_id,
      source, payload, idempotency_key, target_url
    )
    values ($1, $2, 'pending', now() + ($3::int * interval '1 second'), null, $4, $5, $6::jsonb, $7, $8)
    on conflict (tenant_id, idempotency_key) where idempotency_key is not null
    do update set status = case
                    when public.crm_webhook_queue.status = 'sent' then public.crm_webhook_queue.status
                    else 'pending'
                  end,
                  scheduled_at = excluded.scheduled_at,
                  next_attempt_at = null,
                  payload = excluded.payload,
                  target_url = excluded.target_url,
                  updated_at = now()
    returning id, status, scheduled_at as "scheduledAt", lead_id as "leadId"
    `,
    [
      input.tenantId,
      input.studentUserId,
      delaySeconds,
      input.leadId,
      input.source,
      JSON.stringify(payload),
      idempotencyKey,
      config.url,
    ],
  );

  return queued.rows[0];
}

async function bindReferralLead(
  client: pg.PoolClient,
  input: {
    tenantId: string;
    studentUserId: string;
    refCode: string;
    referrerUserId: string | null;
    source: string;
    trackId: string | null;
    bindType?: 'first_touch' | 'manual' | 'imported';
    force?: boolean;
    metadata?: Record<string, unknown>;
  },
) {
  if (input.referrerUserId && input.referrerUserId === input.studentUserId) {
    throw new HttpError(400, 'Cannot bind a lead to itself', 'SELF_REFERRAL_NOT_ALLOWED');
  }

  const existing = await client.query<{
    id: string;
    referrerUserId: string | null;
    refCode: string | null;
    assignedToUserId: string | null;
    assignmentMode: string | null;
    assignedAt: string | null;
    status: string;
    boundAt: string;
  }>(
    `
    select id, referrer_user_id as "referrerUserId", ref_code::text as "refCode",
           assigned_to_user_id as "assignedToUserId", assignment_mode as "assignmentMode",
           assigned_at as "assignedAt", status, bound_at as "boundAt"
    from public.referral_leads
    where tenant_id = $1 and student_user_id = $2
    limit 1
    `,
    [input.tenantId, input.studentUserId],
  );

  if (existing.rows[0] && !input.force) {
    return { item: existing.rows[0], bound: false, protected: true };
  }

  const result = await client.query(
    `
    insert into public.referral_leads (
      tenant_id, student_user_id, referrer_user_id, ref_code, source,
      first_track_id, bind_type, status, metadata
    )
    values ($1, $2, $3, $4::citext, $5, $6::uuid, $7, 'protected', $8::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,
                  first_track_id = coalesce(public.referral_leads.first_track_id, excluded.first_track_id),
                  bind_type = excluded.bind_type,
                  status = 'protected',
                  metadata = public.referral_leads.metadata || excluded.metadata,
                  updated_at = now()
    returning id, student_user_id as "studentUserId", referrer_user_id as "referrerUserId",
              ref_code::text as "refCode", source, bind_type as "bindType", status,
              assigned_to_user_id as "assignedToUserId", assignment_mode as "assignmentMode",
              assigned_at as "assignedAt", assigned_by as "assignedBy",
              bound_at as "boundAt", created_at as "createdAt", updated_at as "updatedAt"
    `,
    [
      input.tenantId,
      input.studentUserId,
      input.referrerUserId,
      normalizeCode(input.refCode),
      input.source,
      input.trackId,
      input.bindType || 'first_touch',
      JSON.stringify(input.metadata || {}),
    ],
  );

  if (input.trackId) {
    await client.query(
      'update public.referral_tracks set lead_id = $3, updated_at = now() where tenant_id = $1 and id = $2',
      [input.tenantId, input.trackId, result.rows[0].id],
    );
  }

  const assignment = await assignReferralLeadFromCrmConfig(client, {
    tenantId: input.tenantId,
    leadId: result.rows[0].id,
    referrerUserId: result.rows[0].referrerUserId,
    actorUserId: typeof input.metadata?.operatorUserId === 'string' ? input.metadata.operatorUserId : null,
    trigger: input.bindType || 'first_touch',
  });

  return { item: assignment?.item || result.rows[0], assignment: assignment?.assignee || null, bound: true, protected: false };
}

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);

  const item = await transaction(async client => {
    const activeMember = await userHasTenantMembership(client, tenantId, userId);
    if (!activeMember) {
      throw new HttpError(403, 'Active tenant membership is required', 'TENANT_MEMBER_REQUIRED');
    }
    const code = await createOrGetReferralCode(client, tenantId, userId, { source: 'api' });
    return { inviteCode: code };
  });

  return item;
}

export async function referralResolveRoute(ctx: RequestContext) {
  const tenantId = await tenantIdFrom(ctx);
  const body = await readJsonBody(ctx);
  const code = optionalString(body, 'code');
  if (!code) return { valid: false };

  const item = await resolveReferralCode(tenantId, code);
  if (!item) return { valid: false };

  return {
    valid: true,
    inviterId: item.userId,
    inviteCode: item.code,
    role: item.role,
    name: item.name || item.username || '推广员',
  };
}

export async function referralTrackEventRoute(ctx: RequestContext) {
  const tenantId = await tenantIdFrom(ctx);
  const body = await readJsonBody(ctx);
  const eventType = optionalChoice(body.eventType, EVENT_TYPES, 'enter');
  const source = optionalChoice(body.source, TRACK_SOURCES, 'unknown');
  const refCode = normalizeCode(requiredString(body, 'refCode'));
  const targetUserId = await optionalUserIdFrom(ctx, {
    userId: nullableString(body.targetUserId) || nullableString(body.userId) || undefined,
  }) || null;
  const referrer = await resolveReferralCode(tenantId, refCode);

  const result = await transaction(async client => {
    const track = await client.query(
      `
      insert into public.referral_tracks (
        tenant_id, event_type, ref_code, ref_user_id, target_user_id, source,
        ip_address, user_agent, metadata
      )
      values ($1, $2, $3::citext, $4, $5, $6, $7, $8, $9::jsonb)
      returning id, event_type as "eventType", ref_code::text as "refCode",
                ref_user_id as "refUserId", target_user_id as "targetUserId",
                source, created_at as "createdAt"
      `,
      [
        tenantId,
        eventType,
        refCode,
        referrer?.userId || null,
        targetUserId,
        source,
        clientIpFrom(ctx),
        userAgentFrom(ctx),
        JSON.stringify(objectValue(body.metadata)),
      ],
    );

    let lead = null;
    let crmQueue = null;
    if (targetUserId && referrer?.userId && (eventType === 'register' || eventType === 'enter' || eventType === 'scan')) {
      lead = await bindReferralLead(client, {
        tenantId,
        studentUserId: targetUserId,
        refCode,
        referrerUserId: referrer.userId,
        source,
        trackId: track.rows[0].id,
        metadata: { eventType },
      });
      if (lead.bound) {
        crmQueue = await enqueueCrmLead(client, {
          tenantId,
          leadId: lead.item.id,
          studentUserId: targetUserId,
          source,
          metadata: { trigger: 'referral.track_event', eventType },
        });
      }
    }

    return { track: track.rows[0], lead, crmQueue };
  });

  return { item: result.track, lead: result.lead, crmQueue: result.crmQueue };
}

export async function referralBindRoute(ctx: RequestContext) {
  const tenantId = await tenantIdFrom(ctx);
  const body = await readJsonBody(ctx);
  const studentUserId = await userIdFrom(ctx, body);
  const refCode = normalizeCode(requiredString(body, 'refCode'));
  const referrer = await resolveReferralCode(tenantId, refCode);
  if (!referrer) throw new HttpError(404, 'Referral code not found', 'REFERRAL_CODE_NOT_FOUND');

  const result = await transaction(async client => {
    const lead = await bindReferralLead(client, {
      tenantId,
      studentUserId,
      refCode,
      referrerUserId: referrer.userId,
      source: optionalString(body, 'source') || 'manual',
      trackId: null,
      metadata: objectValue(body.metadata),
    });
    const crmQueue = lead.bound ? await enqueueCrmLead(client, {
      tenantId,
      leadId: lead.item.id,
      studentUserId,
      source: optionalString(body, 'source') || 'manual',
      metadata: { trigger: 'referral.bind' },
    }) : null;
    return { lead, crmQueue };
  });

  return result;
}

export async function referralStatsRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  if (!canViewSelfReferral(auth)) {
    throw new HttpError(403, 'Referral access is required', 'REFERRAL_ACCESS_REQUIRED');
  }
  const requestedReferrerId = stringParam(ctx, 'referrerUserId');
  const referrerUserId = canViewAllReferral(auth) && requestedReferrerId ? requestedReferrerId : auth.userId;

  const item = await queryOne<{
    referrerUserId: string;
    leadCount: string;
    paidLeadCount: string;
    paidAmountCents: string;
    trackCount: string;
  }>(
    `
    with leads as (
      select rl.student_user_id
      from public.referral_leads rl
      where rl.tenant_id = $1 and rl.referrer_user_id = $2 and rl.status = 'protected'
    ),
    paid as (
      select o.user_id, sum(o.amount_cents)::bigint as amount_cents
      from public.orders o
      join leads l on l.student_user_id = o.user_id
      where o.tenant_id = $1 and o.status = 'paid'
      group by o.user_id
    )
    select $2::uuid as "referrerUserId",
           (select count(*) from leads)::text as "leadCount",
           (select count(*) from paid)::text as "paidLeadCount",
           coalesce((select sum(amount_cents) from paid), 0)::text as "paidAmountCents",
           (
             select count(*)
             from public.referral_tracks rt
             where rt.tenant_id = $1 and rt.ref_user_id = $2
           )::text as "trackCount"
    `,
    [auth.tenantId, referrerUserId],
  );

  const paidAmountCents = Number(item?.paidAmountCents || 0);
  return {
    item: {
      referrerUserId,
      leadCount: Number(item?.leadCount || 0),
      paidLeadCount: Number(item?.paidLeadCount || 0),
      paidAmountCents,
      paidAmount: centsToAmount(paidAmountCents),
      trackCount: Number(item?.trackCount || 0),
      conversionRate: Number(item?.leadCount || 0) > 0
        ? Math.round((Number(item?.paidLeadCount || 0) / Number(item?.leadCount || 0)) * 100)
        : 0,
    },
  };
}

export async function referralSalesStatsRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  if (!canViewSelfReferral(auth)) {
    throw new HttpError(403, 'Referral access is required', 'REFERRAL_ACCESS_REQUIRED');
  }
  const limit = intParam(ctx, 'limit', 100, 500);
  const params: unknown[] = [auth.tenantId];
  let scopeSql = '';
  if (!canViewAllReferral(auth)) {
    params.push(auth.userId);
    scopeSql = `and tm.user_id = $${params.length}`;
  }
  params.push(limit);

  const items = await query<{
    referrerUserId: string;
    role: string;
    username: string | null;
    name: string | null;
    phone: string | null;
    inviteCode: string | null;
    leadCount: number;
    paidLeadCount: number;
    paidAmountCents: string;
    trackCount: number;
  }>(
    `
    with referrers as (
      select tm.user_id, tm.role, u.username, u.name, u.phone,
             rc.code::text as code
      from public.tenant_memberships tm
      join public.platform_users u on u.id = tm.user_id
      left join public.referral_codes rc on rc.tenant_id = tm.tenant_id and rc.user_id = tm.user_id
      where tm.tenant_id = $1
        and tm.status = 'active'
        and tm.role in ('tenant_owner', 'tenant_admin', 'tenant_operator', 'teacher', 'sales', 'agent')
        ${scopeSql}
    ),
    lead_stats as (
      select referrer_user_id,
             count(*)::int as lead_count,
             count(*) filter (where assigned_to_user_id is not null)::int as assigned_lead_count,
             count(*) filter (where exists (
               select 1 from public.orders o
               where o.tenant_id = rl.tenant_id
                 and o.user_id = rl.student_user_id
                 and o.status = 'paid'
             ))::int as paid_lead_count
      from public.referral_leads rl
      where rl.tenant_id = $1 and rl.status = 'protected'
      group by referrer_user_id
    ),
    order_stats as (
      select rl.referrer_user_id, coalesce(sum(o.amount_cents), 0)::bigint as paid_amount_cents
      from public.referral_leads rl
      join public.orders o on o.tenant_id = rl.tenant_id and o.user_id = rl.student_user_id and o.status = 'paid'
      where rl.tenant_id = $1 and rl.status = 'protected'
      group by rl.referrer_user_id
    ),
    track_stats as (
      select ref_user_id, count(*)::int as track_count
      from public.referral_tracks
      where tenant_id = $1
      group by ref_user_id
    )
    select r.user_id as "referrerUserId", r.role, r.username, r.name, r.phone,
           r.code as "inviteCode",
           coalesce(ls.lead_count, 0) as "leadCount",
           coalesce(ls.assigned_lead_count, 0) as "assignedLeadCount",
           coalesce(ls.paid_lead_count, 0) as "paidLeadCount",
           coalesce(os.paid_amount_cents, 0)::text as "paidAmountCents",
           coalesce(ts.track_count, 0) as "trackCount"
    from referrers r
    left join lead_stats ls on ls.referrer_user_id = r.user_id
    left join order_stats os on os.referrer_user_id = r.user_id
    left join track_stats ts on ts.ref_user_id = r.user_id
    order by coalesce(ls.lead_count, 0) desc, r.name asc nulls last
    limit $${params.length}
    `,
    params,
  );

  return {
    items: items.map((item: Record<string, unknown>) => ({
      ...item,
      paidAmount: centsToAmount(Number(item.paidAmountCents || 0)),
      conversionRate: Number(item.leadCount || 0) > 0
        ? Math.round((Number(item.paidLeadCount || 0) / Number(item.leadCount || 0)) * 100)
        : 0,
    })),
  };
}

export async function referralConversionReportRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  if (!canViewSelfReferral(auth)) {
    throw new HttpError(403, 'Referral access is required', 'REFERRAL_ACCESS_REQUIRED');
  }

  const { startDate, endDate, days } = conversionPeriodFromParams(ctx);
  const requestedReferrerId = optionalUuidString(stringParam(ctx, 'referrerUserId'), 'referrerUserId') || null;
  const scopedReferrerId = restrictReferralScope(auth, requestedReferrerId);
  const limit = intParam(ctx, 'limit', 20, 100);
  const params = [auth.tenantId, startDate, endDate, scopedReferrerId, limit];

  const byReferrerRows = await query<Record<string, unknown>>(
    `
    with settings as (
      select coalesce(default_rate, 0.2000)::numeric(6,4) as default_rate
      from public.tenant_commission_settings
      where tenant_id = $1
      union all
      select 0.2000::numeric(6,4)
      limit 1
    ),
    referrers as (
      select distinct on (tm.user_id)
             tm.user_id, tm.role, u.username, u.name, u.phone, rc.code::text as invite_code
      from public.tenant_memberships tm
      join public.platform_users u on u.id = tm.user_id
      left join public.referral_codes rc on rc.tenant_id = tm.tenant_id and rc.user_id = tm.user_id
      where tm.tenant_id = $1
        and tm.status = 'active'
        and tm.role in ('tenant_owner', 'tenant_admin', 'tenant_operator', 'teacher', 'sales', 'agent')
        and ($4::uuid is null or tm.user_id = $4::uuid)
      order by tm.user_id,
               case tm.role
                 when 'sales' then 1
                 when 'agent' then 2
                 when 'tenant_operator' then 3
                 when 'teacher' then 4
                 else 9
               end
    ),
    lead_base as (
      select rl.*
      from public.referral_leads rl
      where rl.tenant_id = $1
        and rl.status = 'protected'
        and rl.referrer_user_id is not null
        and ($4::uuid is null or rl.referrer_user_id = $4::uuid)
    ),
    new_leads as (
      select *
      from lead_base
      where bound_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and bound_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
    ),
    order_sources as (
      select 'order'::text as source_type,
             o.id as source_id,
             coalesce(o.paid_at, o.updated_at, o.created_at) as source_paid_at,
             o.user_id as student_user_id,
             lb.referrer_user_id,
             o.amount_cents::integer as gross_amount_cents,
             null::numeric(6,4) as batch_rate
      from public.orders o
      join lead_base lb on lb.tenant_id = o.tenant_id
       and lb.student_user_id = o.user_id
       and coalesce(o.paid_at, o.updated_at, o.created_at) >= lb.bound_at
      where o.tenant_id = $1
        and o.status = 'paid'
        and o.user_id is not null
        and coalesce(o.paid_at, o.updated_at, o.created_at) >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and coalesce(o.paid_at, o.updated_at, o.created_at) < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
    ),
    code_sources as (
      select 'activation_code'::text as source_type,
             ac.id as source_id,
             ac.used_at as source_paid_at,
             ac.used_by as student_user_id,
             lb.referrer_user_id,
             greatest(coalesce(ac.unit_price_cents, cb.default_unit_price_cents, 0), 0)::integer as gross_amount_cents,
             cb.commission_rate as batch_rate
      from public.activation_codes ac
      left join public.code_batches cb on cb.tenant_id = ac.tenant_id and cb.id = ac.batch_id
      join lead_base lb on lb.tenant_id = ac.tenant_id
       and lb.student_user_id = ac.used_by
       and ac.used_at >= lb.bound_at
      where ac.tenant_id = $1
        and ac.is_used is true
        and ac.used_by is not null
        and ac.used_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and ac.used_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
    ),
    sources as (
      select * from order_sources
      union all
      select * from code_sources
    ),
    referrer_members as (
      select distinct on (tm.user_id)
             tm.user_id, tm.commission_rate
      from public.tenant_memberships tm
      where tm.tenant_id = $1
        and tm.status = 'active'
        and tm.role in ('tenant_owner', 'tenant_admin', 'tenant_operator', 'teacher', 'sales', 'agent')
      order by tm.user_id,
               case tm.role
                 when 'sales' then 1
                 when 'agent' then 2
                 when 'tenant_operator' then 3
                 when 'teacher' then 4
                 else 9
               end
    ),
    commission_sources as (
      select s.*,
             coalesce(s.batch_rate, rm.commission_rate, settings.default_rate)::numeric(6,4) as commission_rate,
             round(s.gross_amount_cents * coalesce(s.batch_rate, rm.commission_rate, settings.default_rate))::integer as commission_amount_cents
      from sources s
      cross join settings
      left join referrer_members rm on rm.user_id = s.referrer_user_id
      where s.gross_amount_cents > 0
    ),
    lead_stats as (
      select referrer_user_id,
             count(*)::int as lead_count,
             count(*) filter (where assigned_to_user_id is not null)::int as assigned_lead_count
      from new_leads
      group by referrer_user_id
    ),
    converted_stats as (
      select nl.referrer_user_id,
             count(distinct nl.student_user_id)::int as converted_lead_count
      from new_leads nl
      where exists (
        select 1
        from commission_sources cs
        where cs.referrer_user_id = nl.referrer_user_id
          and cs.student_user_id = nl.student_user_id
          and cs.source_paid_at >= nl.bound_at
      )
      group by nl.referrer_user_id
    ),
    source_stats as (
      select referrer_user_id,
             count(*)::int as paid_source_count,
             count(distinct student_user_id)::int as paid_lead_count,
             coalesce(sum(gross_amount_cents), 0)::bigint as gross_amount_cents,
             coalesce(sum(commission_amount_cents), 0)::bigint as commission_amount_cents
      from commission_sources
      group by referrer_user_id
    ),
    paid_users as (
      select lb.referrer_user_id, lb.student_user_id,
             min(lb.bound_at) as bound_at,
             min(cs.source_paid_at) as first_paid_at
      from lead_base lb
      join commission_sources cs on cs.referrer_user_id = lb.referrer_user_id
       and cs.student_user_id = lb.student_user_id
       and cs.source_paid_at >= lb.bound_at
      group by lb.referrer_user_id, lb.student_user_id
    ),
    paid_user_stats as (
      select referrer_user_id,
             count(*)::int as first_pay_user_count,
             coalesce(sum(extract(epoch from (first_paid_at - bound_at)) / 3600), 0)::numeric as first_pay_hours_sum,
             coalesce(avg(extract(epoch from (first_paid_at - bound_at)) / 3600), 0)::numeric as avg_first_pay_hours
      from paid_users
      group by referrer_user_id
    ),
    track_stats as (
      select ref_user_id as referrer_user_id,
             count(*)::int as track_count,
             count(*) filter (where event_type = 'share')::int as share_count,
             count(*) filter (where event_type in ('scan', 'enter', 'register'))::int as intent_count
      from public.referral_tracks
      where tenant_id = $1
        and ref_user_id is not null
        and created_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and created_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
        and ($4::uuid is null or ref_user_id = $4::uuid)
      group by ref_user_id
    ),
    crm_stats as (
      select lb.referrer_user_id,
             count(*)::int as crm_queue_count,
             count(*) filter (where q.status = 'sent')::int as crm_sent_count,
             count(*) filter (where q.status in ('failed', 'discarded'))::int as crm_failed_count,
             count(*) filter (where q.status in ('pending', 'processing', 'retrying'))::int as crm_pending_count
      from public.crm_webhook_queue q
      join lead_base lb on lb.id::text = q.lead_id
      where q.tenant_id = $1
        and q.created_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and q.created_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
      group by lb.referrer_user_id
    ),
    followup_stats as (
      select assigned_to_user_id as referrer_user_id,
             count(*)::int as followup_count,
             count(*) filter (where status in ('open', 'in_progress'))::int as open_followup_count,
             count(*) filter (where status = 'done')::int as done_followup_count,
             count(*) filter (where due_at < now() and status in ('open', 'in_progress'))::int as overdue_followup_count
      from public.tenant_student_followups
      where tenant_id = $1
        and assigned_to_user_id is not null
        and created_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and created_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
        and ($4::uuid is null or assigned_to_user_id = $4::uuid)
      group by assigned_to_user_id
    )
    select r.user_id as "referrerUserId",
           r.role,
           r.username,
           r.name,
           r.phone,
           r.invite_code as "inviteCode",
           coalesce(ts.track_count, 0)::int as "trackCount",
           coalesce(ts.share_count, 0)::int as "shareCount",
           coalesce(ts.intent_count, 0)::int as "intentCount",
           coalesce(ls.lead_count, 0)::int as "leadCount",
           coalesce(ls.assigned_lead_count, 0)::int as "assignedLeadCount",
           coalesce(cs.converted_lead_count, 0)::int as "convertedLeadCount",
           coalesce(ss.paid_lead_count, 0)::int as "paidLeadCount",
           coalesce(ss.paid_source_count, 0)::int as "paidSourceCount",
           coalesce(ss.gross_amount_cents, 0)::text as "grossAmountCents",
           coalesce(ss.commission_amount_cents, 0)::text as "commissionAmountCents",
           coalesce(pus.first_pay_user_count, 0)::int as "firstPayUserCount",
           coalesce(pus.first_pay_hours_sum, 0)::text as "firstPayHoursSum",
           coalesce(pus.avg_first_pay_hours, 0)::text as "avgFirstPayHours",
           coalesce(crm.crm_queue_count, 0)::int as "crmQueueCount",
           coalesce(crm.crm_sent_count, 0)::int as "crmSentCount",
           coalesce(crm.crm_failed_count, 0)::int as "crmFailedCount",
           coalesce(crm.crm_pending_count, 0)::int as "crmPendingCount",
           coalesce(fs.followup_count, 0)::int as "followupCount",
           coalesce(fs.open_followup_count, 0)::int as "openFollowupCount",
           coalesce(fs.done_followup_count, 0)::int as "doneFollowupCount",
           coalesce(fs.overdue_followup_count, 0)::int as "overdueFollowupCount"
    from referrers r
    left join lead_stats ls on ls.referrer_user_id = r.user_id
    left join converted_stats cs on cs.referrer_user_id = r.user_id
    left join source_stats ss on ss.referrer_user_id = r.user_id
    left join paid_user_stats pus on pus.referrer_user_id = r.user_id
    left join track_stats ts on ts.referrer_user_id = r.user_id
    left join crm_stats crm on crm.referrer_user_id = r.user_id
    left join followup_stats fs on fs.referrer_user_id = r.user_id
    order by coalesce(ss.gross_amount_cents, 0) desc,
             coalesce(ls.lead_count, 0) desc,
             r.name asc nulls last
    limit $5
    `,
    params,
  );

  const dailyRows = await query<Record<string, unknown>>(
    `
    with days as (
      select generate_series($2::date, $3::date, interval '1 day')::date as day
    ),
    settings as (
      select coalesce(default_rate, 0.2000)::numeric(6,4) as default_rate
      from public.tenant_commission_settings
      where tenant_id = $1
      union all
      select 0.2000::numeric(6,4)
      limit 1
    ),
    lead_base as (
      select rl.*
      from public.referral_leads rl
      where rl.tenant_id = $1
        and rl.status = 'protected'
        and rl.referrer_user_id is not null
        and ($4::uuid is null or rl.referrer_user_id = $4::uuid)
    ),
    referrer_members as (
      select distinct on (tm.user_id)
             tm.user_id, tm.commission_rate
      from public.tenant_memberships tm
      where tm.tenant_id = $1
        and tm.status = 'active'
      order by tm.user_id,
               case tm.role
                 when 'sales' then 1
                 when 'agent' then 2
                 when 'tenant_operator' then 3
                 when 'teacher' then 4
                 else 9
               end
    ),
    lead_daily as (
      select timezone('Asia/Shanghai', bound_at)::date as day,
             count(*)::int as lead_count,
             count(*) filter (where assigned_to_user_id is not null)::int as assigned_lead_count
      from lead_base
      where bound_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and bound_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
      group by timezone('Asia/Shanghai', bound_at)::date
    ),
    track_daily as (
      select timezone('Asia/Shanghai', created_at)::date as day,
             count(*)::int as track_count
      from public.referral_tracks
      where tenant_id = $1
        and ref_user_id is not null
        and created_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and created_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
        and ($4::uuid is null or ref_user_id = $4::uuid)
      group by timezone('Asia/Shanghai', created_at)::date
    ),
    order_sources as (
      select coalesce(o.paid_at, o.updated_at, o.created_at) as source_paid_at,
             o.user_id as student_user_id,
             lb.referrer_user_id,
             o.amount_cents::integer as gross_amount_cents,
             null::numeric(6,4) as batch_rate
      from public.orders o
      join lead_base lb on lb.tenant_id = o.tenant_id
       and lb.student_user_id = o.user_id
       and coalesce(o.paid_at, o.updated_at, o.created_at) >= lb.bound_at
      where o.tenant_id = $1
        and o.status = 'paid'
        and o.user_id is not null
        and coalesce(o.paid_at, o.updated_at, o.created_at) >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and coalesce(o.paid_at, o.updated_at, o.created_at) < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
    ),
    code_sources as (
      select ac.used_at as source_paid_at,
             ac.used_by as student_user_id,
             lb.referrer_user_id,
             greatest(coalesce(ac.unit_price_cents, cb.default_unit_price_cents, 0), 0)::integer as gross_amount_cents,
             cb.commission_rate as batch_rate
      from public.activation_codes ac
      left join public.code_batches cb on cb.tenant_id = ac.tenant_id and cb.id = ac.batch_id
      join lead_base lb on lb.tenant_id = ac.tenant_id
       and lb.student_user_id = ac.used_by
       and ac.used_at >= lb.bound_at
      where ac.tenant_id = $1
        and ac.is_used is true
        and ac.used_by is not null
        and ac.used_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and ac.used_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
    ),
    commission_sources as (
      select s.source_paid_at,
             s.student_user_id,
             s.referrer_user_id,
             s.gross_amount_cents,
             round(s.gross_amount_cents * coalesce(s.batch_rate, rm.commission_rate, settings.default_rate))::integer as commission_amount_cents
      from (
        select * from order_sources
        union all
        select * from code_sources
      ) s
      cross join settings
      left join referrer_members rm on rm.user_id = s.referrer_user_id
      where s.gross_amount_cents > 0
    ),
    source_daily as (
      select timezone('Asia/Shanghai', source_paid_at)::date as day,
             count(*)::int as paid_source_count,
             count(distinct student_user_id)::int as paid_lead_count,
             coalesce(sum(gross_amount_cents), 0)::bigint as gross_amount_cents,
             coalesce(sum(commission_amount_cents), 0)::bigint as commission_amount_cents
      from commission_sources
      group by timezone('Asia/Shanghai', source_paid_at)::date
    )
    select d.day::text as date,
           coalesce(td.track_count, 0)::int as "trackCount",
           coalesce(ld.lead_count, 0)::int as "leadCount",
           coalesce(ld.assigned_lead_count, 0)::int as "assignedLeadCount",
           coalesce(sd.paid_lead_count, 0)::int as "paidLeadCount",
           coalesce(sd.paid_source_count, 0)::int as "paidSourceCount",
           coalesce(sd.gross_amount_cents, 0)::text as "grossAmountCents",
           coalesce(sd.commission_amount_cents, 0)::text as "commissionAmountCents"
    from days d
    left join track_daily td on td.day = d.day
    left join lead_daily ld on ld.day = d.day
    left join source_daily sd on sd.day = d.day
    order by d.day asc
    `,
    [auth.tenantId, startDate, endDate, scopedReferrerId],
  );

  const recentUnconvertedRows = await query<Record<string, unknown>>(
    `
    with scoped_leads as (
      select rl.*
      from public.referral_leads rl
      where rl.tenant_id = $1
        and rl.status = 'protected'
        and rl.referrer_user_id is not null
        and rl.bound_at >= ($2::date::timestamp at time zone 'Asia/Shanghai')
        and rl.bound_at < (($3::date + 1)::timestamp at time zone 'Asia/Shanghai')
        and ($4::uuid is null or rl.referrer_user_id = $4::uuid)
    )
    select rl.id,
           rl.student_user_id as "studentUserId",
           student.name as "studentName",
           student.phone as "studentPhone",
           rl.referrer_user_id as "referrerUserId",
           referrer.name as "referrerName",
           referrer.phone as "referrerPhone",
           rl.assigned_to_user_id as "assignedToUserId",
           assignee.name as "assignedToName",
           rl.ref_code::text as "refCode",
           rl.source,
           rl.bound_at as "boundAt"
    from scoped_leads rl
    join public.platform_users student on student.id = rl.student_user_id
    left join public.platform_users referrer on referrer.id = rl.referrer_user_id
    left join public.platform_users assignee on assignee.id = rl.assigned_to_user_id
    where not exists (
      select 1
      from public.orders o
      where o.tenant_id = rl.tenant_id
        and o.user_id = rl.student_user_id
        and o.status = 'paid'
        and coalesce(o.paid_at, o.updated_at, o.created_at) >= rl.bound_at
    )
      and not exists (
        select 1
        from public.activation_codes ac
        where ac.tenant_id = rl.tenant_id
          and ac.used_by = rl.student_user_id
          and ac.is_used is true
          and ac.used_at >= rl.bound_at
      )
    order by rl.bound_at desc
    limit $5
    `,
    params,
  );

  const byReferrer = byReferrerRows.map(row => {
    const leadCount = intValue(row.leadCount);
    const convertedLeadCount = intValue(row.convertedLeadCount);
    const paidLeadCount = intValue(row.paidLeadCount);
    const grossAmountCents = intValue(row.grossAmountCents);
    const commissionAmountCents = intValue(row.commissionAmountCents);
    return {
      referrerUserId: String(row.referrerUserId),
      role: String(row.role || ''),
      username: row.username ? String(row.username) : null,
      name: row.name ? String(row.name) : null,
      phone: row.phone ? String(row.phone) : null,
      inviteCode: row.inviteCode ? String(row.inviteCode) : null,
      trackCount: intValue(row.trackCount),
      shareCount: intValue(row.shareCount),
      intentCount: intValue(row.intentCount),
      leadCount,
      assignedLeadCount: intValue(row.assignedLeadCount),
      convertedLeadCount,
      paidLeadCount,
      paidSourceCount: intValue(row.paidSourceCount),
      grossAmountCents,
      grossAmount: centsToAmount(grossAmountCents),
      commissionAmountCents,
      commissionAmount: centsToAmount(commissionAmountCents),
      firstPayUserCount: intValue(row.firstPayUserCount),
      firstPayHoursSum: Number(row.firstPayHoursSum || 0),
      avgFirstPayHours: Number(Number(row.avgFirstPayHours || 0).toFixed(2)),
      crmQueueCount: intValue(row.crmQueueCount),
      crmSentCount: intValue(row.crmSentCount),
      crmFailedCount: intValue(row.crmFailedCount),
      crmPendingCount: intValue(row.crmPendingCount),
      followupCount: intValue(row.followupCount),
      openFollowupCount: intValue(row.openFollowupCount),
      doneFollowupCount: intValue(row.doneFollowupCount),
      overdueFollowupCount: intValue(row.overdueFollowupCount),
      conversionRate: ratio(convertedLeadCount, leadCount),
      paidLeadRate: ratio(paidLeadCount, leadCount),
    };
  });

  const summaryBase = byReferrer.reduce((acc, item) => {
    acc.trackCount += item.trackCount;
    acc.shareCount += item.shareCount;
    acc.intentCount += item.intentCount;
    acc.leadCount += item.leadCount;
    acc.assignedLeadCount += item.assignedLeadCount;
    acc.convertedLeadCount += item.convertedLeadCount;
    acc.paidLeadCount += item.paidLeadCount;
    acc.paidSourceCount += item.paidSourceCount;
    acc.grossAmountCents += item.grossAmountCents;
    acc.commissionAmountCents += item.commissionAmountCents;
    acc.firstPayUserCount += item.firstPayUserCount;
    acc.firstPayHoursSum += item.firstPayHoursSum;
    acc.crmQueueCount += item.crmQueueCount;
    acc.crmSentCount += item.crmSentCount;
    acc.crmFailedCount += item.crmFailedCount;
    acc.crmPendingCount += item.crmPendingCount;
    acc.followupCount += item.followupCount;
    acc.openFollowupCount += item.openFollowupCount;
    acc.doneFollowupCount += item.doneFollowupCount;
    acc.overdueFollowupCount += item.overdueFollowupCount;
    return acc;
  }, {
    trackCount: 0,
    shareCount: 0,
    intentCount: 0,
    leadCount: 0,
    assignedLeadCount: 0,
    convertedLeadCount: 0,
    paidLeadCount: 0,
    paidSourceCount: 0,
    grossAmountCents: 0,
    commissionAmountCents: 0,
    firstPayUserCount: 0,
    firstPayHoursSum: 0,
    crmQueueCount: 0,
    crmSentCount: 0,
    crmFailedCount: 0,
    crmPendingCount: 0,
    followupCount: 0,
    openFollowupCount: 0,
    doneFollowupCount: 0,
    overdueFollowupCount: 0,
  });

  return {
    item: {
      tenantId: auth.tenantId,
      range: { startDate, endDate, days, referrerUserId: scopedReferrerId },
      summary: {
        ...summaryBase,
        grossAmount: centsToAmount(summaryBase.grossAmountCents),
        commissionAmount: centsToAmount(summaryBase.commissionAmountCents),
        conversionRate: ratio(summaryBase.convertedLeadCount, summaryBase.leadCount),
        paidLeadRate: ratio(summaryBase.paidLeadCount, summaryBase.leadCount),
        avgFirstPayHours: summaryBase.firstPayUserCount
          ? Number((summaryBase.firstPayHoursSum / summaryBase.firstPayUserCount).toFixed(2))
          : 0,
      },
      funnel: [
        { key: 'tracks', label: '触达/扫码', count: summaryBase.trackCount },
        { key: 'leads', label: '新增客资', count: summaryBase.leadCount },
        { key: 'assigned', label: '已分配跟进', count: summaryBase.assignedLeadCount },
        { key: 'converted', label: '有效成交客资', count: summaryBase.convertedLeadCount },
        { key: 'paid_sources', label: '成交来源', count: summaryBase.paidSourceCount },
      ],
      byReferrer,
      dailyTrend: dailyRows.map(row => ({
        date: String(row.date),
        trackCount: intValue(row.trackCount),
        leadCount: intValue(row.leadCount),
        assignedLeadCount: intValue(row.assignedLeadCount),
        paidLeadCount: intValue(row.paidLeadCount),
        paidSourceCount: intValue(row.paidSourceCount),
        grossAmountCents: intValue(row.grossAmountCents),
        commissionAmountCents: intValue(row.commissionAmountCents),
      })),
      recentUnconvertedLeads: recentUnconvertedRows.map(row => ({
        id: String(row.id),
        studentUserId: String(row.studentUserId),
        studentName: row.studentName ? String(row.studentName) : null,
        studentPhone: row.studentPhone ? String(row.studentPhone) : null,
        referrerUserId: String(row.referrerUserId),
        referrerName: row.referrerName ? String(row.referrerName) : null,
        referrerPhone: row.referrerPhone ? String(row.referrerPhone) : null,
        assignedToUserId: row.assignedToUserId ? String(row.assignedToUserId) : null,
        assignedToName: row.assignedToName ? String(row.assignedToName) : null,
        refCode: row.refCode ? String(row.refCode) : null,
        source: row.source ? String(row.source) : null,
        boundAt: row.boundAt,
      })),
    },
  };
}

export async function referralSalesClientsRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  if (!canViewSelfReferral(auth)) {
    throw new HttpError(403, 'Referral access is required', 'REFERRAL_ACCESS_REQUIRED');
  }
  const requestedReferrerId = stringParam(ctx, 'referrerUserId');
  const referrerUserId = canViewAllReferral(auth) && requestedReferrerId ? requestedReferrerId : auth.userId;
  const limit = intParam(ctx, 'limit', 100, 500);

  const items = await query<{
    id: string;
    studentUserId: string;
    username: string | null;
    name: string | null;
    phone: string | null;
    email: string | null;
    referrerUserId: string;
    refCode: string | null;
    source: string | null;
    bindType: string;
    status: string;
    assignedToUserId: string | null;
    assignedToName: string | null;
    assignmentMode: string | null;
    assignedAt: string | null;
    boundAt: string;
    paidAmountCents: string;
    lastPaidAt: string | null;
  }>(
    `
    select rl.id, rl.student_user_id as "studentUserId",
           u.username, u.name, u.phone, u.email::text as email,
           rl.referrer_user_id as "referrerUserId", rl.ref_code::text as "refCode",
           rl.source, rl.bind_type as "bindType", rl.status,
           rl.assigned_to_user_id as "assignedToUserId",
           coalesce(assignee.name, assignee.username) as "assignedToName",
           rl.assignment_mode as "assignmentMode", rl.assigned_at as "assignedAt",
           rl.bound_at as "boundAt",
           coalesce(sum(o.amount_cents) filter (where o.status = 'paid'), 0)::text as "paidAmountCents",
           max(o.paid_at) filter (where o.status = 'paid') as "lastPaidAt"
    from public.referral_leads rl
    join public.platform_users u on u.id = rl.student_user_id
    left join public.platform_users assignee on assignee.id = rl.assigned_to_user_id
    left join public.orders o on o.tenant_id = rl.tenant_id and o.user_id = rl.student_user_id
    where rl.tenant_id = $1
      and rl.referrer_user_id = $2
      and rl.status = 'protected'
    group by rl.id, u.id, assignee.id
    order by rl.bound_at desc
    limit $3
    `,
    [auth.tenantId, referrerUserId, limit],
  );

  return {
    items: items.map((item: Record<string, unknown>) => ({
      ...item,
      paidAmount: centsToAmount(Number(item.paidAmountCents || 0)),
    })),
  };
}

export async function referralManualBindRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  requireTenantPermission(auth, 'referral:write');
  const body = await readJsonBody(ctx);
  const studentUserId = requiredString(body, 'studentUserId');
  const referrerUserId = requiredString(body, 'referrerUserId');

  const code = await transaction(async client => {
    const member = await userHasTenantMembership(client, auth.tenantId, referrerUserId);
    if (!member) throw new HttpError(400, 'Referrer is not an active tenant member', 'REFERRER_NOT_TENANT_MEMBER');
    return createOrGetReferralCode(client, auth.tenantId, referrerUserId, { source: 'manual_bind' });
  });

  const result = await transaction(async client => {
    const lead = await bindReferralLead(client, {
      tenantId: auth.tenantId,
      studentUserId,
      refCode: code,
      referrerUserId,
      source: optionalString(body, 'source') || 'manual',
      trackId: null,
      bindType: 'manual',
      force: body.force === true,
      metadata: { operatorUserId: auth.userId, ...objectValue(body.metadata) },
    });
    const crmQueue = lead.bound ? await enqueueCrmLead(client, {
      tenantId: auth.tenantId,
      leadId: lead.item.id,
      studentUserId,
      source: optionalString(body, 'source') || 'manual',
      metadata: { trigger: 'referral.manual_bind', operatorUserId: auth.userId },
    }) : null;
    return { lead, crmQueue };
  });

  return result;
}

export async function referralTeamRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  if (!canViewSelfReferral(auth)) {
    throw new HttpError(403, 'Referral access is required', 'REFERRAL_ACCESS_REQUIRED');
  }
  const requestedLeaderId = stringParam(ctx, 'leaderUserId');
  const leaderUserId = canViewAllReferral(auth) && requestedLeaderId ? requestedLeaderId : auth.userId;

  const items = await query(
    `
    select rte.id, rte.member_user_id as "memberUserId", rte.leader_user_id as "leaderUserId",
           rte.relation_type as "relationType", rte.status, rte.metadata,
           u.username, u.name, u.phone,
           tm.role
    from public.referral_team_edges rte
    join public.platform_users u on u.id = rte.member_user_id
    left join public.tenant_memberships tm on tm.tenant_id = rte.tenant_id
      and tm.user_id = rte.member_user_id
      and tm.status = 'active'
    where rte.tenant_id = $1
      and rte.leader_user_id = $2
      and rte.status = 'active'
    order by rte.created_at asc
    `,
    [auth.tenantId, leaderUserId],
  );

  return { items };
}

export async function upsertReferralTeamRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  requireTenantPermission(auth, 'referral:write');
  const body = await readJsonBody(ctx);
  const memberUserId = requiredString(body, 'memberUserId');
  const leaderUserId = nullableString(body.leaderUserId);
  const relationType = optionalChoice(body.relationType, ['sales_team', 'agent_network', 'teacher_class'], 'sales_team');
  const status = optionalChoice(body.status, ['active', 'disabled'], 'active');

  const item = await transaction(async client => {
    const member = await userHasTenantMembership(client, auth.tenantId, memberUserId);
    if (!member) throw new HttpError(400, 'Member is not an active tenant member', 'MEMBER_NOT_TENANT_MEMBER');
    if (leaderUserId) {
      const leader = await userHasTenantMembership(client, auth.tenantId, leaderUserId);
      if (!leader) throw new HttpError(400, 'Leader is not an active tenant member', 'LEADER_NOT_TENANT_MEMBER');
      if (leaderUserId === memberUserId) throw new HttpError(400, 'Leader cannot be the member itself', 'SELF_LEADER_NOT_ALLOWED');
    }

    const result = await client.query(
      `
      insert into public.referral_team_edges (
        tenant_id, member_user_id, leader_user_id, relation_type, status, metadata
      )
      values ($1, $2, $3, $4, $5, $6::jsonb)
      on conflict (tenant_id, member_user_id, relation_type)
      do update set leader_user_id = excluded.leader_user_id,
                    status = excluded.status,
                    metadata = excluded.metadata,
                    updated_at = now()
      returning id, member_user_id as "memberUserId", leader_user_id as "leaderUserId",
                relation_type as "relationType", status, metadata, updated_at as "updatedAt"
      `,
      [auth.tenantId, memberUserId, leaderUserId, relationType, status, JSON.stringify(objectValue(body.metadata))],
    );

    return result.rows[0];
  });

  return { item };
}

export async function referralQrcodeRoute(ctx: RequestContext) {
  const tenantId = await tenantIdFrom(ctx);
  const userId = await userIdFrom(ctx);
  const body = await readJsonBody(ctx);
  const page = optionalString(body, 'page') || 'pages/index/index';
  const provider = optionalString(body, 'provider') || 'wechat-miniapp';

  const item = await transaction(async client => {
    const activeMember = await userHasTenantMembership(client, tenantId, userId);
    if (!activeMember) throw new HttpError(403, 'Active tenant membership is required', 'TENANT_MEMBER_REQUIRED');
    const refCode = await createOrGetReferralCode(client, tenantId, userId, { source: 'qrcode' });
    const scene = optionalString(body, 'scene') || `ref_${refCode}`;
    const result = await client.query(
      `
      insert into public.referral_qrcodes (tenant_id, user_id, ref_code, scene, page, provider, qrcode_url, status, metadata)
      values ($1, $2, $3::citext, $4, $5, $6, $7, 'ready', $8::jsonb)
      on conflict (tenant_id, provider, scene, page)
      do update set user_id = excluded.user_id,
                    ref_code = excluded.ref_code,
                    qrcode_url = coalesce(public.referral_qrcodes.qrcode_url, excluded.qrcode_url),
                    status = 'ready',
                    metadata = public.referral_qrcodes.metadata || excluded.metadata,
                    updated_at = now()
      returning id, ref_code::text as "refCode", scene, page, provider,
                qrcode_url as "qrcodeUrl", status, created_at as "createdAt", updated_at as "updatedAt"
      `,
      [
        tenantId,
        userId,
        refCode,
        scene,
        page,
        provider,
        optionalString(body, 'qrcodeUrl') || `miniapp://${page}?scene=${encodeURIComponent(scene)}`,
        JSON.stringify({ generatedBy: 'local-placeholder', ...objectValue(body.metadata) }),
      ],
    );
    return result.rows[0];
  });

  return { item };
}

export async function crmConfigRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  crmPermission(auth);
  const item = await queryOne(
    `
    select id, enabled, url, secret_ref as "secretRef", form_name as "formName",
           exam_type as "examType", timeout_sec as "timeoutSec", delay_sec as "delaySec",
           assignment_mode as "assignmentMode",
           assignment_pool as "assignmentPool",
           assignment_cursor as "assignmentCursor",
           assignment_config as "assignmentConfig",
           created_at as "createdAt", updated_at as "updatedAt"
    from public.crm_config
    where tenant_id = $1
    limit 1
    `,
    [auth.tenantId],
  );
  return { item };
}

export async function upsertCrmConfigRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  requireTenantPermission(auth, 'crm:write');
  const body = await readJsonBody(ctx);
  const secretRef = optionalString(body, 'secretRef') || (body.secret ? 'app_private.tenant_secrets:crm:webhook' : null);
  const assignmentMode = optionalChoice(body.assignmentMode, CRM_ASSIGNMENT_MODES, 'none');
  const assignmentPool = stringArray(body.assignmentPool);
  const assignmentConfig = objectValue(body.assignmentConfig);

  const item = await transaction(async client => {
    const validAssignmentPool = await validateCrmAssignmentPool(client, auth.tenantId, assignmentPool, assignmentMode);
    if (body.secret && typeof body.secret === 'string') {
      await client.query(
        `
        insert into app_private.tenant_secrets (tenant_id, secret_scope, secret_key, secret_value, provider, last_rotated_at)
        values ($1, 'crm', 'webhook', $2, '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()
        `,
        [auth.tenantId, body.secret],
      );
    }

    const result = await client.query(
      `
      insert into public.crm_config (
        tenant_id, enabled, url, secret_ref, form_name, exam_type, timeout_sec, delay_sec,
        assignment_mode, assignment_pool, assignment_config
      )
      values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb)
      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,
                    assignment_mode = excluded.assignment_mode,
                    assignment_pool = excluded.assignment_pool,
                    assignment_config = excluded.assignment_config,
                    updated_at = now()
      returning id, enabled, url, secret_ref as "secretRef", form_name as "formName",
                exam_type as "examType", timeout_sec as "timeoutSec", delay_sec as "delaySec",
                assignment_mode as "assignmentMode", assignment_pool as "assignmentPool",
                assignment_cursor as "assignmentCursor", assignment_config as "assignmentConfig",
                updated_at as "updatedAt"
      `,
      [
        auth.tenantId,
        body.enabled === true,
        nullableString(body.url),
        secretRef,
        optionalString(body, 'formName') || '刷题题库',
        optionalString(body, 'examType') || '成人本科',
        Number(body.timeoutSec || 10),
        Number(body.delaySec || 60),
        assignmentMode,
        JSON.stringify(validAssignmentPool),
        JSON.stringify(assignmentConfig),
      ],
    );
    await client.query(
      `
      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
      values ($1, $2, 'crm.config.updated', 'crm_config', $3, $4::jsonb)
      `,
      [
        auth.tenantId,
        auth.userId,
        result.rows[0].id,
        JSON.stringify({ enabled: body.enabled === true, assignmentMode, assignmentPoolSize: validAssignmentPool.length }),
      ],
    );
    return result.rows[0];
  });

  return { item };
}

export async function crmQueueRoute(ctx: RequestContext) {
  const auth = await requireTenantAdmin(ctx);
  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(
    `
    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 ${filters.join(' and ')}
    order by created_at desc
    limit $${params.length}
    `,
    params,
  );
  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>) };
}
