from __future__ import annotations

import hashlib
import hmac
import os
import sqlite3
import time
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from pathlib import Path

from src.private_state import connect_private_sqlite
from src.state_encryption import (
    PrivateStateCipher,
    decode_private_state_encryption_key,
)


@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
    snoozed_until: float | None


@dataclass(frozen=True)
class StartDayReminderDevice:
    session_id: str
    subscription: dict
    timezone: str
    reminder_hour: int
    delivered_plan_date: str | None


@dataclass(frozen=True)
class FollowingNotificationDevice:
    session_id: str
    subscription: dict
    delivered_fingerprint: str | None


class DisabledPushSubscriptionStore:
    """No-persistence store used when Web Push is not configured."""

    def is_subscribed(self, session_id: str) -> bool:
        return False

    def subscription_for_session(self, session_id: str) -> dict | None:
        return None

    def deadline_preferences(
        self, session_id: str, *, now: float | None = None
    ) -> dict:
        return {
            "enabled": False,
            "timezone": "UTC",
            "reminder_hour": 9,
            "reminder_days": 2,
            "snoozed_until": None,
        }

    def deadline_reminder_devices(self) -> list[DeadlineReminderDevice]:
        return []

    def start_day_preferences(self, session_id: str) -> dict:
        return {"enabled": False, "timezone": "UTC", "reminder_hour": 9}

    def start_day_reminder_devices(self) -> list[StartDayReminderDevice]:
        return []

    def following_preferences(self, session_id: str) -> dict:
        return {"enabled": False}

    def following_notification_devices(self) -> list[FollowingNotificationDevice]:
        return []

    def claim_unseen(self, thread_revisions) -> list[PushDelivery]:
        return []

    def acquire_dispatch_lease(self, *args, **kwargs) -> bool:
        return False

    def release_dispatch_lease(self, *args, **kwargs) -> bool:
        return False

    def snooze_deadline_reminder(self, *args, **kwargs) -> bool:
        return False

    def upsert(self, *args, **kwargs) -> None:
        raise RuntimeError("push notifications are not configured")

    def delete_session(self, *args, **kwargs) -> None:
        return None

    def delete_all(self, *args, **kwargs) -> None:
        return None

    def set_deadline_preferences(self, *args, **kwargs) -> None:
        return None

    def clear_deadline_snooze(self, *args, **kwargs) -> None:
        return None

    def mark_deadline_reminder_delivered(self, *args, **kwargs) -> None:
        return None

    def set_start_day_preferences(self, *args, **kwargs) -> None:
        return None

    def mark_start_day_reminder_delivered(self, *args, **kwargs) -> None:
        return None

    def set_following_preferences(self, *args, **kwargs) -> None:
        return None

    def mark_following_delivered(self, *args, **kwargs) -> None:
        return None

    def reconcile_unread(self, *args, **kwargs) -> None:
        return None

    def mark_digest_pending(self, *args, **kwargs) -> None:
        return None

    def mark_delivered(self, *args, **kwargs) -> None:
        return None

    def delivery_health(self, *args, **kwargs) -> dict:
        return {}

    def mark_delivery_failed(self, *args, **kwargs) -> None:
        return None

    def mark_delivery_succeeded(self, *args, **kwargs) -> None:
        return None


