import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
import { Prisma } from '@prisma/client';
import { moneyUnitsToFixedYuan } from '../common/money';
import { PrismaService } from '../prisma/prisma.service';

const SHANGHAI_OFFSET_MS = 8 * 60 * 60 * 1000;
const DAY_MS = 24 * 60 * 60 * 1000;
const DEFAULT_REFRESH_INTERVAL_MS = 60 * 60 * 1000;

export type ReportListQuery = {
  dateFrom?: string;
  dateTo?: string;
  tenantId?: string;
  applicationId?: string;
  channelId?: string;
  dimensionType?: string;
  page?: number;
  pageSize?: number;
};

@Injectable()
export class ReportsService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(ReportsService.name);
  private refreshTimer?: ReturnType<typeof setInterval>;
  private refreshRunning = false;
  private lastRefreshBusinessDate?: string;

  constructor(private readonly prisma: PrismaService) {}

  onModuleInit() {
    if (process.env.REPORT_DAILY_REFRESH_ENABLED === 'false') return;
    const startupTimer = setTimeout(() => void this.runScheduledRefresh(), 15_000);
    startupTimer.unref?.();
    this.refreshTimer = setInterval(
      () => void this.runScheduledRefresh(),
      positiveInteger(process.env.REPORT_REFRESH_INTERVAL_MS, DEFAULT_REFRESH_INTERVAL_MS),
    );
    this.refreshTimer.unref?.();
  }

  onModuleDestroy() {
    if (this.refreshTimer) clearInterval(this.refreshTimer);
  }

  async listReconciliation(query: ReportListQuery) {
    const { page, pageSize, skip } = pagination(query);
    const where = reconciliationWhere(query);
    const [items, total] = await Promise.all([
      this.prisma.dailyReconciliationReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }], skip, take: pageSize }),
      this.prisma.dailyReconciliationReport.count({ where }),
    ]);
    return { items, total, page, pageSize };
  }

  async listProfit(query: ReportListQuery) {
    const { page, pageSize, skip } = pagination(query);
    const { dimensionType, where } = profitWhere(query);
    const [items, total] = await Promise.all([
      this.prisma.dailyProfitReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }], skip, take: pageSize }),
      this.prisma.dailyProfitReport.count({ where }),
    ]);
    return { items, total, page, pageSize, dimensionType };
  }

  async listQuality(query: ReportListQuery) {
    const { page, pageSize, skip } = pagination(query);
    const { dimensionType, where } = qualityWhere(query);
    const [items, total] = await Promise.all([
      this.prisma.dailyQualityReport.findMany({ where, orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }], skip, take: pageSize }),
      this.prisma.dailyQualityReport.count({ where }),
    ]);
    return { items, total, page, pageSize, dimensionType };
  }

  async exportReconciliation(query: ReportListQuery) {
    const items = await this.prisma.dailyReconciliationReport.findMany({ where: reconciliationWhere(query), orderBy: [{ reportDate: 'desc' }, { tenantName: 'asc' }, { applicationName: 'asc' }] });
    return csvExport('对账单', ['发送日期', '企业', '企业应用', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.tenantName, item.applicationName, item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, formatCsvDate(item.generatedAt)]));
  }

  async exportProfit(query: ReportListQuery) {
    const { dimensionType, where } = profitWhere(query);
    const items = await this.prisma.dailyProfitReport.findMany({ where, orderBy: [{ reportDate: 'desc' }, { dimensionName: 'asc' }] });
    return csvExport(`利润报表-${dimensionType === 'channel' ? '通道' : '企业应用'}`, ['发送日期', '统计维度', '企业', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '净消费金额(元)', '返还金额(元)', '成本金额(元)', '利润(元)', '利润率(%)', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.dimensionName, item.tenantName ?? '', item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, moneyUnitsToFixedYuan(item.revenueCents), moneyUnitsToFixedYuan(item.refundCents), moneyUnitsToFixedYuan(item.costCents), moneyUnitsToFixedYuan(item.profitCents), (item.profitRateBps / 100).toFixed(2), formatCsvDate(item.generatedAt)]));
  }

  async exportQuality(query: ReportListQuery) {
    const { dimensionType, where } = qualityWhere(query);
    const items = await this.prisma.dailyQualityReport.findMany({ where, orderBy: [{ sentUnits: 'desc' }, { reportDate: 'desc' }, { dimensionName: 'asc' }] });
    return csvExport(`发送质量报表-${dimensionType}`, ['发送日期', '统计对象', '企业', '提交条数', '发送条数', '未知条数', '成功条数', '失败条数', '成功率(%)', '平均到达时长(毫秒)', '生成时间'], items.map((item) => [dateKey(item.reportDate), item.dimensionName, item.tenantName ?? '', item.submittedUnits, item.sentUnits, item.unknownUnits, item.successUnits, item.failedUnits, (item.successRateBps / 100).toFixed(2), item.avgArrivalMs ?? '', formatCsvDate(item.generatedAt)]));
  }

  async refreshRollingWindow(now = new Date()) {
    const days = completedBusinessDays(now, 4);
    for (const day of days) await this.refreshBusinessDay(day);
    return { refreshedDates: days.map((day) => day.key) };
  }

  private async runScheduledRefresh() {
    const businessDate = shanghaiDateKey(new Date());
    if (this.refreshRunning || this.lastRefreshBusinessDate === businessDate) return;
    this.refreshRunning = true;
    try {
      const result = await this.refreshRollingWindow();
      this.lastRefreshBusinessDate = businessDate;
      this.logger.log(`Daily reports refreshed for ${result.refreshedDates.join(', ')}`);
    } catch (error) {
      this.logger.error('Daily report refresh failed', error instanceof Error ? error.stack : String(error));
    } finally {
      this.refreshRunning = false;
    }
  }

  private async refreshBusinessDay(day: BusinessDay) {
    await this.prisma.$transaction(async (tx) => {
      await tx.dailyReconciliationReport.deleteMany({ where: { reportDate: day.reportDate } });
      await tx.dailyProfitReport.deleteMany({ where: { reportDate: day.reportDate } });
      await tx.dailyQualityReport.deleteMany({ where: { reportDate: day.reportDate } });

      await tx.$executeRaw(Prisma.sql`
        INSERT INTO "DailyReconciliationReport" (
          "id", "reportDate", "tenantId", "tenantName", "applicationId", "applicationName",
          "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "generatedAt", "updatedAt"
        )
        SELECT
          CONCAT('recon-', MD5(${day.key} || ':' || tenant.id || ':' || application.id)),
          ${day.reportDate}::date,
          tenant.id,
          tenant.name,
          application.id,
          application.name,
          COALESCE(SUM(message."billingUnits"), 0)::integer,
          COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected'
            AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
            AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
            THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
            AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
            THEN message."billingUnits" ELSE 0 END), 0)::integer,
          CURRENT_TIMESTAMP,
          CURRENT_TIMESTAMP
        FROM "SmsMessageRecord" message
        JOIN "Tenant" tenant ON tenant.id = message."tenantId"
        JOIN "SmsApplication" application ON application.id = message."applicationId"
        WHERE message."queuedAt" >= ${day.startAt}
          AND message."queuedAt" < ${day.endAt}
        GROUP BY tenant.id, tenant.name, application.id, application.name
      `);

      await tx.$executeRaw(Prisma.sql`
        WITH billing AS (
          SELECT "messageId",
            SUM(CASE WHEN "billingStatus" = 'charged' THEN "amountCents" ELSE 0 END)::bigint AS revenue,
            SUM(CASE WHEN "billingStatus" = 'refunded' THEN "amountCents" ELSE 0 END)::bigint AS refund
          FROM "SmsBillingRecord"
          GROUP BY "messageId"
        ), costs AS (
          SELECT
            submit."messageRecordId",
            SUM(submit."costUnitPrice" * CASE
              WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
              WHEN legacy_receipt.delivered THEN message."billingUnits"
              ELSE 0
            END)::bigint AS cost
          FROM "SmsSubmitRecord" submit
          JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
          LEFT JOIN LATERAL (
            SELECT
              COUNT(*)::integer AS audit_count,
              COUNT(*) FILTER (WHERE audit."receiptStatus" = 'delivered')::integer AS delivered_count
            FROM "SmsMessageSegmentAudit" audit
            WHERE audit."submitRecordId" = submit.id
          ) segment_receipts ON TRUE
          LEFT JOIN LATERAL (
            SELECT EXISTS (
              SELECT 1
              FROM "SmsReceiptRecord" receipt
              WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
                AND receipt."channelId" = submit."channelId"
                AND receipt."receiptStatus" = 'delivered'
            ) AS delivered
          ) legacy_receipt ON TRUE
          WHERE submit."submitStatus" = 'accepted'
          GROUP BY submit."messageRecordId"
        )
        INSERT INTO "DailyProfitReport" (
          "id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
          "tenantId", "tenantName", "applicationId", "channelId",
          "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "revenueCents", "refundCents", "costCents", "profitCents", "profitRateBps",
          "generatedAt", "updatedAt"
        )
        SELECT
          CONCAT('profit-app-', MD5(${day.key} || ':' || application.id)),
          ${day.reportDate}::date,
          'application',
          application.id,
          application.name,
          tenant.id,
          tenant.name,
          application.id,
          NULL,
          COALESCE(SUM(message."billingUnits"), 0)::integer,
          COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN COALESCE(message.status, '') <> 'rejected'
            AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
            AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
            THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
            AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
            THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(billing.revenue), 0)::bigint,
          COALESCE(SUM(billing.refund), 0)::bigint,
          COALESCE(SUM(costs.cost), 0)::bigint,
          (COALESCE(SUM(billing.revenue), 0) - COALESCE(SUM(costs.cost), 0))::bigint,
          CASE WHEN COALESCE(SUM(billing.revenue), 0) = 0 THEN 0
            ELSE ROUND((COALESCE(SUM(billing.revenue), 0) - COALESCE(SUM(costs.cost), 0)) * 10000.0 / SUM(billing.revenue))::integer END,
          CURRENT_TIMESTAMP,
          CURRENT_TIMESTAMP
        FROM "SmsMessageRecord" message
        JOIN "Tenant" tenant ON tenant.id = message."tenantId"
        JOIN "SmsApplication" application ON application.id = message."applicationId"
        LEFT JOIN billing ON billing."messageId" = message."messageId"
        LEFT JOIN costs ON costs."messageRecordId" = message.id
        WHERE message."queuedAt" >= ${day.startAt}
          AND message."queuedAt" < ${day.endAt}
        GROUP BY tenant.id, tenant.name, application.id, application.name
      `);

      await tx.$executeRaw(Prisma.sql`
        WITH billing AS (
          SELECT "messageId",
            SUM(CASE WHEN "billingStatus" = 'charged' THEN "amountCents" ELSE 0 END)::bigint AS revenue,
            SUM(CASE WHEN "billingStatus" = 'refunded' THEN "amountCents" ELSE 0 END)::bigint AS refund
          FROM "SmsBillingRecord"
          GROUP BY "messageId"
        )
        INSERT INTO "DailyProfitReport" (
          "id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
          "tenantId", "tenantName", "applicationId", "channelId",
          "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "revenueCents", "refundCents", "costCents", "profitCents", "profitRateBps",
          "generatedAt", "updatedAt"
        )
        SELECT
          CONCAT('profit-channel-', MD5(${day.key} || ':' || channel.id)),
          ${day.reportDate}::date,
          'channel',
          channel.id,
          channel.name,
          NULL,
          NULL,
          NULL,
          channel.id,
          COALESCE(SUM(message."billingUnits"), 0)::integer,
          COALESCE(SUM(message."billingUnits"), 0)::integer,
          COALESCE(SUM(CASE WHEN NOT EXISTS (
            SELECT 1 FROM "SmsReceiptRecord" receipt
            WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
              AND receipt."channelId" = submit."channelId"
              AND receipt."receiptStatus" IN ('delivered', 'undelivered')
          ) AND submit."submitStatus" NOT IN ('rejected', 'timeout') THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN EXISTS (
            SELECT 1 FROM "SmsReceiptRecord" receipt
            WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
              AND receipt."channelId" = submit."channelId"
              AND receipt."receiptStatus" = 'delivered'
          ) THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN NOT EXISTS (
            SELECT 1 FROM "SmsReceiptRecord" receipt
            WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
              AND receipt."channelId" = submit."channelId"
              AND receipt."receiptStatus" = 'delivered'
          ) AND (EXISTS (
            SELECT 1 FROM "SmsReceiptRecord" receipt
            WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
              AND receipt."channelId" = submit."channelId"
              AND receipt."receiptStatus" = 'undelivered'
          ) OR submit."submitStatus" IN ('rejected', 'timeout')) THEN message."billingUnits" ELSE 0 END), 0)::integer,
          COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.revenue ELSE 0 END), 0)::bigint,
          COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.refund ELSE 0 END), 0)::bigint,
          COALESCE(SUM(submit."costUnitPrice" * CASE
            WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
            WHEN legacy_receipt.delivered THEN message."billingUnits"
            ELSE 0
          END), 0)::bigint,
          (COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.revenue ELSE 0 END), 0) - COALESCE(SUM(submit."costUnitPrice" * CASE
            WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
            WHEN legacy_receipt.delivered THEN message."billingUnits"
            ELSE 0
          END), 0))::bigint,
          CASE WHEN COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.revenue ELSE 0 END), 0) = 0 THEN 0
            ELSE ROUND((COALESCE(SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.revenue ELSE 0 END), 0) - COALESCE(SUM(submit."costUnitPrice" * CASE
              WHEN segment_receipts.audit_count > 0 THEN segment_receipts.delivered_count
              WHEN legacy_receipt.delivered THEN message."billingUnits"
              ELSE 0
            END), 0)) * 10000.0 /
              SUM(CASE WHEN message."submitId" = submit."submitId" THEN billing.revenue ELSE 0 END))::integer END,
          CURRENT_TIMESTAMP,
          CURRENT_TIMESTAMP
        FROM "SmsSubmitRecord" submit
        JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
        JOIN "SmsChannel" channel ON channel.id = submit."channelId"
        LEFT JOIN billing ON billing."messageId" = message."messageId"
        LEFT JOIN LATERAL (
          SELECT
            COUNT(*)::integer AS audit_count,
            COUNT(*) FILTER (WHERE audit."receiptStatus" = 'delivered')::integer AS delivered_count
          FROM "SmsMessageSegmentAudit" audit
          WHERE audit."submitRecordId" = submit.id
        ) segment_receipts ON TRUE
        LEFT JOIN LATERAL (
          SELECT EXISTS (
            SELECT 1
            FROM "SmsReceiptRecord" receipt
            WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
              AND receipt."channelId" = submit."channelId"
              AND receipt."receiptStatus" = 'delivered'
          ) AS delivered
        ) legacy_receipt ON TRUE
        WHERE submit."submitStatus" = 'accepted'
          AND COALESCE(submit."submittedAt", submit."createdAt") >= ${day.startAt}
          AND COALESCE(submit."submittedAt", submit."createdAt") < ${day.endAt}
        GROUP BY channel.id, channel.name
      `);

      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'application'));
      await tx.$executeRaw(qualityByChannelSql(day));
      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'signature'));
      await tx.$executeRaw(qualityByMessageDimensionSql(day, 'drainage'));
    });
  }
}

