import { BadRequestException, NotFoundException } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { randomUUID } from 'node:crypto';
import { moneyToNumber } from '../../common/money';
import { PrismaService } from '../../prisma/prisma.service';
import type { MessageQuery, TraceQuery, OperationLogQuery, GatewaySubmitDeadLetterQuery, DownstreamDeliveryQuery, DownstreamDeliveryDashboardQuery, DownstreamRecoveryStatusQuery, MessageSegmentAuditQuery, SignatureQualityQuery } from '../operations.contracts';
import { messageWhere, recognizedCarrierValues, carrierWhere, startOfShanghaiDay, endOfShanghaiDay, qualityBusinessDay, shanghaiDateKey, normalizeGroupBy, returnedTransactionWhere, createdAtRange, downstreamAlertPendingMinutes, downstreamAlertRecentFailedHours, downstreamAlertWindows, downstreamAlertWhere, stalledPendingWhere, downstreamDeliveryScopedWhere, parseDateBoundary, downstreamRecoveryStatusWhere, escapeCsvCell, formatCsvDate, formatExportTimestamp, clientApplicationView, clientReceiptView, clientMessageView, clientBatchTaskView, clientUplinkView, clientAccountView, clientRechargeView, summarizeMessageGroups, groupDownstreamByType, groupDownstreamByApplication, positiveInteger, operationLogLevelWhere, normalizeOperationLog, sanitizeGatewaySubmitException, redactGatewayCommandValue } from '../operations.helpers';

