import sqlite3
import threading
from typing import Optional


SCHEMA = """
CREATE TABLE IF NOT EXISTS targets (
    target_id      TEXT PRIMARY KEY,
    target_type    TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
    name           TEXT,
    first_seen_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    reaction_count INTEGER NOT NULL DEFAULT 0
);

CREATE TABLE IF NOT EXISTS reactions_log (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
    message_id          TEXT NOT NULL,
    channel_id          TEXT NOT NULL,
    target_id           TEXT NOT NULL,
    author_id           TEXT NOT NULL,
    emoji               TEXT NOT NULL,
    reacted_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
    message_created_at  TEXT,
    source              TEXT NOT NULL CHECK(source IN ('backlog','live')),
    UNIQUE(message_id, emoji)
);
CREATE INDEX IF NOT EXISTS idx_reactions_log_target ON reactions_log(target_id);

CREATE TABLE IF NOT EXISTS channel_scan_state (
    channel_id         TEXT PRIMARY KEY,
    target_id          TEXT NOT NULL,
    newest_id_seen     TEXT,
    oldest_covered_at  TEXT,
    backlog_complete   INTEGER NOT NULL DEFAULT 0,
    last_scanned_at    TEXT
);

CREATE TABLE IF NOT EXISTS found_messages (
    message_id  TEXT PRIMARY KEY,
    channel_id  TEXT NOT NULL,
    target_id   TEXT NOT NULL,
    author_id   TEXT NOT NULL,
    found_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_found_messages_channel ON found_messages(channel_id);
"""


