feat: add platform usage worker

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

@@ -884,11 +884,23 @@ export async function platformOverviewRoute(ctx: RequestContext) {
     ),
     queryOne<{ students: string; questions: string; storageGb: string }>(
       `
+      with latest_usage as (
+        select tenant_id, metric_key, metric_value,
+               row_number() over (
+                 partition by tenant_id, metric_key
+                 order by case when metadata->>'source' = 'platform_usage_worker' then 0 else 1 end,
+                          period_end desc,
+                          created_at desc
+               ) as rn
+        from public.tenant_usage_records
+        where period_end >= current_date - interval '31 days'
+          and metric_key in ('students', 'questions', 'storage_gb')
+      )
       select coalesce(sum(metric_value) filter (where metric_key = 'students'), 0)::text as students,
              coalesce(sum(metric_value) filter (where metric_key = 'questions'), 0)::text as questions,
              coalesce(sum(metric_value) filter (where metric_key = 'storage_gb'), 0)::text as "storageGb"
-      from public.tenant_usage_records
-      where period_end >= current_date - interval '31 days'
+      from latest_usage
+      where rn = 1
       `,
     ),
   ]);

apps/worker/package.json

@@ -12,6 +12,7 @@
     "commerce:once": "tsx src/index.ts --once --job commerce",
     "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-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

