import json
import sqlite3
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from pathlib import Path

from src.private_state import connect_private_sqlite


@dataclass(frozen=True)
class PushDelivery:
    session_id: str
    subscription: dict
    thread_revisions: tuple[tuple[int, str], ...]
    digest_revisions: tuple[tuple[int, str], ...] = ()

    @property
    def thread_ids(self) -> tuple[int, ...]:
        return tuple(thread_id for thread_id, _revision in self.thread_revisions)

    @property
    def digest_ids(self) -> tuple[int, ...]:
        return tuple(thread_id for thread_id, _revision in self.digest_revisions)


@dataclass(frozen=True)
class DeadlineReminderDevice:
    session_id: str
    subscription: dict
    timezone: str
    reminder_hour: int
    reminder_days: int
    delivered_local_day: str | None


def _revisions(
    values: Mapping[int, str] | Iterable[int | tuple[int, str]],
) -> tuple[tuple[int, str], ...]:
    if isinstance(values, Mapping):
        items = list(values.items())
    else:
        items = []
        for value in values:
            if isinstance(value, tuple):
                items.append(value)
            else:
                items.append((value, ""))
    normalized = {}
    for thread_id, revision in items:
        thread_id = int(thread_id)
        if thread_id > 0:
            normalized[thread_id] = str(revision or "")
    return tuple(sorted(normalized.items()))


