-- Petstore Pilot Metrics v2（MySQL 5.7+/8.x，只读）
-- 默认时间窗为最近 30 天，但默认门店集合为 0（不返回业务数据）。
-- 执行人必须把 @pilot_store_ids 显式改为 3～5 个试点门店内部 ID，例如 '101,205,309'。
-- 输出仅含 store_id、计数、比率和时效，不含客户/员工 PII。

SET @pilot_to = DATE_ADD(CURRENT_DATE, INTERVAL 1 DAY);
SET @pilot_from = DATE_SUB(@pilot_to, INTERVAL 30 DAY);
SET @pilot_store_ids = '0';

-- 1. 核心服务与报告漏斗。[from, to)，各事实按稳定 aggregate_id 去重。
-- 比率只认同窗且时间顺序正确的 cohort，避免跨窗口事件把比率推到 100% 以上。
SELECT
    e.store_id,
    COUNT(DISTINCT CASE WHEN e.event_type = 'appointment_created' THEN e.aggregate_id END) AS appointment_created_count,
    COUNT(DISTINCT CASE WHEN e.event_type = 'service_started' THEN e.aggregate_id END) AS service_started_count,
    COUNT(DISTINCT CASE WHEN e.event_type = 'service_completed' THEN e.aggregate_id END) AS complete_service_count,
    COUNT(DISTINCT CASE WHEN e.event_type = 'report_submitted' THEN e.aggregate_id END) AS report_submitted_count,
    COUNT(DISTINCT CASE
        WHEN e.event_type = 'report_sent'
         AND EXISTS (
            SELECT 1 FROM t_business_event submitted
            WHERE submitted.store_id = e.store_id
              AND submitted.event_type = 'report_submitted'
              AND submitted.aggregate_type = 'report'
              AND submitted.aggregate_id = e.aggregate_id
              AND submitted.occurred_at >= @pilot_from
              AND submitted.occurred_at <= e.occurred_at
         )
        THEN e.aggregate_id
    END) AS report_sent_count,
    COUNT(DISTINCT CASE
        WHEN e.event_type IN ('report_opened', 'report_reopened')
         AND EXISTS (
            SELECT 1
            FROM t_business_event sent
            JOIN t_business_event submitted
              ON submitted.store_id = sent.store_id
             AND submitted.event_type = 'report_submitted'
             AND submitted.aggregate_type = 'report'
             AND submitted.aggregate_id = sent.aggregate_id
             AND submitted.occurred_at >= @pilot_from
             AND submitted.occurred_at <= sent.occurred_at
            WHERE sent.store_id = e.store_id
              AND sent.event_type = 'report_sent'
              AND sent.aggregate_type = 'report'
              AND sent.aggregate_id = e.aggregate_id
              AND sent.occurred_at >= @pilot_from
              AND sent.occurred_at <= e.occurred_at
         )
        THEN e.aggregate_id
    END) AS report_opened_count,
    COUNT(DISTINCT CASE WHEN e.event_type = 'lead_submitted' THEN e.aggregate_id END) AS lead_submitted_count,
    COUNT(DISTINCT CASE WHEN e.event_type = 'rebook_created' THEN e.aggregate_id END) AS rebook_created_count,
    ROUND(
        100 * COUNT(DISTINCT CASE
            WHEN e.event_type = 'service_completed'
             AND EXISTS (
                SELECT 1 FROM t_business_event started
                WHERE started.store_id = e.store_id
                  AND started.event_type = 'service_started'
                  AND started.aggregate_type = 'appointment'
                  AND started.aggregate_id = e.aggregate_id
                  AND started.occurred_at >= @pilot_from
                  AND started.occurred_at <= e.occurred_at
             )
            THEN e.aggregate_id
        END)
        / NULLIF(COUNT(DISTINCT CASE WHEN e.event_type = 'service_started' THEN e.aggregate_id END), 0),
        1
    ) AS service_completion_rate_pct,
    ROUND(
        100 * COUNT(DISTINCT CASE
            WHEN e.event_type = 'report_sent'
             AND EXISTS (
                SELECT 1 FROM t_business_event submitted
                WHERE submitted.store_id = e.store_id
                  AND submitted.event_type = 'report_submitted'
                  AND submitted.aggregate_type = 'report'
                  AND submitted.aggregate_id = e.aggregate_id
                  AND submitted.occurred_at >= @pilot_from
                  AND submitted.occurred_at <= e.occurred_at
             )
            THEN e.aggregate_id
        END)
        / NULLIF(COUNT(DISTINCT CASE WHEN e.event_type = 'report_submitted' THEN e.aggregate_id END), 0),
        1
    ) AS report_sent_rate_pct,
    ROUND(
        100 * COUNT(DISTINCT CASE
            WHEN e.event_type IN ('report_opened', 'report_reopened')
             AND EXISTS (
                SELECT 1
                FROM t_business_event sent
                JOIN t_business_event submitted
                  ON submitted.store_id = sent.store_id
                 AND submitted.event_type = 'report_submitted'
                 AND submitted.aggregate_type = 'report'
                 AND submitted.aggregate_id = sent.aggregate_id
                 AND submitted.occurred_at >= @pilot_from
                 AND submitted.occurred_at <= sent.occurred_at
                WHERE sent.store_id = e.store_id
                  AND sent.event_type = 'report_sent'
                  AND sent.aggregate_type = 'report'
                  AND sent.aggregate_id = e.aggregate_id
                  AND sent.occurred_at >= @pilot_from
                  AND sent.occurred_at <= e.occurred_at
             )
            THEN e.aggregate_id
        END)
        / NULLIF(COUNT(DISTINCT CASE
            WHEN e.event_type = 'report_sent'
             AND EXISTS (
                SELECT 1 FROM t_business_event submitted
                WHERE submitted.store_id = e.store_id
                  AND submitted.event_type = 'report_submitted'
                  AND submitted.aggregate_type = 'report'
                  AND submitted.aggregate_id = e.aggregate_id
                  AND submitted.occurred_at >= @pilot_from
                  AND submitted.occurred_at <= e.occurred_at
             )
            THEN e.aggregate_id
        END), 0),
        1
    ) AS report_open_rate_pct,
    ROUND(
        100 * COUNT(DISTINCT CASE
            WHEN e.event_type = 'lead_submitted'
             AND lead_fact.report_id IS NOT NULL
             AND EXISTS (
                SELECT 1
                FROM t_business_event opened
                JOIN t_business_event sent
                  ON sent.store_id = opened.store_id
                 AND sent.event_type = 'report_sent'
                 AND sent.aggregate_type = 'report'
                 AND sent.aggregate_id = opened.aggregate_id
                 AND sent.occurred_at >= @pilot_from
                 AND sent.occurred_at <= opened.occurred_at
                JOIN t_business_event submitted
                  ON submitted.store_id = sent.store_id
                 AND submitted.event_type = 'report_submitted'
                 AND submitted.aggregate_type = 'report'
                 AND submitted.aggregate_id = sent.aggregate_id
                 AND submitted.occurred_at >= @pilot_from
                 AND submitted.occurred_at <= sent.occurred_at
                WHERE opened.store_id = e.store_id
                  AND opened.event_type IN ('report_opened', 'report_reopened')
                  AND opened.aggregate_type = 'report'
                  AND opened.aggregate_id = lead_fact.report_id
                  AND opened.occurred_at >= @pilot_from
                  AND opened.occurred_at <= e.occurred_at
             )
            THEN lead_fact.report_id
        END)
        / NULLIF(COUNT(DISTINCT CASE
            WHEN e.event_type IN ('report_opened', 'report_reopened')
             AND EXISTS (
                SELECT 1
                FROM t_business_event sent
                JOIN t_business_event submitted
                  ON submitted.store_id = sent.store_id
                 AND submitted.event_type = 'report_submitted'
                 AND submitted.aggregate_type = 'report'
                 AND submitted.aggregate_id = sent.aggregate_id
                 AND submitted.occurred_at >= @pilot_from
                 AND submitted.occurred_at <= sent.occurred_at
                WHERE sent.store_id = e.store_id
                  AND sent.event_type = 'report_sent'
                  AND sent.aggregate_type = 'report'
                  AND sent.aggregate_id = e.aggregate_id
                  AND sent.occurred_at >= @pilot_from
                  AND sent.occurred_at <= e.occurred_at
             )
            THEN e.aggregate_id
        END), 0),
        1
    ) AS opened_to_lead_rate_pct,
    ROUND(
        100 * COUNT(DISTINCT CASE
            WHEN e.event_type = 'rebook_created'
             AND rebook.source_lead_id IS NOT NULL
             AND EXISTS (
                SELECT 1 FROM t_business_event submitted_lead
                WHERE submitted_lead.store_id = e.store_id
                  AND submitted_lead.event_type = 'lead_submitted'
                  AND submitted_lead.aggregate_type = 'report_lead'
                  AND submitted_lead.aggregate_id = rebook.source_lead_id
                  AND submitted_lead.occurred_at >= @pilot_from
                  AND submitted_lead.occurred_at <= e.occurred_at
             )
            THEN rebook.source_lead_id
        END)
        / NULLIF(COUNT(DISTINCT CASE WHEN e.event_type = 'lead_submitted' THEN e.aggregate_id END), 0),
        1
    ) AS lead_to_rebook_rate_pct
