import type pg from 'pg';
import { HttpError } from '../../core/errors.js';

export function createInvoiceNo(prefix = 'BILL') {
  const now = new Date();
  const stamp = [
    now.getFullYear(),
    String(now.getMonth() + 1).padStart(2, '0'),
    String(now.getDate()).padStart(2, '0'),
    String(now.getHours()).padStart(2, '0'),
    String(now.getMinutes()).padStart(2, '0'),
    String(now.getSeconds()).padStart(2, '0'),
  ].join('');
  const random = Math.random().toString(36).slice(2, 8).toUpperCase();
  return `${prefix}${stamp}${random}`;
}

export function normalizeSlug(slug: string) {
  return slug
    .trim()
    .toLowerCase()
    .replace(/[^a-z0-9-]/g, '-')
    .replace(/-+/g, '-')
    .replace(/^-|-$/g, '');
}

export function normalizeHost(host: string) {
  return host.trim().toLowerCase().replace(/^https?:\/\//, '').split('/')[0]?.split(':')[0] || '';
}

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

export function quantityFrom(value: unknown, fallback = 1) {
  const parsed = Number(value ?? fallback);
  return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback;
}

export interface InvoiceItemInput {
  itemType: string;
  description: string;
  quantity: number;
  unitAmountCents: number;
  metadata?: Record<string, unknown>;
}

export interface CreateInvoiceInput {
  tenantId: string;
  invoiceNo?: string;
  invoiceType?: string;
  status?: string;
  currency?: string;
  discountCents?: number;
  taxCents?: number;
  billingPeriodStart?: string | null;
  billingPeriodEnd?: string | null;
  dueDate?: string | null;
  note?: string | null;
  metadata?: Record<string, unknown>;
  items: InvoiceItemInput[];
}

export function normalizeInvoiceItems(items: unknown): InvoiceItemInput[] {
  if (!Array.isArray(items)) return [];

  return items
    .map(item => (item && typeof item === 'object' ? (item as Record<string, unknown>) : null))
    .filter((item): item is Record<string, unknown> => !!item)
    .map(item => ({
      itemType: typeof item.itemType === 'string' && item.itemType.trim() ? item.itemType.trim() : 'service_fee',
      description: typeof item.description === 'string' && item.description.trim() ? item.description.trim() : '服务费',
      quantity: quantityFrom(item.quantity, 1),
      unitAmountCents: centsFrom(item.unitAmountCents, 0),
      metadata: item.metadata && typeof item.metadata === 'object' && !Array.isArray(item.metadata) ? (item.metadata as Record<string, unknown>) : {},
    }))
    .filter(item => item.unitAmountCents > 0);
}

export function invoiceSubtotal(items: InvoiceItemInput[]) {
  return items.reduce((sum, item) => sum + Math.round(item.quantity * item.unitAmountCents), 0);
}

export async function recalculateInvoiceTotals(client: pg.PoolClient, invoiceId: string) {
  const itemResult = await client.query<{ subtotal: string }>(
    `
    select coalesce(sum(amount_cents), 0)::text as subtotal
    from public.tenant_invoice_items
    where invoice_id = $1
    `,
    [invoiceId],
  );

  const paymentResult = await client.query<{ paid: string }>(
    `
    select coalesce(sum(amount_cents), 0)::text as paid
    from public.tenant_invoice_payments
    where invoice_id = $1 and status = 'paid'
    `,
    [invoiceId],
  );

  const subtotal = Number(itemResult.rows[0]?.subtotal || 0);
  const paid = Number(paymentResult.rows[0]?.paid || 0);

  const updateResult = await client.query(
    `
    update public.tenant_invoices
    set subtotal_cents = $2,
        total_cents = greatest(0, $2 - discount_cents + tax_cents),
        paid_cents = $3,
        balance_cents = greatest(0, greatest(0, $2 - discount_cents + tax_cents) - $3),
        status = case
          when status = 'void' then status
          when $3 >= greatest(0, $2 - discount_cents + tax_cents) and greatest(0, $2 - discount_cents + tax_cents) > 0 then 'paid'
          else status
        end,
        paid_at = case
          when $3 >= greatest(0, $2 - discount_cents + tax_cents) and greatest(0, $2 - discount_cents + tax_cents) > 0 then coalesce(paid_at, now())
          else paid_at
        end,
        updated_at = now()
    where id = $1
    returning id, invoice_no as "invoiceNo", status, subtotal_cents as "subtotalCents",
              discount_cents as "discountCents", tax_cents as "taxCents",
              total_cents as "totalCents", paid_cents as "paidCents",
              balance_cents as "balanceCents", paid_at as "paidAt"
    `,
    [invoiceId, subtotal, paid],
  );

  return updateResult.rows[0];
}

export async function createInvoiceRecordWithClient(client: pg.PoolClient, input: CreateInvoiceInput) {
  if (!input.items.length) throw new HttpError(400, 'At least one invoice item is required', 'INVOICE_ITEMS_REQUIRED');

  const discountCents = centsFrom(input.discountCents, 0);
  const taxCents = centsFrom(input.taxCents, 0);
  const subtotalCents = invoiceSubtotal(input.items);
  const totalCents = Math.max(0, subtotalCents - discountCents + taxCents);
  const invoiceNo = input.invoiceNo || createInvoiceNo();

  const invoiceResult = await client.query(
    `
    insert into public.tenant_invoices (
      tenant_id, invoice_no, invoice_type, status, currency,
      subtotal_cents, discount_cents, tax_cents, total_cents, paid_cents, balance_cents,
      billing_period_start, billing_period_end, due_date, issued_at, note, metadata
    )
    values (
      $1, $2, $3, $4, $5,
      $6, $7, $8, $9, 0, $9,
      $10::date, $11::date, $12::date,
      case when $4 = 'draft' then null else now() end,
      $13, $14::jsonb
    )
    returning id, tenant_id as "tenantId", invoice_no as "invoiceNo",
              invoice_type as "invoiceType", status, total_cents as "totalCents",
              balance_cents as "balanceCents", due_date as "dueDate",
              created_at as "createdAt"
    `,
    [
      input.tenantId,
      invoiceNo,
      input.invoiceType || 'subscription',
      input.status || 'issued',
      input.currency || 'CNY',
      subtotalCents,
      discountCents,
      taxCents,
      totalCents,
      input.billingPeriodStart || null,
      input.billingPeriodEnd || null,
      input.dueDate || null,
      input.note || null,
      JSON.stringify(input.metadata || {}),
    ],
  );

  const invoice = invoiceResult.rows[0];
  for (const itemInput of input.items) {
    await client.query(
      `
      insert into public.tenant_invoice_items (
        tenant_id, invoice_id, item_type, description,
        quantity, unit_amount_cents, amount_cents, metadata
      )
      values ($1, $2, $3, $4, $5, $6, $7, $8::jsonb)
      `,
      [
        input.tenantId,
        invoice.id,
        itemInput.itemType,
        itemInput.description,
        itemInput.quantity,
        itemInput.unitAmountCents,
        Math.round(itemInput.quantity * itemInput.unitAmountCents),
        JSON.stringify(itemInput.metadata || {}),
      ],
    );
  }

  await recalculateInvoiceTotals(client, invoice.id);
  return invoice;
}

export interface UsageMetricSnapshot {
  value: number;
  recordId: string;
  source: string | null;
  createdAt: string | null;
}

export interface UsageOverageItem {
  itemType: string;
  description: string;
  quantity: number;
  unitAmountCents: number;
  metadata: Record<string, unknown>;
}

export interface UsageOverageCandidate {
  tenantId: string;
  tenantSlug: string;
  tenantName: string;
  billingStatus: string;
  subscriptionId: string;
  planCode: string;
  planName: string | null;
  subscriptionStatus: string;
  billingCycle: string | null;
  periodStart: string;
  periodEnd: string;
  existingInvoiceId: string | null;
  existingInvoiceNo: string | null;
  existingInvoiceStatus: string | null;
  hasExistingInvoice: boolean;
  wouldCreate: boolean;
  totalCents: number;
  items: UsageOverageItem[];
}

export interface UsageOverageInvoiceBatchOptions {
  tenantIds?: string[];
  periodStart: string;
  periodEnd: string;
  dueDate?: string | null;
  status?: 'draft' | 'issued';
  note?: string | null;
  dryRun?: boolean;
  limit?: number;
  workerId?: string | null;
  actorUserId?: string | null;
  ipAddress?: string | null;
  userAgent?: string | null;
}

const USAGE_METRIC_ALIASES: Record<string, string[]> = {
  students: ['students', 'studentCount'],
  active_students: ['active_students', 'activeStudents', 'activeStudentCount'],
  questions: ['questions', 'questionCount'],
  assets: ['assets', 'assetCount'],
  storage_gb: ['storage_gb', 'storageGb', 'storageGB', 'storage'],
  videos: ['videos', 'videoCount'],
  video_plays: ['video_plays', 'videoPlays', 'videoPlayCount'],
  video_quota_consumed: ['video_quota_consumed', 'videoQuotaConsumed', 'videoQuota'],
  paid_orders: ['paid_orders', 'paidOrders', 'paidOrderCount'],
  paid_order_amount_cents: ['paid_order_amount_cents', 'paidOrderAmountCents', 'paidOrderGmvCents'],
  active_entitlements: ['active_entitlements', 'activeEntitlements', 'activeEntitlementCount'],
};

const USAGE_METRIC_LABELS: Record<string, string> = {
  students: '学生数',
  active_students: '活跃学生数',
  questions: '题目数量',
  assets: '资源数量',
  storage_gb: '存储容量 GB',
  videos: '视频数量',
  video_plays: '视频播放次数',
  video_quota_consumed: '视频次数消耗',
  paid_orders: '已支付订单数',
  paid_order_amount_cents: '已支付订单金额',
  active_entitlements: '有效权益数',
};

function numberOrNull(value: unknown) {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : null;
}

function positiveIntegerOrNull(value: unknown) {
  const parsed = Number(value);
  if (!Number.isFinite(parsed) || parsed <= 0) return null;
  return Math.max(1, Math.trunc(parsed));
}

function metricAliases(metricKey: string) {
  return [...new Set([metricKey, ...(USAGE_METRIC_ALIASES[metricKey] || [])])];
}

function camelMetricKey(metricKey: string) {
  return metricKey.replace(/_([a-z])/g, (_, char: string) => char.toUpperCase());
}

function snakeMetricKey(metricKey: string) {
  return metricKey.replace(/[A-Z]/g, char => `_${char.toLowerCase()}`);
}

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

function nestedMetricSpec(source: Record<string, unknown>, metricKey: string) {
  for (const key of metricAliases(metricKey)) {
    if (Object.prototype.hasOwnProperty.call(source, key)) return source[key];
  }
  return undefined;
}

function firstMetricValue(sources: Array<Record<string, unknown> | null>, metricKey: string, suffixes: string[]) {
  const camel = camelMetricKey(metricKey);
  const snake = snakeMetricKey(metricKey);
  const directKeys = metricAliases(metricKey);
  const generatedKeys = suffixes.flatMap(suffix => [
    `${camel}${suffix}`,
    `${snake}_${suffix.replace(/[A-Z]/g, char => `_${char.toLowerCase()}`).replace(/^_/, '')}`,
  ]);

  for (const source of sources) {
    if (!source) continue;
    for (const key of [...directKeys, ...generatedKeys]) {
      if (Object.prototype.hasOwnProperty.call(source, key)) return source[key];
    }
  }
  return undefined;
}

function quotaForMetric(metricKey: string, planQuotas: Record<string, unknown>, subscriptionMetadata: Record<string, unknown>) {
  const metadataQuotaSources = [
    objectOrNull(subscriptionMetadata.includedQuotas),
    objectOrNull(subscriptionMetadata.quotas),
    objectOrNull(subscriptionMetadata.quotaOverrides),
  ];
  const sources = [...metadataQuotaSources, planQuotas];
  const direct = firstMetricValue(sources, metricKey, ['Included', 'Quota', 'Limit']);
  if (direct && typeof direct === 'object' && !Array.isArray(direct)) {
    const spec = direct as Record<string, unknown>;
    return numberOrNull(spec.included ?? spec.includedQuota ?? spec.quota ?? spec.limit ?? spec.value);
  }
  const numeric = numberOrNull(direct);
  if (numeric !== null) return numeric;

  for (const source of sources) {
    if (!source) continue;
    const nested = objectOrNull(nestedMetricSpec(source, metricKey));
    if (nested) {
      const value = numberOrNull(nested.included ?? nested.includedQuota ?? nested.quota ?? nested.limit ?? nested.value);
      if (value !== null) return value;
    }
  }
  return null;
}

function priceForMetric(metricKey: string, planPrices: Record<string, unknown>, subscriptionMetadata: Record<string, unknown>) {
  const metadataPriceSources = [
    objectOrNull(subscriptionMetadata.overagePrices),
    objectOrNull(subscriptionMetadata.overagePriceOverrides),
    objectOrNull(subscriptionMetadata.prices),
  ];
  const sources = [...metadataPriceSources, planPrices];
  const suffixes = [
    'UnitAmountCents',
    'AmountCents',
    'PriceCents',
    'OverageCents',
    'PerUnitCents',
    'ExtraCents',
    'ExtraPerMonthCents',
    'ExtraPerYearCents',
    'PerMonthCents',
    'PerYearCents',
  ];
  const direct = firstMetricValue(sources, metricKey, suffixes);
  const directObject = objectOrNull(direct);
  let unitAmountCents = directObject
    ? positiveIntegerOrNull(directObject.unitAmountCents ?? directObject.amountCents ?? directObject.priceCents ?? directObject.overageCents ?? directObject.cents ?? directObject.perUnitCents)
    : positiveIntegerOrNull(direct);
  let unitSize = directObject ? numberOrNull(directObject.unitSize ?? directObject.step ?? directObject.per ?? directObject.quantityUnit) : null;

  for (const source of sources) {
    if (!source) continue;
    const nested = objectOrNull(nestedMetricSpec(source, metricKey));
    if (!nested) continue;
    unitAmountCents = unitAmountCents ?? positiveIntegerOrNull(nested.unitAmountCents ?? nested.amountCents ?? nested.priceCents ?? nested.overageCents ?? nested.cents ?? nested.perUnitCents);
    unitSize = unitSize ?? numberOrNull(nested.unitSize ?? nested.step ?? nested.per ?? nested.quantityUnit);
  }

  if (!unitAmountCents) return null;
  return {
    unitAmountCents,
    unitSize: unitSize && unitSize > 0 ? unitSize : 1,
  };
}

function usageSnapshotMap(value: unknown) {
  const usage = objectOrNull(value) || {};
  const output: Record<string, UsageMetricSnapshot> = {};
  for (const [metricKey, rawSnapshot] of Object.entries(usage)) {
    const snapshot = objectOrNull(rawSnapshot);
    if (!snapshot) continue;
    const metricValue = numberOrNull(snapshot.value);
    if (metricValue === null) continue;
    output[metricKey] = {
      value: metricValue,
      recordId: String(snapshot.recordId || ''),
      source: typeof snapshot.source === 'string' ? snapshot.source : null,
      createdAt: typeof snapshot.createdAt === 'string' ? snapshot.createdAt : null,
    };
  }
  return output;
}

function buildUsageOverageItems(row: {
  planCode: string;
  planName: string | null;
  includedQuotas: Record<string, unknown> | null;
  overagePrices: Record<string, unknown> | null;
  subscriptionMetadata: Record<string, unknown> | null;
  usageSnapshots: unknown;
}) {
  const planQuotas = row.includedQuotas || {};
  const planPrices = row.overagePrices || {};
  const subscriptionMetadata = row.subscriptionMetadata || {};
  const usage = usageSnapshotMap(row.usageSnapshots);
  const items: UsageOverageItem[] = [];

  for (const [metricKey, snapshot] of Object.entries(usage)) {
    const includedQuota = quotaForMetric(metricKey, planQuotas, subscriptionMetadata);
    const price = priceForMetric(metricKey, planPrices, subscriptionMetadata);
    if (includedQuota === null || !price) continue;
    const overageValue = snapshot.value - includedQuota;
    if (overageValue <= 0) continue;

    const billableUnits = Math.ceil(overageValue / price.unitSize);
    if (billableUnits <= 0) continue;
    const label = USAGE_METRIC_LABELS[metricKey] || metricKey;
    items.push({
      itemType: 'usage_overage',
      description: `${row.planName || row.planCode} ${label}超额 ${Number(overageValue.toFixed(4))}`,
      quantity: billableUnits,
      unitAmountCents: price.unitAmountCents,
      metadata: {
        metricKey,
        metricLabel: label,
        metricValue: snapshot.value,
        includedQuota,
        overageValue: Number(overageValue.toFixed(4)),
        billableUnits,
        unitSize: price.unitSize,
        unitAmountCents: price.unitAmountCents,
        usageRecordId: snapshot.recordId || null,
        usageSource: snapshot.source,
      },
    });
  }

  return items;
}

function clampUsageOverageLimit(value: unknown, fallback = 100) {
  const parsed = Number(value ?? fallback);
  if (!Number.isFinite(parsed) || parsed <= 0) return fallback;
  return Math.min(Math.trunc(parsed), 500);
}

async function insertUsageOverageAudit(
  client: pg.PoolClient,
  input: {
    tenantId: string | null;
    actorUserId?: string | null;
    action: string;
    targetType: string;
    targetId: string | null;
    details: Record<string, unknown>;
    ipAddress?: string | null;
    userAgent?: string | null;
  },
) {
  await client.query(
    `
    insert into public.audit_logs (
      tenant_id, actor_user_id, action, target_type, target_id,
      details, ip_address, user_agent
    )
    values ($1::uuid, $2::uuid, $3, $4, $5, $6::jsonb, $7, $8)
    `,
    [
      input.tenantId,
      input.actorUserId || null,
      input.action,
      input.targetType,
      input.targetId,
      JSON.stringify(input.details || {}),
      'ipAddress' in input ? input.ipAddress || null : null,
      'userAgent' in input ? input.userAgent || null : null,
    ],
  );
}

export async function usageOverageCandidateQuery(
  db: pg.Pool | pg.PoolClient,
  params: {
    tenantIds?: string[];
    periodStart: string;
    periodEnd: string;
    includeExisting?: boolean;
    includeZero?: boolean;
    limit?: number;
  },
) {
  const tenantIds = params.tenantIds || [];
  const includeExisting = Boolean(params.includeExisting);
  const includeZero = Boolean(params.includeZero);
  const limit = clampUsageOverageLimit(params.limit, 100);
  const rows = await db.query<{
    tenantId: string;
    tenantSlug: string;
    tenantName: string;
    billingStatus: string;
    subscriptionId: string;
    planCode: string;
    planName: string | null;
    subscriptionStatus: string;
    billingCycle: string | null;
    includedQuotas: Record<string, unknown> | null;
    overagePrices: Record<string, unknown> | null;
    subscriptionMetadata: Record<string, unknown> | null;
    usageSnapshots: unknown;
    existingInvoiceId: string | null;
    existingInvoiceNo: string | null;
    existingInvoiceStatus: string | null;
  }>(
    `
    select t.id as "tenantId", t.slug::text as "tenantSlug", t.name as "tenantName",
           t.billing_status as "billingStatus",
           s.id as "subscriptionId", s.plan_code as "planCode",
           p.name as "planName", s.status as "subscriptionStatus",
           s.billing_cycle as "billingCycle",
           p.included_quotas as "includedQuotas",
           p.overage_prices as "overagePrices",
           s.metadata as "subscriptionMetadata",
           coalesce(usage_snapshots.metrics, '{}'::jsonb) as "usageSnapshots",
           existing.id as "existingInvoiceId",
           existing.invoice_no as "existingInvoiceNo",
           existing.status as "existingInvoiceStatus"
    from public.tenants t
    join lateral (
      select id, tenant_id, plan_code, status, billing_cycle, metadata, created_at, expires_at
      from public.tenant_subscriptions
      where tenant_id = t.id
        and status in ('trial', 'active', 'past_due')
      order by case status when 'active' then 0 when 'trial' then 1 else 2 end,
               expires_at desc nulls last,
               created_at desc
      limit 1
    ) s on true
    join public.platform_saas_plans p on p.code = s.plan_code
    left join lateral (
      select jsonb_object_agg(metric_key, jsonb_build_object(
        'value', metric_value,
        'recordId', id,
        'source', metadata->>'source',
        'createdAt', created_at
      )) as metrics
      from (
        select distinct on (u.metric_key)
               u.id, u.metric_key, u.metric_value, u.metadata, u.created_at
        from public.tenant_usage_records u
        where u.tenant_id = t.id
          and u.period_start = $2::date
          and u.period_end = $3::date
        order by u.metric_key,
                 case when u.metadata->>'source' = 'platform_usage_worker' then 0 else 1 end,
                 u.created_at desc
      ) latest
    ) usage_snapshots on true
    left join lateral (
      select id, invoice_no, status
      from public.tenant_invoices i
      where i.tenant_id = t.id
        and i.invoice_type = 'usage_overage'
        and i.status <> 'void'
        and i.billing_period_start = $2::date
        and i.billing_period_end = $3::date
        and i.metadata->>'source' = 'usage_overage_auto'
      order by i.created_at desc
      limit 1
    ) existing on true
    where t.status = 'active'
      and ($1::uuid[] = '{}'::uuid[] or t.id = any($1::uuid[]))
      and ($4::boolean = true or existing.id is null)
    order by t.created_at asc
    limit $5
    `,
    [tenantIds, params.periodStart, params.periodEnd, includeExisting, limit],
  );

  const candidates: UsageOverageCandidate[] = rows.rows.map(row => {
    const items = buildUsageOverageItems(row);
    return {
      tenantId: row.tenantId,
      tenantSlug: row.tenantSlug,
      tenantName: row.tenantName,
      billingStatus: row.billingStatus,
      subscriptionId: row.subscriptionId,
      planCode: row.planCode,
      planName: row.planName,
      subscriptionStatus: row.subscriptionStatus,
      billingCycle: row.billingCycle,
      periodStart: params.periodStart,
      periodEnd: params.periodEnd,
      existingInvoiceId: row.existingInvoiceId,
      existingInvoiceNo: row.existingInvoiceNo,
      existingInvoiceStatus: row.existingInvoiceStatus,
      hasExistingInvoice: Boolean(row.existingInvoiceId),
      wouldCreate: items.length > 0 && !row.existingInvoiceId,
      totalCents: invoiceSubtotal(items),
      items,
    };
  });

  return candidates.filter(item => includeZero || item.items.length > 0);
}

export async function processUsageOverageInvoices(
  client: pg.PoolClient,
  options: UsageOverageInvoiceBatchOptions,
) {
  const tenantIds = options.tenantIds || [];
  const dryRun = Boolean(options.dryRun);
  const status = options.status || 'issued';
  const limit = clampUsageOverageLimit(options.limit, Math.min(Math.max(tenantIds.length || 0, 100), 500));
  const candidates = await usageOverageCandidateQuery(client, {
    tenantIds,
    periodStart: options.periodStart,
    periodEnd: options.periodEnd,
    includeExisting: true,
    includeZero: false,
    limit,
  });

  if (!candidates.length) {
    return { dryRun, createdCount: 0, skippedCount: 0, totalCents: 0, items: [], skipped: [] };
  }

  if (dryRun) {
    const wouldCreate = candidates.filter(item => !item.hasExistingInvoice);
    const existing = candidates.filter(item => item.hasExistingInvoice);
    return {
      dryRun: true,
      createdCount: 0,
      skippedCount: existing.length,
      totalCents: wouldCreate.reduce((sum, item) => sum + item.totalCents, 0),
      items: candidates.map(item => ({ ...item, wouldCreate: !item.hasExistingInvoice })),
      skipped: existing.map(item => ({
        tenantId: item.tenantId,
        subscriptionId: item.subscriptionId,
        reason: 'USAGE_OVERAGE_INVOICE_EXISTS',
        invoiceId: item.existingInvoiceId,
        invoiceNo: item.existingInvoiceNo,
      })),
    };
  }

  const created: Array<UsageOverageCandidate & { invoice: Record<string, unknown> }> = [];
  const skipped: unknown[] = [];

  for (const candidate of candidates) {
    const lock = await client.query(
      `
      select id
      from public.tenant_subscriptions
      where tenant_id = $1
        and id = $2
      for update
      `,
      [candidate.tenantId, candidate.subscriptionId],
    );
    if (!lock.rows[0]) {
      skipped.push({ tenantId: candidate.tenantId, subscriptionId: candidate.subscriptionId, reason: 'SUBSCRIPTION_NOT_FOUND' });
      continue;
    }

    const existing = await client.query<{ id: string; invoiceNo: string; status: string }>(
      `
      select id, invoice_no as "invoiceNo", status
      from public.tenant_invoices
      where tenant_id = $1
        and invoice_type = 'usage_overage'
        and status <> 'void'
        and billing_period_start = $2::date
        and billing_period_end = $3::date
        and metadata->>'source' = 'usage_overage_auto'
      order by created_at desc
      limit 1
      `,
      [candidate.tenantId, options.periodStart, options.periodEnd],
    );
    if (existing.rows[0]) {
      skipped.push({
        tenantId: candidate.tenantId,
        subscriptionId: candidate.subscriptionId,
        reason: 'USAGE_OVERAGE_INVOICE_EXISTS',
        invoiceId: existing.rows[0].id,
        invoiceNo: existing.rows[0].invoiceNo,
      });
      continue;
    }

    if (!candidate.items.length) {
      skipped.push({ tenantId: candidate.tenantId, subscriptionId: candidate.subscriptionId, reason: 'NO_USAGE_OVERAGE' });
      continue;
    }

    const invoice = await createInvoiceRecordWithClient(client, {
      tenantId: candidate.tenantId,
      invoiceType: 'usage_overage',
      status,
      dueDate: options.dueDate || null,
      billingPeriodStart: options.periodStart,
      billingPeriodEnd: options.periodEnd,
      note: options.note || null,
      metadata: {
        source: 'usage_overage_auto',
        periodStart: options.periodStart,
        periodEnd: options.periodEnd,
        subscriptionId: candidate.subscriptionId,
        planCode: candidate.planCode,
        tenantSlug: candidate.tenantSlug,
        workerId: options.workerId || null,
      },
      items: candidate.items.map(item => ({
        ...item,
        metadata: {
          ...item.metadata,
          periodStart: options.periodStart,
          periodEnd: options.periodEnd,
          subscriptionId: candidate.subscriptionId,
          planCode: candidate.planCode,
        },
      })),
    });

    await insertUsageOverageAudit(client, {
      tenantId: candidate.tenantId,
      actorUserId: options.actorUserId || null,
      ipAddress: options.ipAddress || null,
      userAgent: options.userAgent || null,
      action: 'platform.invoice.usage_overage_created',
      targetType: 'tenant_invoice',
      targetId: String(invoice.id),
      details: {
        tenantId: candidate.tenantId,
        subscriptionId: candidate.subscriptionId,
        planCode: candidate.planCode,
        periodStart: options.periodStart,
        periodEnd: options.periodEnd,
        totalCents: candidate.totalCents,
        metrics: candidate.items.map(item => item.metadata.metricKey),
        invoiceNo: invoice.invoiceNo,
        workerId: options.workerId || null,
      },
    });

    created.push({ ...candidate, invoice });
  }

  const totalCents = created.reduce((sum, item) => sum + Number(item.totalCents || 0), 0);

  await insertUsageOverageAudit(client, {
    tenantId: null,
    actorUserId: options.actorUserId || null,
    ipAddress: options.ipAddress || null,
    userAgent: options.userAgent || null,
    action: 'platform.invoice.usage_overage_batch_created',
    targetType: 'tenant_invoice_batch',
    targetId: null,
    details: {
      createdCount: created.length,
      skippedCount: skipped.length,
      totalCents,
      tenantIds,
      periodStart: options.periodStart,
      periodEnd: options.periodEnd,
      dueDate: options.dueDate || null,
      status,
      workerId: options.workerId || null,
    },
  });

  return {
    dryRun: false,
    createdCount: created.length,
    skippedCount: skipped.length,
    totalCents,
    items: created,
    skipped,
  };
}

interface ProcessOverduePlatformInvoicesOptions {
  actorUserId?: string | null;
  channel?: string;
  reminderType?: string;
  dryRun?: boolean;
  limit?: number;
  today?: string | null;
  workerId?: string | null;
}

function clampPositiveInteger(value: unknown, fallback: number, max: number) {
  const parsed = Number(value ?? fallback);
  if (!Number.isFinite(parsed) || parsed <= 0) return fallback;
  return Math.min(Math.trunc(parsed), max);
}

function yyyyMmDd(value: Date) {
  return value.toISOString().slice(0, 10);
}

function currentDateText(value?: string | null) {
  if (typeof value === 'string' && /^\d{4}-\d{2}-\d{2}$/.test(value)) return value;
  return yyyyMmDd(new Date());
}

export async function processOverduePlatformInvoices(
  client: pg.PoolClient,
  options: ProcessOverduePlatformInvoicesOptions = {},
) {
  const limit = clampPositiveInteger(options.limit, 100, 1000);
  const today = currentDateText(options.today);
  const channel = options.channel || 'internal';
  const reminderType = options.reminderType || 'overdue';

  const invoiceResult = await client.query<{
    id: string;
    tenantId: string;
    tenantSlug: string;
    tenantName: string;
    invoiceNo: string;
    status: string;
    balanceCents: number;
    dueDate: string | null;
    existingReminderId: string | null;
    reminderCount: number;
  }>(
    `
    select i.id,
           i.tenant_id as "tenantId",
           t.slug::text as "tenantSlug",
           t.name as "tenantName",
           i.invoice_no as "invoiceNo",
           i.status,
           i.balance_cents as "balanceCents",
           i.due_date as "dueDate",
           existing.id as "existingReminderId",
           coalesce(history.reminder_count, 0)::integer as "reminderCount"
    from public.tenant_invoices i
    join public.tenants t on t.id = i.tenant_id
    left join lateral (
      select id
      from public.tenant_invoice_reminders r
      where r.tenant_id = i.tenant_id
        and r.invoice_id = i.id
        and r.reminder_type = $2
        and r.channel = $3
        and r.reminder_date = $1::date
      limit 1
    ) existing on true
    left join lateral (
      select count(*)::integer as reminder_count
      from public.tenant_invoice_reminders r
      where r.tenant_id = i.tenant_id
        and r.invoice_id = i.id
        and r.reminder_type = $2
    ) history on true
    where i.status in ('issued', 'overdue')
      and i.balance_cents > 0
      and i.due_date is not null
      and i.due_date < $1::date
      and t.status = 'active'
    order by i.due_date asc, i.created_at asc
    limit $4
    for update of i skip locked
    `,
    [today, reminderType, channel, limit],
  );

  if (options.dryRun) {
    return {
      dryRun: true,
      processed: invoiceResult.rowCount,
      markedOverdue: 0,
      reminderCreated: 0,
      skippedReminder: invoiceResult.rows.filter(row => row.existingReminderId).length,
      items: invoiceResult.rows.map(row => ({
        ...row,
        wouldMarkOverdue: row.status !== 'overdue',
        wouldCreateReminder: !row.existingReminderId,
      })),
    };
  }

  const items: unknown[] = [];
  let markedOverdue = 0;
  let reminderCreated = 0;
  let skippedReminder = 0;

  for (const invoice of invoiceResult.rows) {
    if (invoice.status !== 'overdue') {
      await client.query(
        `
        update public.tenant_invoices
        set status = 'overdue',
            metadata = metadata || $3::jsonb,
            updated_at = now()
        where tenant_id = $1 and id = $2
        `,
        [
          invoice.tenantId,
          invoice.id,
          JSON.stringify({
            overdueMarkedAt: new Date().toISOString(),
            overdueMarkedBy: options.workerId || options.actorUserId || 'platform-admin',
          }),
        ],
      );
      markedOverdue += 1;
    }

    await client.query(
      `
      update public.tenants
      set billing_status = case when billing_status = 'active' then 'past_due' else billing_status end,
          updated_at = now()
      where id = $1
      `,
      [invoice.tenantId],
    );

    let reminder = null;
    if (invoice.existingReminderId) {
      skippedReminder += 1;
    } else {
      const reminderResult = await client.query(
        `
        insert into public.tenant_invoice_reminders (
          tenant_id, invoice_id, reminder_type, channel, status,
          reminder_date, reminder_level, due_date,
          balance_cents_snapshot, message, metadata, created_by
        )
        values (
          $1, $2, $3, $4, 'pending',
          $5::date, $6, $7::date,
          $8, $9, $10::jsonb, $11::uuid
        )
        on conflict (tenant_id, invoice_id, reminder_type, channel, reminder_date)
        do nothing
        returning id, tenant_id as "tenantId", invoice_id as "invoiceId",
                  reminder_type as "reminderType", channel, status,
                  reminder_date as "reminderDate", reminder_level as "reminderLevel",
                  due_date as "dueDate", balance_cents_snapshot as "balanceCentsSnapshot",
                  message, metadata, created_at as "createdAt"
        `,
        [
          invoice.tenantId,
          invoice.id,
          reminderType,
          channel,
          today,
          Number(invoice.reminderCount || 0) + 1,
          invoice.dueDate,
          Number(invoice.balanceCents || 0),
          `租户 ${invoice.tenantName} 的平台服务费账单 ${invoice.invoiceNo} 已逾期，请跟进收款。`,
          JSON.stringify({
            source: options.workerId ? 'worker' : 'platform_admin',
            workerId: options.workerId || null,
            invoiceNo: invoice.invoiceNo,
            tenantSlug: invoice.tenantSlug,
          }),
          options.actorUserId || null,
        ],
      );
      reminder = reminderResult.rows[0] || null;
      if (reminder) reminderCreated += 1;
      else skippedReminder += 1;
    }

    await client.query(
      `
      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
      values ($1, $2::uuid, 'platform.invoice.overdue_processed', 'tenant_invoice', $3, $4::jsonb)
      `,
      [
        invoice.tenantId,
        options.actorUserId || null,
        invoice.id,
        JSON.stringify({
          invoiceNo: invoice.invoiceNo,
          dueDate: invoice.dueDate,
          balanceCents: invoice.balanceCents,
          markedOverdue: invoice.status !== 'overdue',
          reminderCreated: Boolean(reminder),
          channel,
          reminderType,
          workerId: options.workerId || null,
        }),
      ],
    );

    items.push({
      ...invoice,
      status: 'overdue',
      markedOverdue: invoice.status !== 'overdue',
      reminderCreated: Boolean(reminder),
      reminder,
    });
  }

  return {
    dryRun: false,
    processed: invoiceResult.rowCount,
    markedOverdue,
    reminderCreated,
    skippedReminder,
    items,
  };
}