function qualityByMessageDimensionSql(day: BusinessDay, dimensionType: 'application' | 'signature' | 'drainage') {
  const dimensionTypeSql = Prisma.raw(`'${dimensionType}'`);
  const dimensionId = dimensionType === 'application'
    ? Prisma.sql`application.id`
    : dimensionType === 'signature'
      ? Prisma.sql`COALESCE(signature.id, 'unmatched:' || COALESCE(application.id, tenant.id))`
      : Prisma.sql`COALESCE(drainage.id, 'unmatched:' || COALESCE(application.id, tenant.id))`;
  const dimensionName = dimensionType === 'application'
    ? Prisma.sql`application.name`
    : dimensionType === 'signature'
      ? Prisma.sql`COALESCE(signature.name, '未关联签名')`
      : Prisma.sql`COALESCE(drainage."siteName", '未关联引流信息')`;
  const applicationJoin = dimensionType === 'application'
    ? Prisma.sql`JOIN "SmsApplication" application ON application.id = message."applicationId"`
    : Prisma.sql`LEFT JOIN "SmsApplication" application ON application.id = message."applicationId"`;

  return Prisma.sql`
    WITH base AS (
      SELECT
        ${dimensionId} AS dimension_id,
        ${dimensionName} AS dimension_name,
        tenant.id AS tenant_id,
        tenant.name AS tenant_name,
        application.id AS application_id,
        message."billingUnits" AS billing_units,
        CASE WHEN COALESCE(message.status, '') <> 'rejected' THEN message."billingUnits" ELSE 0 END AS sent_units,
        CASE WHEN message.status = 'delivered' OR message."receiptStatus" = 'delivered' THEN message."billingUnits" ELSE 0 END AS success_units,
        CASE WHEN NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
          AND (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
          THEN message."billingUnits" ELSE 0 END AS failed_units,
        CASE WHEN COALESCE(message.status, '') <> 'rejected'
          AND NOT (COALESCE(message.status = 'delivered', false) OR COALESCE(message."receiptStatus" = 'delivered', false))
          AND NOT (COALESCE(message.status IN ('submit_failed', 'failed', 'timeout'), false) OR COALESCE(message."receiptStatus" = 'undelivered', false))
          THEN message."billingUnits" ELSE 0 END AS unknown_units,
        CASE WHEN (message.status = 'delivered' OR message."receiptStatus" = 'delivered')
          AND message."submittedAt" IS NOT NULL AND message."deliveredAt" >= message."submittedAt"
          THEN EXTRACT(EPOCH FROM (message."deliveredAt" - message."submittedAt")) * 1000 END AS arrival_ms
      FROM "SmsMessageRecord" message
      JOIN "Tenant" tenant ON tenant.id = message."tenantId"
      ${applicationJoin}
      LEFT JOIN "SmsSignature" signature ON signature.id = message."signatureId"
      LEFT JOIN "SmsDrainageInfo" drainage ON drainage.id = message."drainageInfoId"
      WHERE message."queuedAt" >= ${day.startAt}
        AND message."queuedAt" < ${day.endAt}
    ), thresholds AS (
      SELECT dimension_id, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY arrival_ms) AS p95_ms
      FROM base WHERE arrival_ms IS NOT NULL GROUP BY dimension_id
    )
    INSERT INTO "DailyQualityReport" (
      "id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
      "tenantId", "tenantName", "applicationId", "channelId", "signatureId", "drainageInfoId",
      "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "successRateBps", "avgArrivalMs", "generatedAt", "updatedAt"
    )
    SELECT
      CONCAT('quality-', ${dimensionTypeSql}, '-', MD5(${day.key} || ':' || base.dimension_id)),
      ${day.reportDate}::date,
      ${dimensionTypeSql},
      base.dimension_id,
      MAX(base.dimension_name),
      MAX(base.tenant_id),
      MAX(base.tenant_name),
      CASE WHEN ${dimensionTypeSql} = 'application' THEN base.dimension_id ELSE MAX(base.application_id) END,
      NULL,
      CASE WHEN ${dimensionTypeSql} = 'signature' AND base.dimension_id NOT LIKE 'unmatched:%' THEN base.dimension_id ELSE NULL END,
      CASE WHEN ${dimensionTypeSql} = 'drainage' AND base.dimension_id NOT LIKE 'unmatched:%' THEN base.dimension_id ELSE NULL END,
      SUM(base.billing_units)::integer,
      SUM(base.sent_units)::integer,
      SUM(base.unknown_units)::integer,
      SUM(base.success_units)::integer,
      SUM(base.failed_units)::integer,
      CASE WHEN SUM(base.sent_units) = 0 THEN 0 ELSE ROUND(SUM(base.success_units) * 10000.0 / SUM(base.sent_units))::integer END,
      ROUND(AVG(base.arrival_ms) FILTER (WHERE base.arrival_ms IS NOT NULL AND base.arrival_ms <= thresholds.p95_ms))::integer,
      CURRENT_TIMESTAMP,
      CURRENT_TIMESTAMP
    FROM base
    LEFT JOIN thresholds ON thresholds.dimension_id = base.dimension_id
    GROUP BY base.dimension_id
  `;
}