def build_push_subscription_store(
    path: str | Path, *, push_enabled: bool
) -> PushSubscriptionStore | DisabledPushSubscriptionStore:
    if not push_enabled:
        return DisabledPushSubscriptionStore()
    return PushSubscriptionStore(path)


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, *, encryption_key: bytes | None = None):
        self.path = Path(path)
        key = encryption_key
        if key is None:
            key = decode_private_state_encryption_key(
                os.getenv("STACKCHAIN_PUSH_STATE_ENCRYPTION_KEY", "")
            )
        self._encryption_key = key
        self._cipher = PrivateStateCipher(key, store="push-subscriptions")
        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,
                    snoozed_until REAL,
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                CREATE TABLE IF NOT EXISTS push_delivery_health (
                    session_id TEXT NOT NULL,
                    channel TEXT NOT NULL,
                    consecutive_failures INTEGER NOT NULL DEFAULT 0,
                    last_attempted_at REAL NOT NULL,
                    last_succeeded_at REAL,
                    reason TEXT,
                    PRIMARY KEY (session_id, channel),
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                CREATE TABLE IF NOT EXISTS push_start_day_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,
                    delivered_plan_date TEXT,
                    FOREIGN KEY (session_id) REFERENCES push_subscriptions(session_id)
                        ON DELETE CASCADE
                );
                CREATE TABLE IF NOT EXISTS push_following_preferences (
                    session_id TEXT PRIMARY KEY,
                    enabled INTEGER NOT NULL DEFAULT 0,
                    delivered_fingerprint 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"
                )
            if "snoozed_until" not in preference_columns:
                connection.execute(
                    "ALTER TABLE push_deadline_preferences ADD COLUMN snoozed_until REAL"
                )
            rows = connection.execute(
                "SELECT session_id, endpoint, subscription_json FROM push_subscriptions"
            ).fetchall()
            for session_id, endpoint, payload in rows:
                subscription, plaintext = self._cipher.open(payload, binding=session_id)
                if not isinstance(subscription, dict) or not subscription.get("endpoint"):
                    raise ValueError("push subscription payload is invalid")
                endpoint_index = self._endpoint_index(subscription["endpoint"])
                if plaintext or endpoint != endpoint_index:
                    connection.execute(
                        "UPDATE push_subscriptions SET endpoint = ?, subscription_json = ? WHERE session_id = ?",
                        (
                            endpoint_index,
                            self._cipher.seal(subscription, binding=session_id),
                            session_id,
                        ),
                    )

    def _endpoint_index(self, endpoint: str) -> str:
        return hmac.new(
            self._encryption_key,
            b"stackchain:push-endpoint:v1\0" + endpoint.encode(),
            hashlib.sha256,
        ).hexdigest()

    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"]
        endpoint_index = self._endpoint_index(endpoint)
        encoded = self._cipher.seal(subscription, binding=session_id)
        with self._connect() as connection:
            connection.execute(
                "DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint_index,)
            )
            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_index, 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 subscription_for_session(self, session_id: str) -> dict | None:
        with self._connect() as connection:
            row = connection.execute(
                "SELECT subscription_json FROM push_subscriptions WHERE session_id = ?",
                (session_id,),
            ).fetchone()
        return self._open_subscription(session_id, row[0]) if row else None

    def delivery_health(self, session_id: str) -> dict:
        with self._connect() as connection:
            rows = connection.execute(
                """SELECT channel, consecutive_failures, last_attempted_at,
                          last_succeeded_at, reason
                   FROM push_delivery_health WHERE session_id = ? ORDER BY channel""",
                (session_id,),
            ).fetchall()
        return {
            row[0]: {
                "state": "degraded" if row[1] else "healthy",
                "consecutive_failures": row[1],
                "last_attempted_at": row[2],
                "last_succeeded_at": row[3],
                "reason": row[4],
            }
            for row in rows
        }

    def mark_delivery_failed(
        self, session_id: str, channel: str, reason: str, *, now: float | None = None
    ) -> None:
        attempted_at = time.time() if now is None else now
        with self._connect() as connection:
            connection.execute(
                """INSERT INTO push_delivery_health(
                       session_id, channel, consecutive_failures, last_attempted_at, reason
                   ) VALUES (?, ?, 1, ?, ?)
                   ON CONFLICT(session_id, channel) DO UPDATE SET
                       consecutive_failures = consecutive_failures + 1,
                       last_attempted_at = excluded.last_attempted_at,
                       reason = excluded.reason""",
                (session_id, channel, attempted_at, reason),
            )

    def mark_delivery_succeeded(
        self, session_id: str, channel: str, *, now: float | None = None
    ) -> None:
        attempted_at = time.time() if now is None else now
        with self._connect() as connection:
            connection.execute(
                """INSERT INTO push_delivery_health(
                       session_id, channel, consecutive_failures, last_attempted_at,
                       last_succeeded_at, reason
                   ) VALUES (?, ?, 0, ?, ?, NULL)
                   ON CONFLICT(session_id, channel) DO UPDATE SET
                       consecutive_failures = 0,
                       last_attempted_at = excluded.last_attempted_at,
                       last_succeeded_at = excluded.last_succeeded_at,
                       reason = NULL""",
                (session_id, channel, attempted_at, attempted_at),
            )

    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, *, now: float | None = None) -> dict:
        with self._connect() as connection:
            row = connection.execute(
                """SELECT enabled, timezone, reminder_hour, reminder_days, snoozed_until
                   FROM push_deadline_preferences WHERE session_id = ?""",
                (session_id,),
            ).fetchone()
        current_time = time.time() if now is None else now
        snoozed_until = row[4] if row and row[4] and row[4] > current_time else None
        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,
            "snoozed_until": snoozed_until,
        }

    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,
                          p.snoozed_until
                   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],
                self._open_subscription(row[0], row[1]),
                row[2],
                row[3],
                row[4],
                row[5],
                row[6],
            )
            for row in rows
        ]

    def snooze_deadline_reminder(
        self, session_id: str, *, now: float, delay_seconds: int = 3_600
    ) -> bool:
        with self._connect() as connection:
            result = connection.execute(
                """UPDATE push_deadline_preferences SET snoozed_until = ?
                   WHERE session_id = ? AND enabled = 1""",
                (now + delay_seconds, session_id),
            )
            return result.rowcount == 1

    def clear_deadline_snooze(self, session_id: str) -> None:
        with self._connect() as connection:
            connection.execute(
                """UPDATE push_deadline_preferences SET snoozed_until = NULL
                   WHERE session_id = ?""",
                (session_id,),
            )

    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 = ?, snoozed_until = NULL
                   WHERE session_id = ? AND enabled = 1""",
                (local_day, session_id),
            )

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

    def start_day_preferences(self, session_id: str) -> dict:
        with self._connect() as connection:
            row = connection.execute(
                """SELECT enabled, timezone, reminder_hour
                   FROM push_start_day_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,
        }

    def start_day_reminder_devices(self) -> list[StartDayReminderDevice]:
        with self._connect() as connection:
            rows = connection.execute(
                """SELECT s.session_id, s.subscription_json, p.timezone,
                          p.reminder_hour, p.delivered_plan_date
                   FROM push_subscriptions s
                   JOIN push_start_day_preferences p ON p.session_id = s.session_id
                   WHERE p.enabled = 1 ORDER BY s.session_id"""
            ).fetchall()
        return [
            StartDayReminderDevice(
                row[0], self._open_subscription(row[0], row[1]), row[2], row[3], row[4]
            )
            for row in rows
        ]

    def mark_start_day_reminder_delivered(
        self, session_id: str, plan_date: str
    ) -> None:
        with self._connect() as connection:
            connection.execute(
                """UPDATE push_start_day_preferences SET delivered_plan_date = ?
                   WHERE session_id = ? AND enabled = 1""",
                (plan_date, session_id),
            )

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

    def following_preferences(self, session_id: str) -> dict:
        with self._connect() as connection:
            row = connection.execute(
                "SELECT enabled FROM push_following_preferences WHERE session_id = ?",
                (session_id,),
            ).fetchone()
        return {"enabled": bool(row[0]) if row else False}

    def following_notification_devices(self) -> list[FollowingNotificationDevice]:
        with self._connect() as connection:
            rows = connection.execute(
                """SELECT s.session_id, s.subscription_json, p.delivered_fingerprint
                   FROM push_subscriptions s
                   JOIN push_following_preferences p ON p.session_id = s.session_id
                   WHERE p.enabled = 1 ORDER BY s.session_id"""
            ).fetchall()
        return [
            FollowingNotificationDevice(
                row[0], self._open_subscription(row[0], row[1]), row[2]
            )
            for row in rows
        ]

    def mark_following_delivered(self, session_id: str, fingerprint: str) -> None:
        with self._connect() as connection:
            connection.execute(
                """UPDATE push_following_preferences SET delivered_fingerprint = ?
                   WHERE session_id = ? AND enabled = 1""",
                (fingerprint, 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,
                            self._open_subscription(session_id, encoded),
                            unseen,
                            tuple(digest_revisions),
                        )
                    )
            return deliveries

    def _open_subscription(self, session_id: str, payload: str) -> dict:
        subscription, _plaintext = self._cipher.open(payload, binding=session_id)
        if not isinstance(subscription, dict):
            raise ValueError("push subscription payload is invalid")
        return subscription

    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),
            )