FROM t_business_event e
LEFT JOIN t_report_lead lead_fact
  ON e.event_type = 'lead_submitted'
 AND lead_fact.id = e.aggregate_id
 AND lead_fact.store_id = e.store_id
LEFT JOIN t_follow_up_task rebook
  ON e.event_type = 'rebook_created'
 AND rebook.rebooked_appointment_id = e.aggregate_id
 AND rebook.store_id = e.store_id
WHERE e.occurred_at >= @pilot_from
  AND e.occurred_at < @pilot_to
  AND FIND_IN_SET(CAST(e.store_id AS CHAR), @pilot_store_ids) > 0
GROUP BY e.store_id
ORDER BY e.store_id;

-- 2. 报告发送时效：只纳入已经完整观察 24 小时的提交 cohort。
SELECT
    submitted.store_id,
    COUNT(*) AS matured_submitted_report_count,
    SUM(CASE
        WHEN sent.occurred_at IS NOT NULL
         AND sent.occurred_at >= submitted.occurred_at
         AND sent.occurred_at <= DATE_ADD(submitted.occurred_at, INTERVAL 24 HOUR)
        THEN 1 ELSE 0
    END) AS sent_within_24h_count,
    ROUND(
        100 * SUM(CASE
            WHEN sent.occurred_at IS NOT NULL
             AND sent.occurred_at >= submitted.occurred_at
             AND sent.occurred_at <= DATE_ADD(submitted.occurred_at, INTERVAL 24 HOUR)
            THEN 1 ELSE 0
        END) / NULLIF(COUNT(*), 0),
        1
    ) AS sent_within_24h_rate_pct