function qualityByChannelSql(day: BusinessDay) {
  return Prisma.sql`
    WITH base AS (
      SELECT
        channel.id AS dimension_id,
        channel.name AS dimension_name,
        message."billingUnits" AS billing_units,
        CASE WHEN receipt."deliveredAt" IS NOT NULL THEN message."billingUnits" ELSE 0 END AS success_units,
        CASE WHEN receipt."deliveredAt" IS NULL AND failed_receipt."failedAt" IS NOT NULL THEN message."billingUnits" ELSE 0 END AS failed_units,
        CASE WHEN receipt."deliveredAt" IS NULL AND failed_receipt."failedAt" IS NULL THEN message."billingUnits" ELSE 0 END AS unknown_units,
        CASE WHEN receipt."deliveredAt" >= COALESCE(submit."submittedAt", submit."createdAt")
          THEN EXTRACT(EPOCH FROM (receipt."deliveredAt" - COALESCE(submit."submittedAt", submit."createdAt"))) * 1000 END AS arrival_ms
      FROM "SmsSubmitRecord" submit
      JOIN "SmsMessageRecord" message ON message.id = submit."messageRecordId"
      JOIN "SmsChannel" channel ON channel.id = submit."channelId"
      LEFT JOIN LATERAL (
        SELECT MIN(receipt."deliveredAt") AS "deliveredAt"
        FROM "SmsReceiptRecord" receipt
        WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
          AND receipt."channelId" = submit."channelId"
          AND receipt."receiptStatus" = 'delivered'
      ) receipt ON TRUE
      LEFT JOIN LATERAL (
        SELECT MIN(receipt."deliveredAt") AS "failedAt"
        FROM "SmsReceiptRecord" receipt
        WHERE receipt."gatewayMessageId" = submit."gatewayMessageId"
          AND receipt."channelId" = submit."channelId"
          AND receipt."receiptStatus" = 'undelivered'
      ) failed_receipt ON TRUE
      WHERE submit."submitStatus" = 'accepted'
        AND COALESCE(submit."submittedAt", submit."createdAt") >= ${day.startAt}
        AND COALESCE(submit."submittedAt", submit."createdAt") < ${day.endAt}
    ), thresholds AS (
      SELECT dimension_id, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY arrival_ms) AS p95_ms
      FROM base WHERE arrival_ms IS NOT NULL GROUP BY dimension_id
    )
    INSERT INTO "DailyQualityReport" (
      "id", "reportDate", "dimensionType", "dimensionId", "dimensionName",
      "tenantId", "tenantName", "applicationId", "channelId", "signatureId", "drainageInfoId",
      "submittedUnits", "sentUnits", "unknownUnits", "successUnits", "failedUnits", "successRateBps", "avgArrivalMs", "generatedAt", "updatedAt"
    )
    SELECT
      CONCAT('quality-channel-', MD5(${day.key} || ':' || base.dimension_id)),
      ${day.reportDate}::date,
      'channel',
      base.dimension_id,
      MAX(base.dimension_name),
      NULL, NULL, NULL, base.dimension_id, NULL, NULL,
      SUM(base.billing_units)::integer,
      SUM(base.billing_units)::integer,
      SUM(base.unknown_units)::integer,
      SUM(base.success_units)::integer,
      SUM(base.failed_units)::integer,
      CASE WHEN SUM(base.billing_units) = 0 THEN 0 ELSE ROUND(SUM(base.success_units) * 10000.0 / SUM(base.billing_units))::integer END,
      ROUND(AVG(base.arrival_ms) FILTER (WHERE base.arrival_ms IS NOT NULL AND base.arrival_ms <= thresholds.p95_ms))::integer,
      CURRENT_TIMESTAMP,
      CURRENT_TIMESTAMP
    FROM base
    LEFT JOIN thresholds ON thresholds.dimension_id = base.dimension_id
    GROUP BY base.dimension_id
  `;
}

