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