FROM t_business_event submitted
LEFT JOIN t_business_event sent
  ON sent.store_id = submitted.store_id
 AND sent.event_type = 'report_sent'
 AND sent.aggregate_type = 'report'
 AND sent.aggregate_id = submitted.aggregate_id
WHERE submitted.event_type = 'report_submitted'
  AND submitted.aggregate_type = 'report'
  AND submitted.occurred_at >= @pilot_from
  AND submitted.occurred_at < DATE_SUB(@pilot_to, INTERVAL 24 HOUR)
  AND FIND_IN_SET(CAST(submitted.store_id AS CHAR), @pilot_store_ids) > 0
GROUP BY submitted.store_id
ORDER BY submitted.store_id;

-- 3. 回访执行质量：以任务 due_date 落在窗口内为队列口径；退订不计员工完成。
SELECT
    t.store_id,
    COUNT(*) AS due_follow_up_count,
    SUM(CASE WHEN t.status = 'completed' THEN 1 ELSE 0 END) AS staff_completed_count,
    SUM(CASE WHEN t.status = 'canceled' AND t.outcome = 'unsubscribed' THEN 1 ELSE 0 END) AS unsubscribed_count,
    SUM(CASE WHEN t.status IN ('pending', 'in_progress') THEN 1 ELSE 0 END) AS still_open_count,
    SUM(CASE
        WHEN t.status IN ('pending', 'in_progress') AND t.due_date < CURRENT_DATE
        THEN 1 ELSE 0
    END) AS overdue_open_count,
    SUM(CASE
        WHEN t.status = 'completed'
         AND t.closed_at < DATE_ADD(t.due_date, INTERVAL 3 DAY)
        THEN 1 ELSE 0
    END) AS completed_by_due_plus_2d_count,
    ROUND(
        100 * SUM(CASE
            WHEN t.status = 'completed'
             AND t.closed_at < DATE_ADD(t.due_date, INTERVAL 3 DAY)
            THEN 1 ELSE 0
        END) / NULLIF(SUM(CASE WHEN t.status = 'completed' THEN 1 ELSE 0 END), 0),
        1
    ) AS on_time_completion_rate_pct,
    ROUND(AVG(t.attempt_count), 2) AS average_attempt_count,
    SUM(CASE WHEN t.outcome = 'rebooked' THEN 1 ELSE 0 END) AS rebooked_task_count,
    SUM(CASE WHEN t.outcome = 'not_interested' THEN 1 ELSE 0 END) AS not_interested_count,
    SUM(CASE WHEN t.outcome = 'invalid_contact' THEN 1 ELSE 0 END) AS invalid_contact_count
FROM t_follow_up_task t
WHERE t.due_date >= DATE(@pilot_from)
  AND t.due_date < DATE(@pilot_to)
  AND FIND_IN_SET(CAST(t.store_id AS CHAR), @pilot_store_ids) > 0
GROUP BY t.store_id
ORDER BY t.store_id;