type BusinessDay = { key: string; reportDate: Date; startAt: Date; endAt: Date };

function completedBusinessDays(now: Date, count: number): BusinessDay[] {
  const shifted = new Date(now.getTime() + SHANGHAI_OFFSET_MS);
  const today = Date.UTC(shifted.getUTCFullYear(), shifted.getUTCMonth(), shifted.getUTCDate());
  return Array.from({ length: count }, (_, index) => businessDay(today - (count - index) * DAY_MS));
}

function businessDay(localDateUtc: number): BusinessDay {
  const reportDate = new Date(localDateUtc);
  return {
    key: reportDate.toISOString().slice(0, 10),
    reportDate,
    startAt: new Date(localDateUtc - SHANGHAI_OFFSET_MS),
    endAt: new Date(localDateUtc - SHANGHAI_OFFSET_MS + DAY_MS),
  };
}

function shanghaiDateKey(now: Date) {
  return new Date(now.getTime() + SHANGHAI_OFFSET_MS).toISOString().slice(0, 10);
}

function dateFilter(from?: string, to?: string): Prisma.DateTimeFilter | undefined {
  const gte = parseDate(from);
  const lte = parseDate(to);
  if (!gte && !lte) return undefined;
  return { gte, lte };
}

function parseDate(value?: string) {
  if (!value || !/^\d{4}-\d{2}-\d{2}$/.test(value)) return undefined;
  const date = new Date(`${value}T00:00:00.000Z`);
  return Number.isNaN(date.getTime()) ? undefined : date;
}