class PushSubscriptionStore:
    """Durable, device-bound Web Push subscriptions and delivery deduplication."""

    def __init__(self, path: str | Path):
        self.path = Path(path)
        with self._connect() as connection:
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS push_subscriptions (
                    session_id TEXT PRIMARY KEY,
                    endpoint TEXT NOT NULL UNIQUE,
                    subscription_json TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS push_deliveries (
                    session_id TEXT NOT NULL,
                    thread_id INTEGER NOT NULL,
                    revision TEXT NOT NULL DEFAULT '',
                    PRIMARY KEY (session_id, thread_id),
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                CREATE TABLE IF NOT EXISTS push_dispatch_lease (
                    channel TEXT PRIMARY KEY,
                    owner TEXT NOT NULL,
                    expires_at REAL NOT NULL
                );
                CREATE TABLE IF NOT EXISTS push_digest_pending (
                    session_id TEXT NOT NULL,
                    thread_id INTEGER NOT NULL,
                    revision TEXT NOT NULL DEFAULT '',
                    PRIMARY KEY (session_id, thread_id),
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                CREATE TABLE IF NOT EXISTS push_deadline_preferences (
                    session_id TEXT PRIMARY KEY,
                    enabled INTEGER NOT NULL DEFAULT 0,
                    timezone TEXT NOT NULL DEFAULT 'UTC',
                    reminder_hour INTEGER NOT NULL DEFAULT 9,
                    reminder_days INTEGER NOT NULL DEFAULT 2,
                    delivered_local_day TEXT,
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                """
            )
            delivery_columns = {
                row[1] for row in connection.execute("PRAGMA table_info(push_deliveries)")
            }
            if "revision" not in delivery_columns:
                connection.execute(
                    "ALTER TABLE push_deliveries ADD COLUMN revision TEXT NOT NULL DEFAULT '*'"
                )
            digest_columns = {
                row[1]
                for row in connection.execute("PRAGMA table_info(push_digest_pending)")
            }
            if "revision" not in digest_columns:
                connection.execute(
                    "ALTER TABLE push_digest_pending ADD COLUMN revision TEXT NOT NULL DEFAULT '*'"
                )
            lease_columns = {
                row[1] for row in connection.execute("PRAGMA table_info(push_dispatch_lease)")
            }
            if "singleton" in lease_columns:
                connection.executescript(
                    """
                    ALTER TABLE push_dispatch_lease RENAME TO push_dispatch_lease_legacy;
                    CREATE TABLE push_dispatch_lease (
                        channel TEXT PRIMARY KEY,
                        owner TEXT NOT NULL,
                        expires_at REAL NOT NULL
                    );
                    INSERT INTO push_dispatch_lease(channel, owner, expires_at)
                        SELECT 'unread', owner, expires_at
                        FROM push_dispatch_lease_legacy WHERE singleton = 1;
                    DROP TABLE push_dispatch_lease_legacy;
                    """
                )
            preference_columns = {
                row[1]
                for row in connection.execute("PRAGMA table_info(push_deadline_preferences)")
            }
            if "reminder_days" not in preference_columns:
                connection.execute(
                    "ALTER TABLE push_deadline_preferences ADD COLUMN reminder_days INTEGER NOT NULL DEFAULT 2"
                )
    def _connect(self):
        connection = connect_private_sqlite(self.path, timeout=2)
        connection.execute("PRAGMA foreign_keys = ON")
        return connection

    def acquire_dispatch_lease(
        self, owner: str, *, channel: str = "unread", now: float, lease_seconds: float
    ) -> bool:
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            current = connection.execute(
                "SELECT owner, expires_at FROM push_dispatch_lease WHERE channel = ?",
                (channel,),
            ).fetchone()
            if current is not None and current[0] != owner and current[1] > now:
                return False
            connection.execute(
                """INSERT INTO push_dispatch_lease(channel, owner, expires_at)
                   VALUES (?, ?, ?)
                   ON CONFLICT(channel) DO UPDATE SET
                       owner = excluded.owner, expires_at = excluded.expires_at""",
                (channel, owner, now + lease_seconds),
            )
            return True

    def release_dispatch_lease(self, owner: str, *, channel: str = "unread") -> bool:
        with self._connect() as connection:
            result = connection.execute(
                "DELETE FROM push_dispatch_lease WHERE channel = ? AND owner = ?",
                (channel, owner),
            )
            return result.rowcount == 1

    def upsert(self, session_id: str, subscription: dict) -> None:
        endpoint = subscription["endpoint"]
        encoded = json.dumps(subscription, separators=(",", ":"), sort_keys=True)
        with self._connect() as connection:
            connection.execute("DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint,))
            connection.execute("DELETE FROM push_subscriptions WHERE session_id = ?", (session_id,))
            connection.execute(
                "INSERT INTO push_subscriptions(session_id, endpoint, subscription_json) VALUES (?, ?, ?)",
                (session_id, endpoint, encoded),
            )

    def delete_session(self, session_id: str) -> None:
        with self._connect() as connection:
            connection.execute("DELETE FROM push_subscriptions WHERE session_id = ?", (session_id,))

    def delete_all(self) -> None:
        with self._connect() as connection:
            connection.execute("DELETE FROM push_subscriptions")

    def is_subscribed(self, session_id: str) -> bool:
        with self._connect() as connection:
            return connection.execute(
                "SELECT 1 FROM push_subscriptions WHERE session_id = ?", (session_id,)
            ).fetchone() is not None

    def set_deadline_preferences(
        self, session_id: str, *, enabled: bool, timezone: str, reminder_hour: int,
        reminder_days: int = 2,
    ) -> None:
        with self._connect() as connection:
            connection.execute(
                """INSERT INTO push_deadline_preferences(
                       session_id, enabled, timezone, reminder_hour, reminder_days
                   ) VALUES (?, ?, ?, ?, ?)
                   ON CONFLICT(session_id) DO UPDATE SET
                       enabled = excluded.enabled,
                       timezone = excluded.timezone,
                       reminder_hour = excluded.reminder_hour,
                       reminder_days = excluded.reminder_days""",
                (session_id, int(enabled), timezone, reminder_hour, reminder_days),
            )

    def deadline_preferences(self, session_id: str) -> dict:
        with self._connect() as connection:
            row = connection.execute(
                """SELECT enabled, timezone, reminder_hour, reminder_days
                   FROM push_deadline_preferences WHERE session_id = ?""",
                (session_id,),
            ).fetchone()
        return {
            "enabled": bool(row[0]) if row else False,
            "timezone": row[1] if row else "UTC",
            "reminder_hour": row[2] if row else 9,
            "reminder_days": row[3] if row else 2,
        }

    def deadline_reminder_devices(self) -> list[DeadlineReminderDevice]:
        with self._connect() as connection:
            rows = connection.execute(
                """SELECT s.session_id, s.subscription_json, p.timezone,
                          p.reminder_hour, p.reminder_days, p.delivered_local_day
                   FROM push_subscriptions s
                   JOIN push_deadline_preferences p ON p.session_id = s.session_id
                   WHERE p.enabled = 1 ORDER BY s.session_id"""
            ).fetchall()
        return [
            DeadlineReminderDevice(row[0], json.loads(row[1]), row[2], row[3], row[4], row[5])
            for row in rows
        ]

    def mark_deadline_reminder_delivered(self, session_id: str, local_day: str) -> None:
        with self._connect() as connection:
            connection.execute(
                """UPDATE push_deadline_preferences SET delivered_local_day = ?
                   WHERE session_id = ? AND enabled = 1""",
                (local_day, session_id),
            )

    def claim_unseen(
        self, thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]]
    ) -> list[PushDelivery]:
        candidates = _revisions(thread_revisions)
        if not candidates:
            return []
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT session_id, subscription_json FROM push_subscriptions ORDER BY session_id"
            ).fetchall()
            deliveries = []
            for session_id, encoded in rows:
                delivered = {
                    row[0]: row[1]
                    for row in connection.execute(
                        "SELECT thread_id, revision FROM push_deliveries WHERE session_id = ?",
                        (session_id,),
                    )
                }
                for thread_id, revision in candidates:
                    if delivered.get(thread_id) == "*":
                        connection.execute(
                            """UPDATE push_deliveries SET revision = ?
                               WHERE session_id = ? AND thread_id = ?""",
                            (revision, session_id, thread_id),
                        )
                        delivered[thread_id] = revision
                unseen = tuple(
                    candidate
                    for candidate in candidates
                    if (
                        candidate[0] not in delivered
                        or (
                            bool(candidate[1])
                            and (
                                not delivered[candidate[0]]
                                or candidate[1] > delivered[candidate[0]]
                            )
                        )
                    )
                )
                if unseen:
                    pending = {
                        row[0]: row[1]
                        for row in connection.execute(
                            "SELECT thread_id, revision FROM push_digest_pending WHERE session_id = ?",
                            (session_id,),
                        )
                    }
                    digest_revisions = []
                    for thread_id, revision in unseen:
                        if pending.get(thread_id) in {revision, "*"}:
                            digest_revisions.append((thread_id, revision))
                            if pending[thread_id] == "*":
                                connection.execute(
                                    """UPDATE push_digest_pending SET revision = ?
                                       WHERE session_id = ? AND thread_id = ?""",
                                    (revision, session_id, thread_id),
                                )
                    deliveries.append(
                        PushDelivery(
                            session_id,
                            json.loads(encoded),
                            unseen,
                            tuple(digest_revisions),
                        )
                    )
            return deliveries

    def reconcile_unread(self, thread_ids: Iterable[int]) -> None:
        """Prune per-device checkpoints that are absent from a complete snapshot."""
        unread_ids = tuple(sorted({int(value) for value in thread_ids if int(value) > 0}))
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            connection.execute(
                "CREATE TEMP TABLE current_unread(thread_id INTEGER PRIMARY KEY)"
            )
            connection.executemany(
                "INSERT INTO current_unread(thread_id) VALUES (?)",
                ((thread_id,) for thread_id in unread_ids),
            )
            connection.execute(
                """DELETE FROM push_deliveries
                   WHERE NOT EXISTS (
                       SELECT 1 FROM current_unread
                       WHERE current_unread.thread_id = push_deliveries.thread_id
                   )"""
            )
            connection.execute(
                """DELETE FROM push_digest_pending
                   WHERE NOT EXISTS (
                       SELECT 1 FROM current_unread
                       WHERE current_unread.thread_id = push_digest_pending.thread_id
                   )"""
            )

    def mark_digest_pending(
        self,
        session_id: str,
        thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
    ) -> None:
        values = _revisions(thread_revisions)
        with self._connect() as connection:
            connection.executemany(
                """INSERT INTO push_digest_pending(session_id, thread_id, revision)
                   VALUES (?, ?, ?)
                   ON CONFLICT(session_id, thread_id) DO UPDATE SET
                       revision = excluded.revision""",
                ((session_id, thread_id, revision) for thread_id, revision in values),
            )

    def mark_delivered(
        self,
        session_id: str,
        thread_revisions: Mapping[int, str] | Iterable[int | tuple[int, str]],
    ) -> None:
        values = _revisions(thread_revisions)
        with self._connect() as connection:
            connection.executemany(
                """INSERT INTO push_deliveries(session_id, thread_id, revision)
                   VALUES (?, ?, ?)
                   ON CONFLICT(session_id, thread_id) DO UPDATE SET
                       revision = excluded.revision""",
                ((session_id, thread_id, revision) for thread_id, revision in values),
            )
            connection.executemany(
                "DELETE FROM push_digest_pending WHERE session_id = ? AND thread_id = ?",
                ((session_id, thread_id) for thread_id, _revision in values),
            )