// R2 dashboard query domain. Method bodies are preserved byte-for-byte from the facade baseline.
export class OperationsDashboardQueries {
  constructor(private readonly prisma: PrismaService) {}

async dashboard(query: { tenantId?: string }) {
    const businessDay = qualityBusinessDay();
    const sinceToday = businessDay.startAt;
    const downstreamAlertWindow = downstreamAlertWindows();
    const messageWhereClause = messageWhere({ tenantId: query.tenantId });
    const todayMessageWhereClause = {
      ...messageWhereClause,
      queuedAt: { gte: sinceToday, lt: businessDay.endAt },
    };
    const [
      taskCount,
      messageGroups,
      todayMessageGroups,
      uplinkCount,
      billingAggregate,
      transactionAggregate,
      connectionGroups,
      pendingAudits,
      tenantAccounts,
      recentTasks,
      recentRecharges,
      enterpriseSpendRows,
      downstreamPendingCount,
      downstreamFailedCount,
      downstreamDeliveredCount,
      downstreamStalledPendingCount,
      downstreamStalledAckCount,
      downstreamRecentFailedCount,
      hourlySendRows,
      auditSpeedRows,
    ] = await Promise.all([
      this.prisma.smsBatchTask.count({ where: { tenantId: query.tenantId } }),
      this.prisma.smsMessageRecord.groupBy({
        by: ['status'],
        where: messageWhereClause,
        _count: { _all: true },
        _sum: { amountCents: true, billingUnits: true },
      }),
      this.prisma.smsMessageRecord.groupBy({
        by: ['status'],
        where: todayMessageWhereClause,
        _count: { _all: true },
        _sum: { amountCents: true, billingUnits: true },
      }),
      this.prisma.smsUplinkMessage.count({ where: { tenantId: query.tenantId } }),
      this.prisma.smsBillingRecord.aggregate({
        where: { tenantId: query.tenantId },
        _sum: { amountCents: true, billingUnits: true },
        _count: { _all: true },
      }),
      this.prisma.accountTransaction.aggregate({
        where: returnedTransactionWhere(sinceToday, query.tenantId),
        _sum: { amountCents: true },
        _count: { _all: true },
      }),
      this.prisma.cmppConnectionState.groupBy({
        by: ['status'],
        where: { tenantId: query.tenantId },
        _count: { _all: true },
        _sum: { currentConnections: true, desiredConnections: true },
      }),
      this.pendingAudits(query.tenantId),
      this.prisma.tenantAccount.findMany({
        where: query.tenantId ? { tenantId: query.tenantId } : undefined,
        include: { tenant: true },
        orderBy: { updatedAt: 'desc' },
        take: 20,
      }),
      this.prisma.smsBatchTask.findMany({
        where: query.tenantId ? { tenantId: query.tenantId } : undefined,
        include: { application: true, messages: { take: 1, include: { channel: true } } },
        orderBy: { createdAt: 'desc' },
        take: 10,
      }),
      this.prisma.rechargeOrder.findMany({
        where: {
          tenantId: query.tenantId,
          payMethod: 'manual_topup',
        },
        include: { tenant: true },
        orderBy: { createdAt: 'desc' },
        take: 10,
      }),
      this.prisma.$queryRaw<Array<{
        tenantId: string;
        tenantName: string;
        todaySpendCents: bigint;
        balanceCents: bigint;
        creditCents: bigint;
      }>>(Prisma.sql`
        SELECT
          tenant.id AS "tenantId",
          tenant.name AS "tenantName",
          COALESCE(SUM(billing."amountCents") FILTER (WHERE billing."billingStatus" = 'charged'), 0)::bigint AS "todaySpendCents",
          account."balanceCents" AS "balanceCents",
          account."creditCents" AS "creditCents"
        FROM "TenantAccount" account
        JOIN "Tenant" tenant ON tenant.id = account."tenantId"
        LEFT JOIN "SmsBillingRecord" billing
          ON billing."tenantId" = tenant.id
         AND billing."createdAt" >= ${businessDay.startAt}
         AND billing."createdAt" < ${businessDay.endAt}
        WHERE tenant.status <> 'deleted'
          AND (${query.tenantId ?? null}::text IS NULL OR tenant.id = ${query.tenantId ?? null})
        GROUP BY tenant.id, tenant.name, account."balanceCents", account."creditCents"
        ORDER BY "todaySpendCents" DESC, tenant.name ASC
      `),
      this.prisma.cmppDownstreamDelivery.count({
        where: { tenantId: query.tenantId, status: 'pending' },
      }),
      this.prisma.cmppDownstreamDelivery.count({
        where: { tenantId: query.tenantId, status: 'failed' },
      }),
      this.prisma.cmppDownstreamDelivery.count({
        where: { tenantId: query.tenantId, status: 'delivered' },
      }),
      this.prisma.cmppDownstreamDelivery.count({
        where: {
          tenantId: query.tenantId,
          ...stalledPendingWhere(downstreamAlertWindow.stalledPendingAt),
        },
      }),
      this.prisma.cmppDownstreamDelivery.count({
        where: {
          tenantId: query.tenantId,
          status: 'awaiting_ack',
          ackDeadlineAt: { lte: downstreamAlertWindow.now },
        },
      }),
      this.prisma.cmppDownstreamDelivery.count({
        where: {
          tenantId: query.tenantId,
          status: { in: ['failed', 'unconfirmed', 'rejected'] },
          updatedAt: { gte: downstreamAlertWindow.recentFailedAt },
        },
      }),
      this.prisma.$queryRaw<Array<{
        hour: number;
        submittedCount: bigint;
        successCount: bigint;
      }>>(Prisma.sql`
        SELECT
          EXTRACT(
            HOUR FROM (message."queuedAt" AT TIME ZONE 'UTC') AT TIME ZONE 'Asia/Shanghai'
          )::integer AS hour,
          COUNT(*)::bigint AS "submittedCount",
          COUNT(*) FILTER (WHERE message.status = 'delivered')::bigint AS "successCount"
        FROM "SmsMessageRecord" message
        WHERE message."queuedAt" >= ${businessDay.startAt}
          AND message."queuedAt" < ${businessDay.endAt}
          AND (${query.tenantId ?? null}::text IS NULL OR message."tenantId" = ${query.tenantId ?? null})
        GROUP BY 1
        ORDER BY 1
      `),
      // Signature/template tables have no review timestamps, so their latest pending audit is paired with the review audit.
      this.prisma.$queryRaw<Array<{
        category: string;
        count: bigint;
        averageProcessingMs: bigint | null;
      }>>(Prisma.sql`
        WITH review_samples AS (
          SELECT
            'enterpriseCertifications'::text AS category,
            certification."submittedAt" AS "submittedAt",
            certification."reviewedAt" AS "reviewedAt"
          FROM "EnterpriseCertification" certification
          WHERE certification."reviewedAt" >= ${businessDay.startAt}
            AND certification."reviewedAt" < ${businessDay.endAt}
            AND (${query.tenantId ?? null}::text IS NULL OR certification."tenantId" = ${query.tenantId ?? null})

          UNION ALL

          SELECT
            'smsAudits'::text,
            task."createdAt",
            task."reviewedAt"
          FROM "SmsSendTask" task
          WHERE task."reviewedAt" >= ${businessDay.startAt}
            AND task."reviewedAt" < ${businessDay.endAt}
            AND (${query.tenantId ?? null}::text IS NULL OR task."tenantId" = ${query.tenantId ?? null})

          UNION ALL

          SELECT
            'drainageInfos'::text,
            drainage."submittedAt",
            drainage."reviewedAt"
          FROM "SmsDrainageInfo" drainage
          WHERE drainage."reviewedAt" >= ${businessDay.startAt}
            AND drainage."reviewedAt" < ${businessDay.endAt}
            AND (${query.tenantId ?? null}::text IS NULL OR drainage."tenantId" = ${query.tenantId ?? null})

          UNION ALL

          SELECT
            CASE review."targetType"
              WHEN 'sms_signature' THEN 'signatures'
              WHEN 'sms_template' THEN 'templates'
            END,
            submission."createdAt",
            review."createdAt"
          FROM "AuditRecord" review
          JOIN LATERAL (
            SELECT pending."createdAt"
            FROM "AuditRecord" pending
            WHERE pending."targetType" = review."targetType"
              AND pending."targetId" = review."targetId"
              AND pending."statusAfter" = 'pending'
              AND pending."createdAt" <= review."createdAt"
            ORDER BY pending."createdAt" DESC
            LIMIT 1
          ) submission ON true
          WHERE review."targetType" IN ('sms_signature', 'sms_template')
            AND review."statusBefore" = 'pending'
            AND review."statusAfter" IN ('approved', 'rejected')
            AND review."createdAt" >= ${businessDay.startAt}
            AND review."createdAt" < ${businessDay.endAt}
            AND (${query.tenantId ?? null}::text IS NULL OR review."tenantId" = ${query.tenantId ?? null})
        )
        SELECT
          category,
          COUNT(*)::bigint AS count,
          ROUND(AVG(EXTRACT(EPOCH FROM ("reviewedAt" - "submittedAt")) * 1000))::bigint AS "averageProcessingMs"
        FROM review_samples
        WHERE "reviewedAt" >= "submittedAt"
        GROUP BY category
      `),
    ]);
    const todayTotals = summarizeMessageGroups(todayMessageGroups);
    const hourlyRowsByHour = new Map(hourlySendRows.map((row) => [Number(row.hour), row]));
    // Always return all 24 Shanghai-time buckets so the line chart does not imply missing hours are missing data.
    const hourlySendTrend = Array.from({ length: 24 }, (_, hour) => {
      const row = hourlyRowsByHour.get(hour);
      return {
        hour,
        label: `${String(hour).padStart(2, '0')}:00`,
        submittedCount: Number(row?.submittedCount ?? 0),
        successCount: Number(row?.successCount ?? 0),
      };
    });
    const auditSpeedByCategory = new Map(auditSpeedRows.map((row) => [row.category, row]));
    const auditProcessingSpeed = [
      ['enterpriseCertifications', '企业认证'],
      ['smsAudits', '短信审核'],
      ['templates', '模板'],
      ['signatures', '签名'],
      ['drainageInfos', '引流信息'],
    ].map(([category, label]) => {
      const row = auditSpeedByCategory.get(category);
      return {
        category,
        label,
        count: Number(row?.count ?? 0),
        averageProcessingMs: row?.averageProcessingMs == null ? null : Number(row.averageProcessingMs),
      };
    });
    const downstreamAlertCount = downstreamStalledPendingCount + downstreamStalledAckCount + downstreamRecentFailedCount;
    return {
      taskCount,
      messageStatus: messageGroups,
      today: {
        sent: todayTotals.total,
        delivered: todayTotals.delivered,
        failed: todayTotals.failed,
        unknown: todayTotals.unknown,
        successRate: todayTotals.total > 0 ? Number(((todayTotals.delivered / todayTotals.total) * 100).toFixed(1)) : 0,
        spendCents: todayTotals.amountCents,
        returnedCents: moneyToNumber(transactionAggregate._sum.amountCents),
        billingUnits: todayTotals.billingUnits,
      },
      uplinkCount,
      billing: billingAggregate,
      transactions: transactionAggregate,
      gatewayConnections: connectionGroups,
      pendingAuditCount: pendingAudits.total,
      pendingAudits,
      hourlySendTrend,
      auditProcessingSpeed,
      downstreamDeliverySummary: {
        pending: downstreamPendingCount,
        failed: downstreamFailedCount,
        delivered: downstreamDeliveredCount,
        stalledPending: downstreamStalledPendingCount,
        stalledAck: downstreamStalledAckCount,
        recentFailed: downstreamRecentFailedCount,
        alertCount: downstreamAlertCount,
      },
      accounts: tenantAccounts,
      enterpriseSpendRanks: enterpriseSpendRows.map((row) => ({
        tenantId: row.tenantId,
        tenantName: row.tenantName,
        todaySpendCents: moneyToNumber(row.todaySpendCents),
        balanceCents: moneyToNumber(row.balanceCents),
        creditCents: moneyToNumber(row.creditCents),
      })),
      recentTasks,
      recentRecharges,
    };
  }
async clientDashboard(query: { tenantId?: string }) {
    const tenantId = query.tenantId;
    const [dashboard, tenant, approvedCertification, signatureCount, pendingBatchTaskCount] = await Promise.all([
      this.dashboard(query),
      tenantId
        ? this.prisma.tenant.findUnique({ where: { id: tenantId }, select: { id: true, name: true } })
        : Promise.resolve(null),
      tenantId
        ? this.prisma.enterpriseCertification.findFirst({
            where: { tenantId, status: 'approved' },
            select: { id: true },
          })
        : Promise.resolve(null),
      tenantId
        ? this.prisma.smsSignature.count({
            where: { tenantId, auditStatus: { notIn: ['deleted', 'disabled'] } },
          })
        : Promise.resolve(0),
      tenantId
        ? this.prisma.smsBatchTask.count({
            where: { tenantId, sourceType: 'client', status: 'pending_review' },
          })
        : Promise.resolve(0),
    ]);
    return {
      taskCount: dashboard.taskCount,
      messageStatus: dashboard.messageStatus,
      today: dashboard.today,
      uplinkCount: dashboard.uplinkCount,
      billing: dashboard.billing,
      transactions: dashboard.transactions,
      gatewayConnections: [],
      pendingAuditCount: dashboard.pendingAuditCount,
      pendingAudits: dashboard.pendingAudits,
      hourlySendTrend: dashboard.hourlySendTrend,
      auditProcessingSpeed: dashboard.auditProcessingSpeed,
      downstreamDeliverySummary: dashboard.downstreamDeliverySummary,
      accounts: dashboard.accounts.map(clientAccountView),
      enterpriseSpendRanks: dashboard.enterpriseSpendRanks,
      recentTasks: dashboard.recentTasks.map(clientBatchTaskView),
      recentRecharges: dashboard.recentRecharges.map(clientRechargeView),
      clientOverview: {
        enterpriseName: tenant?.name ?? null,
        certificationStatus: approvedCertification ? 'certified' : 'uncertified',
        signatureCount,
        pendingBatchTaskCount,
      },
    };
  }
pendingAudits(tenantId?: string) {
    return Promise.all([
      this.prisma.smsTemplate.count({ where: { tenantId, auditStatus: 'pending' } }),
      this.prisma.smsSignature.count({ where: { tenantId, auditStatus: 'pending' } }),
      this.prisma.smsDrainageInfo.count({ where: { tenantId, auditStatus: 'pending' } }),
      this.prisma.enterpriseCertification.count({ where: { tenantId, status: 'pending' } }),
      this.prisma.smsSendTask.count({ where: { tenantId, status: 'pending_review' } }),
    ]).then(([templates, signatures, drainageInfos, enterpriseCertifications, smsAudits]) => ({
      templates,
      signatures,
      drainageInfos,
      enterpriseCertifications,
      smsAudits,
      total: templates + signatures + drainageInfos + enterpriseCertifications + smsAudits,
    }));
  }
}