feat: automate usage overage billing

apps/api/src/features/platform-admin/routes.ts

@@ -10,17 +10,20 @@ import {
   requiredString,
   requirePlatformAdmin,
 } from '../../core/request.js';
-import { query, queryOne, transaction } from '../../core/db.js';
+import { pool, query, queryOne, transaction } from '../../core/db.js';
 import {
   centsFrom,
   createInvoiceNo,
-  invoiceSubtotal,
+  createInvoiceRecordWithClient,
+  type CreateInvoiceInput,
   normalizeHost,
   normalizeInvoiceItems,
   normalizeSlug,
   processOverduePlatformInvoices,
+  processUsageOverageInvoices,
   quantityFrom,
   recalculateInvoiceTotals,
+  usageOverageCandidateQuery as loadUsageOverageInvoiceCandidates,
 } from './service.js';
 
 function jsonBodyValue(value: unknown) {
@@ -349,12 +352,12 @@ function tenantInvoiceStatusFrom(value: string) {
   return status;
 }
 
-function usageOverageInvoiceStatusFrom(value: string) {
+function usageOverageInvoiceStatusFrom(value: string): 'draft' | 'issued' {
   const status = tenantInvoiceStatusFrom(value);
   if (!['draft', 'issued'].includes(status)) {
     throw new HttpError(400, 'usage overage invoices can only be draft or issued', 'INVALID_INVOICE_STATUS');
   }
-  return status;
+  return status as 'draft' | 'issued';
 }
 
 function dateTextFrom(value: unknown, key: string, required = true) {
@@ -2424,95 +2427,6 @@ export async function tenantInvoicesRoute(ctx: RequestContext) {
   return { items };
 }
 
-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: ReturnType<typeof normalizeInvoiceItems>;
-}
-
-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;
-}
-
 async function createInvoiceRecord(input: CreateInvoiceInput) {
   return transaction(async client => createInvoiceRecordWithClient(client, input));
 }
@@ -2762,372 +2676,6 @@ export async function tenantUsageRoute(ctx: RequestContext) {
   return { items };
 }
 
-interface UsageMetricSnapshot {
-  value: number;
-  recordId: string;
-  source: string | null;
-  createdAt: string | null;
-}
-
-interface UsageOverageItem {
-  itemType: string;
-  description: string;
-  quantity: number;
-  unitAmountCents: number;
-  metadata: Record<string, unknown>;
-}
-
-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[];
-}
-
-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;
-}
-
-async function usageOverageCandidateQuery(params: {
-  tenantIds: string[];
-  periodStart: string;
-  periodEnd: string;
-  includeExisting: boolean;
-  includeZero: boolean;
-  limit: number;
-}) {
-  const rows = await 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
-    `,
-    [params.tenantIds, params.periodStart, params.periodEnd, params.includeExisting, params.limit],
-  );
-
-  const candidates: UsageOverageCandidate[] = 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 => params.includeZero || item.items.length > 0);
-}
-
 export async function usageOverageInvoiceCandidatesRoute(ctx: RequestContext) {
   await requirePlatformAdmin(ctx, 'platform:billing:read');
 
@@ -3138,7 +2686,7 @@ export async function usageOverageInvoiceCandidatesRoute(ctx: RequestContext) {
   const includeExisting = listQuery(ctx, 'includeExisting') === 'true';
   const includeZero = listQuery(ctx, 'includeZero') === 'true';
   const limit = intParam(ctx, 'limit', 100, 500);
-  const items = await usageOverageCandidateQuery({ tenantIds, periodStart, periodEnd, includeExisting, includeZero, limit });
+  const items = await loadUsageOverageInvoiceCandidates(pool, { tenantIds, periodStart, periodEnd, includeExisting, includeZero, limit });
 
   return { items };
 }
@@ -3460,155 +3008,20 @@ export async function createTenantInvoicesFromUsageOverageRoute(ctx: RequestCont
   const dryRun = booleanFrom(body.dryRun, false);
   const limit = Math.min(Math.max(tenantIds.length || 0, 100), 500);
 
-  const candidates = await usageOverageCandidateQuery({
+  const session = currentSessionFromContext(ctx);
+  const result = await transaction(async client => processUsageOverageInvoices(client, {
     tenantIds,
     periodStart,
     periodEnd,
-    includeExisting: true,
-    includeZero: false,
+    dueDate,
+    status,
+    note,
+    dryRun,
     limit,
-  });
-
-  if (!candidates.length) {
-    return { item: { 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 {
-      item: {
-        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 result = await transaction(async client => {
-    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, periodStart, 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,
-        billingPeriodStart: periodStart,
-        billingPeriodEnd: periodEnd,
-        note,
-        metadata: {
-          source: 'usage_overage_auto',
-          periodStart,
-          periodEnd,
-          subscriptionId: candidate.subscriptionId,
-          planCode: candidate.planCode,
-          tenantSlug: candidate.tenantSlug,
-        },
-        items: candidate.items.map(item => ({
-          ...item,
-          metadata: {
-            ...item.metadata,
-            periodStart,
-            periodEnd,
-            subscriptionId: candidate.subscriptionId,
-            planCode: candidate.planCode,
-          },
-        })),
-      });
-
-      await recordPlatformAudit(client, ctx, 'platform.invoice.usage_overage_created', 'tenant_invoice', invoice.id, {
-        tenantId: candidate.tenantId,
-        subscriptionId: candidate.subscriptionId,
-        planCode: candidate.planCode,
-        periodStart,
-        periodEnd,
-        totalCents: candidate.totalCents,
-        metrics: candidate.items.map(item => item.metadata.metricKey),
-        invoiceNo: invoice.invoiceNo,
-      }, candidate.tenantId);
-
-      created.push({ ...candidate, invoice });
-    }
-
-    const totalCents = created.reduce((sum, item) => sum + Number(item.totalCents || 0), 0);
-
-    await recordPlatformAudit(client, ctx, 'platform.invoice.usage_overage_batch_created', 'tenant_invoice_batch', null, {
-      createdCount: created.length,
-      skippedCount: skipped.length,
-      totalCents,
-      tenantIds,
-      periodStart,
-      periodEnd,
-      dueDate,
-      status,
-    });
-
-    return {
-      dryRun: false,
-      createdCount: created.length,
-      skippedCount: skipped.length,
-      totalCents,
-      items: created,
-      skipped,
-    };
-  });
+    actorUserId: session?.id || null,
+    ipAddress: requestIp(ctx),
+    userAgent: getHeader(ctx.req, 'user-agent') || null,
+  }));
 
   return { item: result };
 }

apps/api/src/features/platform-admin/service.ts

@@ -1,4 +1,5 @@
 import type pg from 'pg';
+import { HttpError } from '../../core/errors.js';
 
 export function createInvoiceNo(prefix = 'BILL') {
   const now = new Date();
@@ -45,6 +46,22 @@ export interface InvoiceItemInput {
   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 [];
 
@@ -116,6 +133,683 @@ export async function recalculateInvoiceTotals(client: pg.PoolClient, invoiceId:
   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;

apps/worker/package.json

@@ -13,6 +13,7 @@
     "provider-bills:once": "tsx src/index.ts --once --job provider-bills",
     "platform-billing:once": "tsx src/index.ts --once --job platform-billing",
     "platform-usage:once": "tsx src/index.ts --once --job platform-usage",
+    "platform-usage-overage:once": "tsx src/index.ts --once --job platform-usage-overage",
     "platform-dunning:once": "tsx src/index.ts --once --job platform-dunning",
     "platform-dunning-notifications:once": "tsx src/index.ts --once --job platform-dunning-notifications",
     "platform-audit-alerts:once": "tsx src/index.ts --once --job platform-audit-alerts",

apps/worker/src/config.ts

@@ -25,6 +25,10 @@ export interface WorkerConfig {
   platformUsageBatchSize: number;
   platformUsageWorkerId: string;
   platformUsageMonth: string;
+  platformUsageOverageBatchSize: number;
+  platformUsageOverageWorkerId: string;
+  platformUsageOverageMonth: string;
+  platformUsageOverageDueDays: number;
   platformDunningBatchSize: number;
   platformDunningWorkerId: string;
   platformDunningNotificationBatchSize: number;
@@ -204,6 +208,10 @@ const loadedConfig: WorkerConfig = {
   platformUsageBatchSize: envNumber('WORKER_PLATFORM_USAGE_BATCH_SIZE', 100),
   platformUsageWorkerId: envString('WORKER_PLATFORM_USAGE_ID', `platform-usage-${process.pid}`),
   platformUsageMonth: envString('WORKER_PLATFORM_USAGE_MONTH', ''),
+  platformUsageOverageBatchSize: envNumber('WORKER_PLATFORM_USAGE_OVERAGE_BATCH_SIZE', 100),
+  platformUsageOverageWorkerId: envString('WORKER_PLATFORM_USAGE_OVERAGE_ID', `platform-usage-overage-${process.pid}`),
+  platformUsageOverageMonth: envString('WORKER_PLATFORM_USAGE_OVERAGE_MONTH', ''),
+  platformUsageOverageDueDays: envNumber('WORKER_PLATFORM_USAGE_OVERAGE_DUE_DAYS', 15),
   platformDunningBatchSize: envNumber('WORKER_PLATFORM_DUNNING_BATCH_SIZE', 100),
   platformDunningWorkerId: envString('WORKER_PLATFORM_DUNNING_ID', `platform-dunning-${process.pid}`),
   platformDunningNotificationBatchSize: envNumber('WORKER_PLATFORM_DUNNING_NOTIFICATION_BATCH_SIZE', 50),

apps/worker/src/index.ts

@@ -61,6 +61,15 @@ async function runOnce() {
     );
     return;
   }
+  if (job === 'platform-usage-overage') {
+    const { processPlatformUsageOverageBatch } = await import('./jobs/platform-usage-overage.js');
+    const result = await processPlatformUsageOverageBatch();
+    console.log(
+      `[worker] platform-usage-overage batch processed=${result.processed}`
+        + ` created=${result.created} skipped=${result.skipped} totalCents=${result.totalCents}`,
+    );
+    return;
+  }
   if (job === 'platform-dunning') {
     const { processPlatformDunningBatch } = await import('./jobs/platform-dunning.js');
     const result = await processPlatformDunningBatch();

apps/worker/src/jobs/platform-usage-overage.ts

@@ -0,0 +1,97 @@
+import { pool } from '../db.js';
+import { config } from '../config.js';
+import { processUsageOverageInvoices } from '../../../api/src/features/platform-admin/service.js';
+
+export interface PlatformUsageOverageWorkerResult {
+  processed: number;
+  created: number;
+  skipped: number;
+  totalCents: number;
+}
+
+const MONTH_RE = /^\d{4}-\d{2}$/;
+
+function positiveInteger(value: number, fallback: number, max: number) {
+  if (!Number.isFinite(value) || value <= 0) return fallback;
+  return Math.min(Math.trunc(value), max);
+}
+
+function nonNegativeInteger(value: number, fallback: number, max: number) {
+  if (!Number.isFinite(value) || value < 0) return fallback;
+  return Math.min(Math.trunc(value), max);
+}
+
+function shanghaiYearMonth(value = new Date()) {
+  return new Intl.DateTimeFormat('en-CA', {
+    timeZone: 'Asia/Shanghai',
+    year: 'numeric',
+    month: '2-digit',
+  }).format(value);
+}
+
+function previousMonth(monthText: string) {
+  const [yearText, monthNumberText] = monthText.split('-');
+  const date = new Date(Date.UTC(Number(yearText), Number(monthNumberText) - 2, 1));
+  return date.toISOString().slice(0, 7);
+}
+
+function targetMonth(value?: string) {
+  const normalized = String(value || '').trim();
+  if (normalized) return normalized;
+  return previousMonth(shanghaiYearMonth());
+}
+
+function monthPeriod(monthText: string) {
+  const normalized = monthText.trim();
+  if (!MONTH_RE.test(normalized)) {
+    throw new Error(`Invalid platform usage overage month: ${monthText}. Expected YYYY-MM.`);
+  }
+  const [yearText, monthNumberText] = normalized.split('-');
+  const year = Number(yearText);
+  const monthNumber = Number(monthNumberText);
+  const periodStart = `${yearText}-${monthNumberText}-01`;
+  const periodEnd = new Date(Date.UTC(year, monthNumber, 0)).toISOString().slice(0, 10);
+  return { periodStart, periodEnd };
+}
+
+function dueDateText(days: number) {
+  const now = new Date();
+  now.setUTCDate(now.getUTCDate() + nonNegativeInteger(days, 15, 365));
+  return now.toISOString().slice(0, 10);
+}
+
+export async function processPlatformUsageOverageBatch(options: {
+  limit?: number;
+  month?: string;
+  dueDays?: number;
+} = {}): Promise<PlatformUsageOverageWorkerResult> {
+  const limit = positiveInteger(options.limit ?? config.platformUsageOverageBatchSize, 100, 500);
+  const month = targetMonth(options.month || config.platformUsageOverageMonth);
+  const { periodStart, periodEnd } = monthPeriod(month);
+  const client = await pool.connect();
+  try {
+    await client.query('begin');
+    const result = await processUsageOverageInvoices(client, {
+      periodStart,
+      periodEnd,
+      dueDate: dueDateText(options.dueDays ?? config.platformUsageOverageDueDays),
+      status: 'issued',
+      note: `平台自动生成 ${periodStart} 至 ${periodEnd} 用量超额服务费账单`,
+      dryRun: false,
+      limit,
+      workerId: config.platformUsageOverageWorkerId,
+    });
+    await client.query('commit');
+    return {
+      processed: Number(result.createdCount || 0) + Number(result.skippedCount || 0),
+      created: Number(result.createdCount || 0),
+      skipped: Number(result.skippedCount || 0),
+      totalCents: Number(result.totalCents || 0),
+    };
+  } catch (error) {
+    await client.query('rollback').catch(() => {});
+    throw error;
+  } finally {
+    client.release();
+  }
+}