class Storage:
    """Thin synchronous sqlite3 wrapper. Callers from async code should
    dispatch through asyncio.to_thread() to avoid blocking the event loop."""

    def __init__(self, db_path: str):
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._lock = threading.Lock()
        with self._lock, self._conn:
            self._conn.execute("PRAGMA journal_mode=WAL")
            self._conn.executescript(SCHEMA)
            self._migrate()

    def _migrate(self) -> None:
        # Older DBs created before message_created_at existed.
        columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(reactions_log)")}
        if "message_created_at" not in columns:
            self._conn.execute("ALTER TABLE reactions_log ADD COLUMN message_created_at TEXT")

        # Older DBs created before oldest_covered_at existed (widening days_back
        # couldn't be detected/backfilled without it — see scan_channel_backlog).
        scan_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(channel_scan_state)")}
        if "oldest_covered_at" not in scan_columns:
            self._conn.execute("ALTER TABLE channel_scan_state ADD COLUMN oldest_covered_at TEXT")

        # Older DBs created before 'dm' targets (1:1 DMs) existed have a CHECK
        # constraint that only allows 'guild'/'group_dm' — rebuild the table
        # with the wider constraint, preserving all existing rows.
        targets_sql_row = self._conn.execute(
            "SELECT sql FROM sqlite_master WHERE type='table' AND name='targets'"
        ).fetchone()
        if targets_sql_row and "'dm'" not in targets_sql_row["sql"]:
            self._conn.executescript(
                """
                ALTER TABLE targets RENAME TO targets_old;
                CREATE TABLE targets (
                    target_id      TEXT PRIMARY KEY,
                    target_type    TEXT NOT NULL CHECK(target_type IN ('guild','group_dm','dm')),
                    name           TEXT,
                    first_seen_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                    reaction_count INTEGER NOT NULL DEFAULT 0
                );
                """
            )
            old_columns = {row["name"] for row in self._conn.execute("PRAGMA table_info(targets_old)")}
            common = [c for c in ("target_id", "target_type", "name", "first_seen_at", "reaction_count") if c in old_columns]
            col_list = ", ".join(common)
            self._conn.execute(f"INSERT INTO targets ({col_list}) SELECT {col_list} FROM targets_old")
            self._conn.execute("DROP TABLE targets_old")

    def upsert_target(self, target_id, target_type: str, name: str) -> None:
        with self._lock, self._conn:
            self._conn.execute(
                """
                INSERT INTO targets (target_id, target_type, name)
                VALUES (?, ?, ?)
                ON CONFLICT(target_id) DO UPDATE SET name = excluded.name
                """,
                (str(target_id), target_type, name),
            )

    def prune_targets(self, keep_target_ids: list) -> int:
        """Remove targets not present in the latest discovery run (e.g. a server
        that's no longer mutual, or a stale entry from an earlier bug/test run).
        reactions_log/channel_scan_state rows are left alone for history/undo."""
        keep = [str(t) for t in keep_target_ids]
        with self._lock, self._conn:
            placeholders = ",".join("?" * len(keep)) if keep else None
            if placeholders:
                cur = self._conn.execute(
                    f"DELETE FROM targets WHERE target_id NOT IN ({placeholders})", keep
                )
            else:
                cur = self._conn.execute("DELETE FROM targets")
            return cur.rowcount

    def record_message_found(self, message_id, channel_id, target_id, author_id) -> bool:
        """Log that a friend message was found, deduped by message_id (the
        PRIMARY KEY) so rescanning the same message ever after (e.g. after an
        interrupted run) is a no-op. This table is the sole source of truth
        for the "found" count — see get_stats() — precisely so that count can
        never drift out of sync the way a separately-maintained counter can."""
        with self._lock, self._conn:
            cur = self._conn.execute(
                """
                INSERT OR IGNORE INTO found_messages (message_id, channel_id, target_id, author_id)
                VALUES (?, ?, ?, ?)
                """,
                (str(message_id), str(channel_id), str(target_id), str(author_id)),
            )
            return cur.rowcount > 0

    def get_pending_found_message_ids(self, channel_id, emoji: str) -> list:
        """Messages already logged as found for this channel that don't have
        this emoji's reaction logged yet — lets a channel's reacting phase
        resume across a restart even if the resume checkpoint already moved
        past them."""
        with self._lock:
            cur = self._conn.execute(
                """
                SELECT f.message_id FROM found_messages f
                WHERE f.channel_id = ?
                  AND NOT EXISTS (
                      SELECT 1 FROM reactions_log r
                      WHERE r.message_id = f.message_id AND r.emoji = ?
                  )
                ORDER BY f.message_id
                """,
                (str(channel_id), emoji),
            )
            return [row["message_id"] for row in cur.fetchall()]

    def reset_message_counts(self) -> None:
        with self._lock, self._conn:
            self._conn.execute("DELETE FROM found_messages")

    def has_reacted(self, message_id, emoji: str) -> bool:
        with self._lock:
            cur = self._conn.execute(
                "SELECT 1 FROM reactions_log WHERE message_id = ? AND emoji = ? LIMIT 1",
                (str(message_id), emoji),
            )
            return cur.fetchone() is not None

    def record_reaction(
        self, message_id, channel_id, target_id, author_id, emoji: str, source: str, message_created_at=None
    ) -> bool:
        created_at = message_created_at.strftime("%Y-%m-%dT%H:%M:%fZ") if message_created_at else None
        with self._lock, self._conn:
            cur = self._conn.execute(
                """
                INSERT OR IGNORE INTO reactions_log
                    (message_id, channel_id, target_id, author_id, emoji, source, message_created_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (str(message_id), str(channel_id), str(target_id), str(author_id), emoji, source, created_at),
            )
            inserted = cur.rowcount > 0
            if inserted:
                self._conn.execute(
                    "UPDATE targets SET reaction_count = reaction_count + 1 WHERE target_id = ?",
                    (str(target_id),),
                )
            return inserted

    def clear_scan_state(self) -> int:
        """Wipe all per-channel resume points, forcing a full fresh backlog
        scan on the next run (widening days_back alone no longer requires
        this — see oldest_covered_at in scan_channel_backlog — but --undo
        still wants a clean slate since it also removes the reactions log)."""
        with self._lock, self._conn:
            cur = self._conn.execute("DELETE FROM channel_scan_state")
            return cur.rowcount

    def get_scan_state(self, channel_id) -> Optional[dict]:
        with self._lock:
            cur = self._conn.execute(
                "SELECT * FROM channel_scan_state WHERE channel_id = ?",
                (str(channel_id),),
            )
            row = cur.fetchone()
            return dict(row) if row else None

    def set_scan_state(self, channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete: bool) -> None:
        with self._lock, self._conn:
            self._conn.execute(
                """
                INSERT INTO channel_scan_state
                    (channel_id, target_id, newest_id_seen, oldest_covered_at, backlog_complete, last_scanned_at)
                VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
                ON CONFLICT(channel_id) DO UPDATE SET
                    newest_id_seen = excluded.newest_id_seen,
                    oldest_covered_at = excluded.oldest_covered_at,
                    backlog_complete = excluded.backlog_complete,
                    last_scanned_at = excluded.last_scanned_at
                """,
                (
                    str(channel_id), str(target_id),
                    str(newest_id_seen) if newest_id_seen else None,
                    oldest_covered_at.isoformat() if oldest_covered_at else None,
                    int(backlog_complete),
                ),
            )

    def get_all_reactions(self) -> list:
        with self._lock:
            cur = self._conn.execute(
                "SELECT id, message_id, channel_id, target_id, emoji FROM reactions_log ORDER BY id"
            )
            return [dict(row) for row in cur.fetchall()]

    def delete_reaction(self, row_id: int) -> None:
        with self._lock, self._conn:
            row = self._conn.execute(
                "SELECT target_id FROM reactions_log WHERE id = ?", (row_id,)
            ).fetchone()
            self._conn.execute("DELETE FROM reactions_log WHERE id = ?", (row_id,))
            if row:
                self._conn.execute(
                    "UPDATE targets SET reaction_count = MAX(reaction_count - 1, 0) WHERE target_id = ?",
                    (row["target_id"],),
                )

    def get_stats(self) -> list:
        with self._lock:
            cur = self._conn.execute(
                """
                SELECT t.target_id, t.target_type, t.name, t.reaction_count,
                       COALESCE(fm.found_count, 0) AS messages_found,
                       MAX(r.message_created_at) AS last_reaction_at
                FROM targets t
                LEFT JOIN reactions_log r ON r.target_id = t.target_id
                LEFT JOIN (
                    SELECT target_id, COUNT(*) AS found_count
                    FROM found_messages
                    GROUP BY target_id
                ) fm ON fm.target_id = t.target_id
                GROUP BY t.target_id
                ORDER BY t.reaction_count DESC
                """
            )
            return [dict(row) for row in cur.fetchall()]

    def close(self) -> None:
        with self._lock:
            self._conn.close()