@@ -22,6 +22,9 @@ export interface WorkerConfig {
   platformBillingDaysAhead: number;
   platformBillingDueDays: number;
   platformBillingWorkerId: string;
+  platformUsageBatchSize: number;
+  platformUsageWorkerId: string;
+  platformUsageMonth: string;
   platformDunningBatchSize: number;
   platformDunningWorkerId: string;
   platformDunningNotificationBatchSize: number;
@@ -198,6 +201,9 @@ const loadedConfig: WorkerConfig = {
   platformBillingDaysAhead: envNumber('WORKER_PLATFORM_BILLING_DAYS_AHEAD', 45),
   platformBillingDueDays: envNumber('WORKER_PLATFORM_BILLING_DUE_DAYS', 15),
   platformBillingWorkerId: envString('WORKER_PLATFORM_BILLING_ID', `platform-billing-${process.pid}`),
+  platformUsageBatchSize: envNumber('WORKER_PLATFORM_USAGE_BATCH_SIZE', 100),
+  platformUsageWorkerId: envString('WORKER_PLATFORM_USAGE_ID', `platform-usage-${process.pid}`),
+  platformUsageMonth: envString('WORKER_PLATFORM_USAGE_MONTH', ''),
   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

@@ -51,6 +51,16 @@ async function runOnce() {
     );
     return;
   }
+  if (job === 'platform-usage') {
+    const { processPlatformUsageBatch } = await import('./jobs/platform-usage.js');
+    const result = await processPlatformUsageBatch();
+    console.log(
+      `[worker] platform-usage batch processed=${result.processed}`
+        + ` metrics=${result.metrics} created=${result.created}`
+        + ` updated=${result.updated} failed=${result.failed} skipped=${result.skipped}`,
+    );
+    return;
+  }
   if (job === 'platform-dunning') {
     const { processPlatformDunningBatch } = await import('./jobs/platform-dunning.js');
     const result = await processPlatformDunningBatch();

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

@@ -0,0 +1,509 @@
+import type pg from 'pg';
+import { pool } from '../db.js';
+import { config } from '../config.js';
+
+interface UsageTenant {
+  tenantId: string;
+  tenantSlug: string;
+  tenantName: string;
+}
+
+interface UsageMetric {
+  metricKey: string;
+  metricValue: string;
+  metadata: Record<string, unknown>;
+}
+
+export interface PlatformUsageWorkerResult {
+  processed: number;
+  metrics: number;
+  created: number;
+  updated: number;
+  failed: number;
+  skipped: 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 shanghaiMonth(value = new Date()) {
+  return new Intl.DateTimeFormat('en-CA', {
+    timeZone: 'Asia/Shanghai',
+    year: 'numeric',
+    month: '2-digit',
+  }).format(value);
+}
+
+function monthPeriod(monthText: string) {
+  const normalized = monthText.trim();
+  if (!MONTH_RE.test(normalized)) {
+    throw new Error(`Invalid platform usage month: ${monthText}. Expected YYYY-MM.`);
+  }
+  const [yearText, monthNumberText] = normalized.split('-');
+  const year = Number(yearText);
+  const monthNumber = Number(monthNumberText);
+  const periodStart = `${yearText}-${monthNumberText}-01`;
+  const periodEndDate = new Date(Date.UTC(year, monthNumber, 0));
+  const periodEnd = periodEndDate.toISOString().slice(0, 10);
+  return { periodStart, periodEnd };
+}
+
+function nowIso() {
+  return new Date().toISOString();
+}
+
+function errorMessage(error: unknown) {
+  return error instanceof Error ? error.message : String(error);
+}
+
+function errorCode(error: unknown) {
+  return typeof error === 'object' && error !== null && 'code' in error
+    ? String((error as { code?: unknown }).code || 'PLATFORM_USAGE_WORKER_ERROR')
+    : 'PLATFORM_USAGE_WORKER_ERROR';
+}
+
+function truncate(value: unknown, max = 1900) {
+  return String(value ?? '').slice(0, max);
+}
+
+async function loadTenants(limit: number, offset: number) {
+  const result = await pool.query<UsageTenant>(
+    `
+    select id as "tenantId", slug::text as "tenantSlug", name as "tenantName"
+    from public.tenants
+    where status = 'active'
+      and mode in ('saas', 'dedicated', 'platform_owned')
+    order by created_at asc
+    limit $1
+    offset $2
+    `,
+    [limit, offset],
+  );
+  return result.rows;
+}
+
+async function loadUsageMetrics(
+  client: pg.PoolClient,
+  tenant: UsageTenant,
+  periodStart: string,
+  periodEnd: string,
+): Promise<UsageMetric[]> {
+  const result = await client.query<{
+    studentCount: string;
+    activeStudents: string;
+    publishedQuestions: string;
+    draftQuestions: string;
+    activeAssets: string;
+    storageBytes: string;
+    verifiedStorageBytes: string;
+    videoCount: string;
+    videoPlays: string;
+    videoQuotaConsumed: string;
+    paidOrders: string;
+    paidOrderAmountCents: string;
+    activeEntitlements: string;
+  }>(
+    `
+    select
+      (
+        select count(distinct tm.user_id)::text
+        from public.tenant_memberships tm
+        where tm.tenant_id = $1
+          and tm.role = 'student'
+          and tm.status = 'active'
+      ) as "studentCount",
+      (
+        select count(distinct ar.user_id)::text
+        from public.answer_records ar
+        where ar.tenant_id = $1
+          and ar.answered_at >= $2::date
+          and ar.answered_at < ($3::date + interval '1 day')
+      ) as "activeStudents",
+      (
+        select count(*)::text
+        from public.questions q
+        where q.tenant_id = $1
+          and q.status = 'published'
+      ) as "publishedQuestions",
+      (
+        select count(*)::text
+        from public.questions q
+        where q.tenant_id = $1
+          and q.status = 'draft'
+      ) as "draftQuestions",
+      (
+        select count(*)::text
+        from public.content_assets a
+        where a.tenant_id = $1
+          and a.status <> 'archived'
+      ) as "activeAssets",
+      (
+        select coalesce(sum(coalesce(a.verified_size_bytes, a.file_size_bytes, 0)), 0)::text
+        from public.content_assets a
+        where a.tenant_id = $1
+          and a.status <> 'archived'
+          and (
+            a.upload_status in ('verified', 'not_required')
+            or a.storage_provider = 'external_url'
+          )
+          and a.security_scan_status in ('not_required', 'passed')
+      ) as "storageBytes",
+      (
+        select coalesce(sum(coalesce(a.verified_size_bytes, 0)), 0)::text
+        from public.content_assets a
+        where a.tenant_id = $1
+          and a.status <> 'archived'
+          and a.upload_status = 'verified'
+          and a.security_scan_status in ('not_required', 'passed')
+      ) as "verifiedStorageBytes",
+      (
+        select count(*)::text
+        from public.video_explanations v
+        where v.tenant_id = $1
+          and v.is_active = true
+      ) as "videoCount",
+      (
+        select count(*)::text
+        from public.video_play_events e
+        where e.tenant_id = $1
+          and e.created_at >= $2::date
+          and e.created_at < ($3::date + interval '1 day')
+      ) as "videoPlays",
+      (
+        select coalesce(sum(e.consumed_quota), 0)::text
+        from public.video_play_events e
+        where e.tenant_id = $1
+          and e.created_at >= $2::date
+          and e.created_at < ($3::date + interval '1 day')
+      ) as "videoQuotaConsumed",
+      (
+        select count(*)::text
+        from public.orders o
+        where o.tenant_id = $1
+          and o.status = 'paid'
+          and coalesce(o.paid_at, o.updated_at, o.created_at) >= $2::date
+          and coalesce(o.paid_at, o.updated_at, o.created_at) < ($3::date + interval '1 day')
+      ) as "paidOrders",
+      (
+        select coalesce(sum(o.amount_cents), 0)::text
+        from public.orders o
+        where o.tenant_id = $1
+          and o.status = 'paid'
+          and coalesce(o.paid_at, o.updated_at, o.created_at) >= $2::date
+          and coalesce(o.paid_at, o.updated_at, o.created_at) < ($3::date + interval '1 day')
+      ) as "paidOrderAmountCents",
+      (
+        select count(*)::text
+        from public.entitlements e
+        where e.tenant_id = $1
+          and e.status = 'active'
+          and e.starts_at < ($3::date + interval '1 day')
+          and (e.expires_at is null or e.expires_at >= $2::date)
+      ) as "activeEntitlements"
+    `,
+    [tenant.tenantId, periodStart, periodEnd],
+  );
+  const row = result.rows[0];
+  const storageBytes = Number(row?.storageBytes || 0);
+  const verifiedStorageBytes = Number(row?.verifiedStorageBytes || 0);
+
+  const baseMetadata = {
+    source: 'platform_usage_worker',
+    tenantSlug: tenant.tenantSlug,
+    tenantName: tenant.tenantName,
+    periodStart,
+    periodEnd,
+    collectedAt: nowIso(),
+    workerId: config.platformUsageWorkerId,
+  };
+
+  return [
+    {
+      metricKey: 'students',
+      metricValue: row?.studentCount || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'snapshot',
+        query: 'tenant_memberships.role=student,status=active,count_distinct_user',
+      },
+    },
+    {
+      metricKey: 'active_students',
+      metricValue: row?.activeStudents || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'period',
+        query: 'answer_records.answered_at,count_distinct_user',
+      },
+    },
+    {
+      metricKey: 'questions',
+      metricValue: row?.publishedQuestions || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'snapshot',
+        status: 'published',
+        draftQuestions: Number(row?.draftQuestions || 0),
+      },
+    },
+    {
+      metricKey: 'assets',
+      metricValue: row?.activeAssets || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'snapshot',
+        query: 'content_assets.status<>archived',
+      },
+    },
+    {
+      metricKey: 'storage_gb',
+      metricValue: (storageBytes / 1024 / 1024 / 1024).toFixed(6),
+      metadata: {
+        ...baseMetadata,
+        unit: 'gb',
+        scope: 'snapshot',
+        bytes: storageBytes,
+        verifiedBytes: verifiedStorageBytes,
+        query: 'content_assets verified/not_required and security passed/not_required',
+      },
+    },
+    {
+      metricKey: 'videos',
+      metricValue: row?.videoCount || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'snapshot',
+        query: 'video_explanations.is_active=true',
+      },
+    },
+    {
+      metricKey: 'video_plays',
+      metricValue: row?.videoPlays || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'period',
+        query: 'video_play_events.created_at',
+      },
+    },
+    {
+      metricKey: 'video_quota_consumed',
+      metricValue: row?.videoQuotaConsumed || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'period',
+        query: 'sum(video_play_events.consumed_quota)',
+      },
+    },
+    {
+      metricKey: 'paid_orders',
+      metricValue: row?.paidOrders || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'period',
+        query: 'orders.status=paid',
+      },
+    },
+    {
+      metricKey: 'paid_order_amount_cents',
+      metricValue: row?.paidOrderAmountCents || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'cents',
+        currency: 'CNY',
+        scope: 'period',
+        query: 'sum(orders.amount_cents where status=paid)',
+      },
+    },
+    {
+      metricKey: 'active_entitlements',
+      metricValue: row?.activeEntitlements || '0',
+      metadata: {
+        ...baseMetadata,
+        unit: 'count',
+        scope: 'period_intersection',
+        query: 'entitlements.status=active and overlaps period',
+      },
+    },
+  ];
+}
+
+async function upsertUsageMetric(
+  client: pg.PoolClient,
+  tenantId: string,
+  metric: UsageMetric,
+  periodStart: string,
+  periodEnd: string,
+) {
+  const existing = await client.query<{ id: string }>(
+    `
+    select id
+    from public.tenant_usage_records
+    where tenant_id = $1
+      and metric_key = $2
+      and period_start = $3::date
+      and period_end = $4::date
+      and metadata->>'source' = 'platform_usage_worker'
+    limit 1
+    for update
+    `,
+    [tenantId, metric.metricKey, periodStart, periodEnd],
+  );
+
+  if (existing.rows[0]) {
+    await client.query(
+      `
+      update public.tenant_usage_records
+      set metric_value = $5::numeric,
+          metadata = $6::jsonb
+      where id = $1
+        and tenant_id = $2
+        and metric_key = $3
+        and period_start = $4::date
+        and period_end = $7::date
+        and metadata->>'source' = 'platform_usage_worker'
+      `,
+      [
+        existing.rows[0].id,
+        tenantId,
+        metric.metricKey,
+        periodStart,
+        metric.metricValue,
+        JSON.stringify(metric.metadata),
+        periodEnd,
+      ],
+    );
+    return 'updated' as const;
+  }
+
+  await client.query(
+    `
+    insert into public.tenant_usage_records (
+      tenant_id, metric_key, metric_value, period_start, period_end, metadata
+    )
+    values ($1, $2, $3::numeric, $4::date, $5::date, $6::jsonb)
+    `,
+    [
+      tenantId,
+      metric.metricKey,
+      metric.metricValue,
+      periodStart,
+      periodEnd,
+      JSON.stringify(metric.metadata),
+    ],
+  );
+  return 'created' as const;
+}
+
+async function collectTenantUsage(
+  tenant: UsageTenant,
+  periodStart: string,
+  periodEnd: string,
+) {
+  const client = await pool.connect();
+  try {
+    await client.query('begin');
+    await client.query('select pg_advisory_xact_lock(hashtext($1))', [
+      `platform_usage_worker:${tenant.tenantId}:${periodStart}:${periodEnd}`,
+    ]);
+
+    const metrics = await loadUsageMetrics(client, tenant, periodStart, periodEnd);
+    let created = 0;
+    let updated = 0;
+    for (const metric of metrics) {
+      const status = await upsertUsageMetric(client, tenant.tenantId, metric, periodStart, periodEnd);
+      if (status === 'created') created += 1;
+      else updated += 1;
+    }
+
+    await client.query(
+      `
+      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+      values ($1, null, 'platform.usage.worker_collected', 'tenant_usage_records', $2, $3::jsonb)
+      `,
+      [
+        tenant.tenantId,
+        `${periodStart}:${periodEnd}`,
+        JSON.stringify({
+          periodStart,
+          periodEnd,
+          metricKeys: metrics.map(metric => metric.metricKey),
+          created,
+          updated,
+          workerId: config.platformUsageWorkerId,
+        }),
+      ],
+    );
+
+    await client.query('commit');
+    return { status: 'ok' as const, metrics: metrics.length, created, updated };
+  } catch (error) {
+    await client.query('rollback').catch(() => {});
+    await pool.query(
+      `
+      insert into public.audit_logs (tenant_id, actor_user_id, action, target_type, target_id, details)
+      values ($1, null, 'platform.usage.worker_failed', 'tenant_usage_records', $2, $3::jsonb)
+      `,
+      [
+        tenant.tenantId,
+        `${periodStart}:${periodEnd}`,
+        JSON.stringify({
+          periodStart,
+          periodEnd,
+          code: errorCode(error),
+          message: truncate(errorMessage(error)),
+          workerId: config.platformUsageWorkerId,
+        }),
+      ],
+    ).catch(() => {});
+    return { status: 'failed' as const, metrics: 0, created: 0, updated: 0 };
+  } finally {
+    client.release();
+  }
+}
+
+export async function processPlatformUsageBatch(options: {
+  limit?: number;
+  month?: string;
+} = {}): Promise<PlatformUsageWorkerResult> {
+  const pageSize = positiveInteger(options.limit ?? config.platformUsageBatchSize, 100, 1000);
+  const { periodStart, periodEnd } = monthPeriod(options.month || config.platformUsageMonth || shanghaiMonth());
+  const result: PlatformUsageWorkerResult = {
+    processed: 0,
+    metrics: 0,
+    created: 0,
+    updated: 0,
+    failed: 0,
+    skipped: 0,
+  };
+
+  for (let offset = 0; ; offset += pageSize) {
+    const tenants = await loadTenants(pageSize, offset);
+    if (tenants.length === 0) break;
+    result.processed += tenants.length;
+
+    for (const tenant of tenants) {
+      const tenantResult = await collectTenantUsage(tenant, periodStart, periodEnd);
+      if (tenantResult.status === 'failed') {
+        result.failed += 1;
+        continue;
+      }
+      result.metrics += tenantResult.metrics;
+      result.created += tenantResult.created;
+      result.updated += tenantResult.updated;
+    }
+  }
+
+  return result;
+}