-- 4. 约 200 次完整服务样本闸门：严格使用试点窗口，按店 + 最后一行试点店合计。
SELECT
    e.store_id,
    COUNT(DISTINCT e.aggregate_id) AS complete_service_count,
    MIN(e.occurred_at) AS first_complete_service_at,
    MAX(e.occurred_at) AS latest_complete_service_at
FROM t_business_event e
WHERE e.event_type = 'service_completed'
  AND e.aggregate_type = 'appointment'
  AND e.occurred_at >= @pilot_from
  AND e.occurred_at < @pilot_to
  AND FIND_IN_SET(CAST(e.store_id AS CHAR), @pilot_store_ids) > 0
GROUP BY e.store_id WITH ROLLUP;

-- 5. 指标可靠性反例：四项均须为 0；完整发布还必须通过 31 项 production preflight。
SELECT COUNT(*) AS completed_services_without_submitted_report_event_count
FROM t_business_event completed
LEFT JOIN t_report r
  ON r.appointment_id = completed.aggregate_id
 AND r.store_id = completed.store_id
 AND r.deleted = 0
LEFT JOIN t_business_event submitted
  ON submitted.store_id = completed.store_id
 AND submitted.event_type = 'report_submitted'
 AND submitted.aggregate_type = 'report'
 AND submitted.aggregate_id = r.id
WHERE completed.event_type = 'service_completed'
  AND completed.aggregate_type = 'appointment'
  AND (r.id IS NULL OR submitted.id IS NULL)
  AND completed.occurred_at >= @pilot_from
  AND completed.occurred_at < @pilot_to
  AND FIND_IN_SET(CAST(completed.store_id AS CHAR), @pilot_store_ids) > 0;

SELECT COUNT(*) AS invalid_rebook_attribution_count
FROM t_follow_up_task t
LEFT JOIN t_appointment a
  ON a.id = t.rebooked_appointment_id
 AND a.store_id = t.store_id
 AND a.deleted = 0
LEFT JOIN t_business_event e
  ON e.idempotency_key = CONCAT('rebook_created:', t.rebooked_appointment_id)
 AND e.event_type = 'rebook_created'
 AND e.store_id = t.store_id
 AND e.aggregate_type = 'appointment'
 AND e.aggregate_id = t.rebooked_appointment_id
WHERE t.outcome = 'rebooked'
  AND (a.id IS NULL OR e.id IS NULL)
  AND t.closed_at >= @pilot_from
  AND t.closed_at < @pilot_to
  AND FIND_IN_SET(CAST(t.store_id AS CHAR), @pilot_store_ids) > 0;

SELECT COUNT(*) AS duplicate_open_follow_up_task_count
FROM (
    SELECT t.store_id, t.source_lead_id
    FROM t_follow_up_task t
    WHERE t.status IN ('pending', 'in_progress')
      AND FIND_IN_SET(CAST(t.store_id AS CHAR), @pilot_store_ids) > 0
    GROUP BY t.store_id, t.source_lead_id
    HAVING COUNT(*) > 1
) duplicated;

SELECT SUM(anomaly_count) AS invalid_metric_event_order_count
FROM (
    SELECT COUNT(*) AS anomaly_count
    FROM t_business_event sent
    JOIN t_business_event submitted
      ON submitted.store_id = sent.store_id
     AND submitted.event_type = 'report_submitted'
     AND submitted.aggregate_type = 'report'
     AND submitted.aggregate_id = sent.aggregate_id
    WHERE sent.event_type = 'report_sent'
      AND sent.aggregate_type = 'report'
      AND sent.occurred_at < submitted.occurred_at
      AND sent.occurred_at >= @pilot_from
      AND sent.occurred_at < @pilot_to
      AND FIND_IN_SET(CAST(sent.store_id AS CHAR), @pilot_store_ids) > 0

    UNION ALL

    SELECT COUNT(*) AS anomaly_count
    FROM t_business_event rebooked
    JOIN t_follow_up_task task
      ON task.store_id = rebooked.store_id
     AND task.rebooked_appointment_id = rebooked.aggregate_id
    JOIN t_business_event submitted_lead
      ON submitted_lead.store_id = rebooked.store_id
     AND submitted_lead.event_type = 'lead_submitted'
     AND submitted_lead.aggregate_type = 'report_lead'
     AND submitted_lead.aggregate_id = task.source_lead_id
    WHERE rebooked.event_type = 'rebook_created'
      AND rebooked.aggregate_type = 'appointment'
      AND rebooked.occurred_at < submitted_lead.occurred_at
      AND rebooked.occurred_at >= @pilot_from
      AND rebooked.occurred_at < @pilot_to
      AND FIND_IN_SET(CAST(rebooked.store_id AS CHAR), @pilot_store_ids) > 0
) anomalies;