import logging
import os
from datetime import datetime, timedelta
from typing import Any

import aiomysql

from persistence import DB_CONFIG

logger = logging.getLogger(__name__)


def _env(name: str, fallback: str = "") -> str:
    value = str(os.environ.get(name, "")).strip()
    return value or fallback


OFFICIAL_MAIL_DB_CONFIG = {
    "host": _env(
        "OFFICIAL_MAIL_DB_HOST",
        _env("APP_MASTER_DB_HOST", str(DB_CONFIG["host"])),
    ),
    "port": int(
        _env(
            "OFFICIAL_MAIL_DB_PORT",
            _env("APP_MASTER_DB_PORT", str(DB_CONFIG["port"])),
        )
    ),
    "user": _env(
        "OFFICIAL_MAIL_DB_USER",
        _env("APP_MASTER_DB_USER", str(DB_CONFIG["user"])),
    ),
    "password": _env(
        "OFFICIAL_MAIL_DB_PASSWORD",
        _env("APP_MASTER_DB_PASSWORD", str(DB_CONFIG["password"])),
    ),
    "db": _env("OFFICIAL_MAIL_DB_NAME", "official_mail"),
    "charset": "utf8mb4",
    "autocommit": True,
}


class OfficialMailEventService:
    def __init__(self) -> None:
        self._pool: aiomysql.Pool | None = None
        self._billing_history_replayed = False

    @property
    def is_ready(self) -> bool:
        return self._pool is not None

    async def initialize(self) -> None:
        if self._pool:
            return

        self._pool = await aiomysql.create_pool(
            minsize=1,
            maxsize=2,
            connect_timeout=5,
            **OFFICIAL_MAIL_DB_CONFIG,
        )
        async with self._pool.acquire() as connection:
            async with connection.cursor() as cursor:
                await cursor.execute(
                    "SELECT 1 FROM domain_setup_assistance_requests LIMIT 1"
                )

        logger.info(
            "Official Mail DB connected: %s",
            OFFICIAL_MAIL_DB_CONFIG["db"],
        )

    async def close(self) -> None:
        if self._pool:
            self._pool.close()
            await self._pool.wait_closed()
            self._pool = None
            self._billing_history_replayed = False

    async def fetch_events(
        self,
        start_at: datetime,
        end_at: datetime,
    ) -> list[dict[str, Any]]:
        if not self._pool:
            return []

        # Replay recent billing events once after startup. Persistent event keys
        # suppress duplicate Telegram deliveries across service restarts.
        billing_start_at = (
            min(start_at, end_at - timedelta(days=7))
            if not self._billing_history_replayed
            else start_at
        )

        async with self._pool.acquire() as connection:
            async with connection.cursor(aiomysql.DictCursor) as cursor:
                await cursor.execute(
                    """
                    SELECT
                      r.id AS event_id,
                      'official-mail' AS app_key,
                      r.domain AS title,
                      r.provider AS detail,
                      u.email AS actor,
                      r.fee_amount AS amount,
                      'KRW' AS currency,
                      'domain-assistance' AS platform,
                      r.status,
                      r.notification_email,
                      r.requested_at AS event_time
                    FROM domain_setup_assistance_requests r
                    INNER JOIN users u ON u.id = r.user_id
                    WHERE r.requested_at > %s
                      AND r.requested_at <= %s
                    ORDER BY r.requested_at ASC, r.id ASC
                    """,
                    (start_at, end_at),
                )
                domain_rows = list(await cursor.fetchall())

                await cursor.execute(
                    """
                    SELECT
                      p.id AS event_id,
                      'official-mail' AS app_key,
                      '성장 플랜' AS title,
                      p.decline_message AS detail,
                      u.email AS actor,
                      p.amount / 100.0 AS amount,
                      UPPER(p.currency) AS currency,
                      'polar' AS platform,
                      p.status,
                      p.payment_method,
                      p.payment_trigger,
                      p.card_brand,
                      p.card_last_four,
                      p.decline_reason,
                      p.polar_payment_id,
                      p.polar_checkout_id,
                      p.payment_created_at AS occurred_at,
                      p.status_changed_at AS event_time
                    FROM mailbox_polar_payments p
                    INNER JOIN users u ON u.id = p.owner_user_id
                    WHERE p.status_changed_at > %s
                      AND p.status_changed_at <= %s
                      AND p.status IN ('succeeded', 'failed')
                    ORDER BY p.status_changed_at ASC, p.id ASC
                    """,
                    (billing_start_at, end_at),
                )
                polar_payment_rows = list(await cursor.fetchall())

                await cursor.execute(
                    """
                    SELECT
                      c.id AS event_id,
                      'official-mail' AS app_key,
                      c.product_desc AS title,
                      c.error_message AS detail,
                      u.email AS actor,
                      c.amount AS amount,
                      'KRW' AS currency,
                      'toss-pay' AS platform,
                      c.status,
                      c.pay_method AS payment_method,
                      c.charge_kind AS payment_trigger,
                      COALESCE(c.card_company_name, p.card_company_name)
                        AS card_brand,
                      COALESCE(c.card_num4_print, p.card_num4_print)
                        AS card_last_four,
                      c.error_code AS decline_reason,
                      c.order_no AS payment_id,
                      c.transaction_id,
                      c.requested_at AS occurred_at,
                      c.updated_at AS event_time,
                      CASE
                        WHEN c.status = 'success'
                          AND c.subscription_id IS NOT NULL
                          AND c.id = (
                            SELECT MIN(first_charge.id)
                            FROM mailbox_toss_pay_billing_charges first_charge
                            WHERE first_charge.subscription_id = c.subscription_id
                              AND first_charge.status = 'success'
                          )
                        THEN 1
                        ELSE 0
                      END AS is_initial_subscription_charge
                    FROM mailbox_toss_pay_billing_charges c
                    INNER JOIN users u ON u.id = c.owner_user_id
                    INNER JOIN mailbox_toss_pay_billing_profiles p
                      ON p.id = c.billing_profile_id
                    WHERE c.updated_at > %s
                      AND c.updated_at <= %s
                      AND c.status IN ('success', 'failed')
                    ORDER BY c.updated_at ASC, c.id ASC
                    """,
                    (billing_start_at, end_at),
                )
                toss_payment_rows = list(await cursor.fetchall())

                await cursor.execute(
                    """
                    SELECT
                      p.id AS event_id,
                      'official-mail' AS app_key,
                      '결제수단' AS title,
                      u.email AS actor,
                      'toss-pay' AS platform,
                      p.status,
                      p.pay_method AS payment_method,
                      p.card_company_name AS card_brand,
                      p.card_num4_print AS card_last_four,
                      p.account_bank_name,
                      SHA2(p.billing_key, 256) AS registration_key,
                      p.activated_at AS event_time
                    FROM mailbox_toss_pay_billing_profiles p
                    INNER JOIN users u ON u.id = p.owner_user_id
                    WHERE p.activated_at > %s
                      AND p.activated_at <= %s
                      AND p.status = 'active'
                      AND p.last_action = 'ACTIVATED'
                      AND p.billing_key IS NOT NULL
                    ORDER BY p.activated_at ASC, p.id ASC
                    """,
                    (billing_start_at, end_at),
                )
                toss_method_rows = list(await cursor.fetchall())

                await cursor.execute(
                    """
                    SELECT
                      s.id AS event_id,
                      'official-mail' AS app_key,
                      '성장 플랜' AS title,
                      u.email AS actor,
                      s.amount / 100.0 AS amount,
                      UPPER(s.currency) AS currency,
                      'polar' AS platform,
                      s.status,
                      'card' AS payment_method,
                      latest_payment.card_brand,
                      latest_payment.card_last_four,
                      s.polar_subscription_id,
                      s.checkout_id,
                      COALESCE(s.last_event_at, s.updated_at) AS event_time
                    FROM mailbox_polar_subscriptions s
                    INNER JOIN users u ON u.id = s.owner_user_id
                    LEFT JOIN mailbox_polar_payments latest_payment
                      ON latest_payment.id = (
                        SELECT payment.id
                        FROM mailbox_polar_payments payment
                        WHERE payment.owner_user_id = s.owner_user_id
                          AND payment.status = 'succeeded'
                          AND (
                            payment.polar_checkout_id = s.checkout_id
                            OR s.checkout_id IS NULL
                          )
                        ORDER BY payment.payment_created_at DESC, payment.id DESC
                        LIMIT 1
                      )
                    WHERE COALESCE(s.last_event_at, s.updated_at) > %s
                      AND COALESCE(s.last_event_at, s.updated_at) <= %s
                      AND s.status = 'active'
                    ORDER BY event_time ASC, s.id ASC
                    """,
                    (billing_start_at, end_at),
                )
                polar_method_rows = list(await cursor.fetchall())

        self._billing_history_replayed = True

        for row in domain_rows:
            row["source"] = "domain_setup_assistance_requests"
            row["event_type"] = "domain_assistance"
            row["event_key"] = (
                f"domain_setup_assistance_requests:{row.get('event_id')}:requested"
            )

        for row in polar_payment_rows:
            row["source"] = "mailbox_polar_payments"
            row["event_type"] = "polar_payment"
            row["event_key"] = (
                "mailbox_polar_payments:"
                f"{row.get('polar_payment_id')}:{row.get('status')}"
            )

        for row in toss_payment_rows:
            row["source"] = "mailbox_toss_pay_billing_charges"
            row["event_type"] = "toss_payment"
            row["event_key"] = (
                "mailbox_toss_pay_billing_charges:"
                f"{row.get('event_id')}:{row.get('status')}"
            )

        for row in toss_method_rows:
            row["source"] = "mailbox_toss_pay_billing_profiles"
            row["event_type"] = "payment_method_registered"
            row["region"] = "domestic"
            row["event_key"] = (
                "mailbox_toss_pay_billing_profiles:"
                f"{row.get('event_id')}:{row.get('registration_key')}"
            )

        for row in polar_method_rows:
            row["source"] = "mailbox_polar_subscriptions"
            row["event_type"] = "payment_method_registered"
            row["region"] = "overseas"
            row["event_key"] = (
                "mailbox_polar_subscriptions:"
                f"{row.get('polar_subscription_id')}:activated"
            )

        return sorted(
            [
                *domain_rows,
                *polar_payment_rows,
                *toss_payment_rows,
                *toss_method_rows,
                *polar_method_rows,
            ],
            key=lambda row: (row.get("event_time") or end_at, row["event_key"]),
        )


official_mail_service = OfficialMailEventService()