function pagination(query: ReportListQuery) {
  const page = Math.max(1, Math.floor(Number(query.page) || 1));
  const pageSize = Math.min(100, Math.max(1, Math.floor(Number(query.pageSize) || 20)));
  return { page, pageSize, skip: (page - 1) * pageSize };
}

function reconciliationWhere(query: ReportListQuery): Prisma.DailyReconciliationReportWhereInput {
  return { reportDate: dateFilter(query.dateFrom, query.dateTo), tenantId: query.tenantId || undefined, applicationId: query.applicationId || undefined };
}

function profitWhere(query: ReportListQuery) {
  const dimensionType = query.dimensionType === 'channel' ? 'channel' : 'application';
  const where: Prisma.DailyProfitReportWhereInput = {
    dimensionType,
    reportDate: dateFilter(query.dateFrom, query.dateTo),
    tenantId: dimensionType === 'application' ? query.tenantId || undefined : undefined,
    applicationId: dimensionType === 'application' ? query.applicationId || undefined : undefined,
    channelId: dimensionType === 'channel' ? query.channelId || undefined : undefined,
  };
  return { dimensionType, where };
}

function qualityWhere(query: ReportListQuery) {
  const allowedDimensions = new Set(['application', 'channel', 'signature', 'drainage']);
  const dimensionType = allowedDimensions.has(String(query.dimensionType)) ? String(query.dimensionType) : 'application';
  const where: Prisma.DailyQualityReportWhereInput = {
    dimensionType,
    reportDate: dateFilter(query.dateFrom, query.dateTo),
    tenantId: query.tenantId || undefined,
    applicationId: query.applicationId || undefined,
    channelId: query.channelId || undefined,
  };
  return { dimensionType, where };
}

function csvExport(name: string, headers: string[], rows: Array<Array<string | number>>) {
  const content = [headers, ...rows].map((row) => row.map(csvCell).join(',')).join('\n');
  return { fileName: `${name}-${shanghaiDateKey(new Date())}.csv`, content };
}

function csvCell(value: string | number) {
  const text = String(value);
  return /[",\r\n]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
}

function dateKey(value: Date | string) {
  return (value instanceof Date ? value.toISOString() : String(value)).slice(0, 10);
}

function formatCsvDate(value: Date | string) {
  return value instanceof Date ? value.toISOString() : String(value);
}

function positiveInteger(value: string | undefined, fallback: number) {
  const parsed = Number(value);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}