store.py

"""SQLite storage. Idempotent inserts via UPSERT. Single-writer model."""
from __future__ import annotations

import json
import logging
import sqlite3
import time
from pathlib import Path
from typing import Optional

log = logging.getLogger(__name__)

SCHEMA = """
CREATE TABLE IF NOT EXISTS posts (
    id            TEXT PRIMARY KEY,
    subreddit     TEXT NOT NULL,
    author        TEXT,
    title         TEXT,
    selftext      TEXT,
    score         INTEGER,
    num_comments  INTEGER,
    created_utc   INTEGER NOT NULL,
    permalink     TEXT,
    url           TEXT,
    is_self       INTEGER,
    over_18       INTEGER,
    fetched_at    INTEGER NOT NULL,
    source        TEXT NOT NULL,
    media_json    TEXT
);
CREATE INDEX IF NOT EXISTS idx_posts_subreddit ON posts(subreddit);
CREATE INDEX IF NOT EXISTS idx_posts_created   ON posts(created_utc);
CREATE INDEX IF NOT EXISTS idx_posts_author    ON posts(author);

CREATE TABLE IF NOT EXISTS comments (
    id            TEXT PRIMARY KEY,
    post_id       TEXT NOT NULL,
    parent_id     TEXT,
    subreddit     TEXT NOT NULL,
    author        TEXT,
    body          TEXT,
    score         INTEGER,
    created_utc   INTEGER NOT NULL,
    permalink     TEXT,
    fetched_at    INTEGER NOT NULL,
    source        TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_comments_post      ON comments(post_id);
CREATE INDEX IF NOT EXISTS idx_comments_subreddit ON comments(subreddit);
CREATE INDEX IF NOT EXISTS idx_comments_created   ON comments(created_utc);
CREATE INDEX IF NOT EXISTS idx_comments_author    ON comments(author);

CREATE TABLE IF NOT EXISTS pull_runs (
    id               INTEGER PRIMARY KEY AUTOINCREMENT,
    kind             TEXT NOT NULL,
    target           TEXT NOT NULL,
    started_at       INTEGER NOT NULL,
    finished_at      INTEGER,
    posts_added      INTEGER DEFAULT 0,
    comments_added   INTEGER DEFAULT 0,
    error            TEXT,
    params           TEXT
);

CREATE TABLE IF NOT EXISTS subreddits (
    name             TEXT PRIMARY KEY,
    subscribers      INTEGER,
    description      TEXT,
    lang             TEXT,
    over_18          INTEGER,
    discovered_via   TEXT,
    discovered_at    INTEGER,
    selected         INTEGER DEFAULT 0
);

CREATE TABLE IF NOT EXISTS blind_topics (
    slug             TEXT PRIMARY KEY,
    name             TEXT NOT NULL,
    description      TEXT,
    post_count       INTEGER DEFAULT 0,
    comment_count    INTEGER DEFAULT 0,
    last_scraped_at  INTEGER
);

CREATE TABLE IF NOT EXISTS blind_posts (
    id               TEXT PRIMARY KEY,
    topic_slug       TEXT NOT NULL,
    channel          TEXT,
    title            TEXT,
    body             TEXT,
    company          TEXT,
    score            INTEGER,
    comment_count    INTEGER,
    created_at       INTEGER,
    url              TEXT,
    fetched_at       INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_blind_posts_topic   ON blind_posts(topic_slug);
CREATE INDEX IF NOT EXISTS idx_blind_posts_created ON blind_posts(created_at);

CREATE TABLE IF NOT EXISTS blind_comments (
    id               TEXT PRIMARY KEY,
    post_id          TEXT NOT NULL,
    topic_slug       TEXT NOT NULL,
    parent_id        TEXT,
    company          TEXT,
    body             TEXT,
    score            INTEGER,
    created_at       INTEGER,
    fetched_at       INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_blind_comments_post   ON blind_comments(post_id);
CREATE INDEX IF NOT EXISTS idx_blind_comments_topic  ON blind_comments(topic_slug);

CREATE TABLE IF NOT EXISTS cleaner_run_flagged (
    run_id    INTEGER NOT NULL,
    post_id   TEXT NOT NULL,
    subreddit TEXT,
    title     TEXT,
    body      TEXT,
    PRIMARY KEY (run_id, post_id)
);
CREATE INDEX IF NOT EXISTS idx_clf_run ON cleaner_run_flagged(run_id);

CREATE TABLE IF NOT EXISTS cleaner_runs (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at      INTEGER NOT NULL,
    finished_at     INTEGER,
    reason          TEXT NOT NULL,
    scope           TEXT,
    selected_count  INTEGER DEFAULT 0,
    found_count     INTEGER DEFAULT 0,
    total_evaluated INTEGER DEFAULT 0,
    status          TEXT DEFAULT 'running',
    error           TEXT
);
CREATE INDEX IF NOT EXISTS idx_cleaner_runs_started ON cleaner_runs(started_at);

CREATE TABLE IF NOT EXISTS recycle_bin (
    id             TEXT NOT NULL,
    source_table   TEXT NOT NULL,
    platform       TEXT NOT NULL,
    subreddit      TEXT,
    title          TEXT,
    body           TEXT,
    author         TEXT,
    score          INTEGER,
    created_utc    INTEGER,
    url            TEXT,
    record_json    TEXT NOT NULL,
    deleted_at     INTEGER NOT NULL,
    deleted_reason TEXT,
    session_id     TEXT NOT NULL,
    PRIMARY KEY (id, source_table)
);
CREATE INDEX IF NOT EXISTS idx_recycle_platform  ON recycle_bin(platform);
CREATE INDEX IF NOT EXISTS idx_recycle_subreddit ON recycle_bin(subreddit);
CREATE INDEX IF NOT EXISTS idx_recycle_session   ON recycle_bin(session_id);
CREATE INDEX IF NOT EXISTS idx_recycle_deleted   ON recycle_bin(deleted_at);
"""


def _extract_media_json(item: dict) -> Optional[str]:
    """Extract gallery images and video URLs from a Reddit post dict."""
    media: list[dict] = []
    gallery_data = item.get("gallery_data") or {}
    media_metadata = item.get("media_metadata") or {}
    if gallery_data and media_metadata:
        for gitem in (gallery_data.get("items") or []):
            mid = gitem.get("media_id")
            if not mid or mid not in media_metadata:
                continue
            meta = media_metadata[mid]
            if meta.get("status") != "valid":
                continue
            src = meta.get("s") or {}
            url = src.get("u") or src.get("gif")
            if url:
                media.append({"type": "image", "url": url.replace("&", "&")})
    if item.get("is_video"):
        rv = ((item.get("media") or {}).get("reddit_video") or {})
        fb = rv.get("fallback_url")
        if fb:
            media.append({
                "type": "video",
                "url": fb.replace("&", "&"),
                "hls_url": (rv.get("hls_url") or "").replace("&", "&"),
            })
    return json.dumps(media) if media else None


def _strip_t_prefix(s: Optional[str]) -> Optional[str]:
    if not s:
        return s
    if s.startswith(("t1_", "t3_")):
        return s[3:]
    return s


class Store:
    def __init__(self, path: str | Path):
        self.path = str(path)
        self.conn = sqlite3.connect(self.path, timeout=30)
        self.conn.row_factory = sqlite3.Row
        self.conn.executescript(SCHEMA)
        self.conn.execute("PRAGMA journal_mode=WAL")
        # Non-breaking migrations: add columns and indexes introduced after initial schema
        for ddl in [
            "ALTER TABLE posts ADD COLUMN media_json TEXT",
            "ALTER TABLE blind_comments ADD COLUMN parent_id TEXT",
            "CREATE INDEX IF NOT EXISTS idx_blind_comments_parent ON blind_comments(parent_id)",
            "ALTER TABLE blind_posts ADD COLUMN channel TEXT",
            "ALTER TABLE blind_comments ADD COLUMN company TEXT",
            "ALTER TABLE recycle_bin ADD COLUMN run_id INTEGER",
            "CREATE INDEX IF NOT EXISTS idx_recycle_run ON recycle_bin(run_id)",
            "ALTER TABLE cleaner_run_flagged ADD COLUMN qwen_reason TEXT",
        ]:
            try:
                self.conn.execute(ddl)
            except sqlite3.OperationalError:
                pass  # column already exists
        self.conn.commit()

    def upsert_post(self, item: dict, source: str = "pullpush") -> bool:
        """Insert new post; update only mutable fields if it already exists.

        fetched_at and source are write-once so posts remain anchored to the
        run that first captured them.  Returns True only for genuine new rows.
        """
        now = int(time.time())
        pid = _strip_t_prefix(item.get("id"))
        if not pid:
            return False
        media_json = _extract_media_json(item)
        try:
            cur = self.conn.execute(
                """
                INSERT OR IGNORE INTO posts (
                    id, subreddit, author, title, selftext, score, num_comments,
                    created_utc, permalink, url, is_self, over_18, fetched_at, source, media_json
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    pid,
                    (item.get("subreddit") or "").lower(),
                    item.get("author"),
                    item.get("title"),
                    item.get("selftext"),
                    item.get("score"),
                    item.get("num_comments"),
                    int(item.get("created_utc") or 0),
                    item.get("permalink"),
                    item.get("url"),
                    int(bool(item.get("is_self", False))),
                    int(bool(item.get("over_18", False))),
                    now,
                    source,
                    media_json,
                ),
            )
            if cur.rowcount == 0:
                # Already exists — refresh mutable fields; fill media_json if missing
                self.conn.execute(
                    "UPDATE posts SET score=?, num_comments=?, media_json=COALESCE(media_json, ?) WHERE id=?",
                    (item.get("score"), item.get("num_comments"), media_json, pid),
                )
                return False
            return True
        except sqlite3.Error as e:
            log.error("upsert_post(%s) failed: %s", pid, e)
            return False

    def upsert_comment(self, item: dict, source: str = "pullpush") -> bool:
        """Insert new comment; update only mutable fields if it already exists.

        fetched_at and source are write-once.  Returns True only for new rows.
        """
        now = int(time.time())
        cid = _strip_t_prefix(item.get("id"))
        if not cid:
            return False
        post_id = _strip_t_prefix(item.get("link_id") or item.get("post_id"))
        parent_id = _strip_t_prefix(item.get("parent_id"))
        try:
            cur = self.conn.execute(
                """
                INSERT OR IGNORE INTO comments (
                    id, post_id, parent_id, subreddit, author, body, score,
                    created_utc, permalink, fetched_at, source
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    cid,
                    post_id or "",
                    parent_id,
                    (item.get("subreddit") or "").lower(),
                    item.get("author"),
                    item.get("body"),
                    item.get("score"),
                    int(item.get("created_utc") or 0),
                    item.get("permalink"),
                    now,
                    source,
                ),
            )
            if cur.rowcount == 0:
                self.conn.execute(
                    "UPDATE comments SET score=?, body=? WHERE id=?",
                    (item.get("score"), item.get("body"), cid),
                )
                return False
            return True
        except sqlite3.Error as e:
            log.error("upsert_comment(%s) failed: %s", cid, e)
            return False

    def upsert_subreddit(self, meta: dict, discovered_via: str) -> None:
        now = int(time.time())
        self.conn.execute(
            """
            INSERT INTO subreddits (
                name, subscribers, description, lang, over_18, discovered_via, discovered_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(name) DO UPDATE SET
                subscribers = excluded.subscribers,
                description = excluded.description
            """,
            (
                meta.get("name", ""),
                int(meta.get("subscribers") or 0),
                meta.get("description") or "",
                meta.get("lang") or "",
                int(bool(meta.get("over_18", False))),
                discovered_via,
                now,
            ),
        )

    def ensure_subreddit(self, name: str) -> None:
        """Create a minimal subreddit record if none exists yet."""
        self.conn.execute(
            """INSERT OR IGNORE INTO subreddits
               (name, subscribers, description, lang, over_18, discovered_via, discovered_at, selected)
               VALUES (?, 0, '', '', 0, 'keyword_pull', ?, 0)""",
            (name, int(time.time())),
        )

    def set_selected(self, name: str, selected: bool) -> bool:
        cur = self.conn.execute(
            "UPDATE subreddits SET selected = ? WHERE name = ?",
            (1 if selected else 0, name),
        )
        if cur.rowcount == 0:
            self.conn.execute(
                """INSERT INTO subreddits (name, subscribers, description, lang, over_18,
                                           discovered_via, discovered_at, selected)
                   VALUES (?, 0, '', '', 0, 'manual', ?, ?)""",
                (name, int(time.time()), 1 if selected else 0),
            )
        self.conn.commit()
        return True

    def start_run(self, kind: str, target: str, params: Optional[dict] = None) -> int:
        cur = self.conn.execute(
            "INSERT INTO pull_runs (kind, target, started_at, params) VALUES (?, ?, ?, ?)",
            (kind, target, int(time.time()), json.dumps(params or {})),
        )
        self.conn.commit()
        return int(cur.lastrowid)

    def finish_run(
        self,
        run_id: int,
        posts_added: int,
        comments_added: int,
        error: Optional[str] = None,
    ) -> None:
        self.conn.execute(
            """UPDATE pull_runs
               SET finished_at = ?, posts_added = ?, comments_added = ?, error = ?
               WHERE id = ?""",
            (int(time.time()), posts_added, comments_added, error, run_id),
        )
        self.conn.commit()

    def selected_subreddits(self) -> list[str]:
        rows = self.conn.execute(
            "SELECT name FROM subreddits WHERE selected = 1 ORDER BY name"
        ).fetchall()
        return [r["name"] for r in rows]

    def list_subreddits(self) -> list[dict]:
        rows = self.conn.execute(
            """SELECT name, subscribers, description, discovered_via, selected
               FROM subreddits ORDER BY selected DESC, subscribers DESC"""
        ).fetchall()
        return [dict(r) for r in rows]

    def stats(self) -> dict:
        posts = self.conn.execute("SELECT COUNT(*) FROM posts").fetchone()[0]
        comments = self.conn.execute("SELECT COUNT(*) FROM comments").fetchone()[0]
        by_sub_c = self.conn.execute(
            """SELECT subreddit, COUNT(*) AS n FROM comments
               GROUP BY subreddit ORDER BY n DESC LIMIT 20"""
        ).fetchall()
        by_sub_p = self.conn.execute(
            """SELECT subreddit, COUNT(*) AS n FROM posts
               GROUP BY subreddit ORDER BY n DESC LIMIT 20"""
        ).fetchall()
        runs = self.conn.execute("SELECT COUNT(*) FROM pull_runs").fetchone()[0]
        recent_runs = self.conn.execute(
            """SELECT id, kind, target, started_at, finished_at, posts_added, comments_added, error
               FROM pull_runs ORDER BY id DESC LIMIT 10"""
        ).fetchall()
        return {
            "posts": posts,
            "comments": comments,
            "comments_by_subreddit": [dict(r) for r in by_sub_c],
            "posts_by_subreddit": [dict(r) for r in by_sub_p],
            "total_runs": runs,
            "recent_runs": [dict(r) for r in recent_runs],
        }

    def search_posts_like(
        self,
        pattern: str,
        subreddit: Optional[str] = None,
        limit: int = 100,
    ) -> list[dict]:
        sql = """SELECT id, subreddit, author, title, selftext, score, num_comments,
                        created_utc, permalink, url, is_self, over_18, fetched_at, source
                 FROM posts
                 WHERE (title LIKE ? OR selftext LIKE ?)"""
        like = f"%{pattern}%"
        params: list = [like, like]
        if subreddit:
            sql += " AND LOWER(subreddit) = LOWER(?)"
            params.append(subreddit)
        sql += " ORDER BY created_utc DESC LIMIT ?"
        params.append(limit)
        return [dict(r) for r in self.conn.execute(sql, params).fetchall()]

    def search_comments_like(
        self,
        pattern: str,
        subreddit: Optional[str] = None,
        limit: int = 100,
    ) -> list[dict]:
        sql = """SELECT id, post_id, subreddit, author, body, score, created_utc, permalink
                 FROM comments
                 WHERE body LIKE ?"""
        params: list = [f"%{pattern}%"]
        if subreddit:
            sql += " AND LOWER(subreddit) = LOWER(?)"
            params.append(subreddit)
        sql += " ORDER BY created_utc DESC LIMIT ?"
        params.append(limit)
        return [dict(r) for r in self.conn.execute(sql, params).fetchall()]

    def iter_all(self, kind: str, subreddit: Optional[str] = None):
        if kind not in ("posts", "comments"):
            raise ValueError(f"unknown kind: {kind}")
        sql = f"SELECT * FROM {kind}"
        params: list = []
        if subreddit:
            sql += " WHERE subreddit = ?"
            params.append(subreddit)
        for row in self.conn.execute(sql, params):
            yield dict(row)

    # ---------------------------------------------------------------- Blind

    def ensure_blind_topic(self, slug: str, name: str, description: str = "") -> None:
        self.conn.execute(
            """INSERT INTO blind_topics (slug, name, description)
               VALUES (?, ?, ?)
               ON CONFLICT(slug) DO UPDATE SET
                 name        = excluded.name,
                 description = CASE WHEN excluded.description != '' THEN excluded.description
                                    ELSE blind_topics.description END""",
            (slug, name, description),
        )

    def upsert_blind_post(self, item: dict) -> bool:
        """Returns True if a new row was inserted (False = already existed)."""
        now = int(time.time())
        cur = self.conn.execute(
            """INSERT INTO blind_posts
               (id, topic_slug, channel, title, body, company, score, comment_count, created_at, url, fetched_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
               ON CONFLICT(id) DO UPDATE SET
                 channel       = excluded.channel,
                 score         = excluded.score,
                 comment_count = excluded.comment_count,
                 fetched_at    = excluded.fetched_at""",
            (
                item["id"],
                item.get("topic_slug", ""),
                item.get("channel"),
                item.get("title"),
                item.get("body"),
                item.get("company"),
                item.get("score"),
                item.get("comment_count"),
                item.get("created_at"),
                item.get("url"),
                now,
            ),
        )
        return cur.rowcount > 0

    def upsert_blind_comment(self, item: dict) -> bool:
        now = int(time.time())
        cur = self.conn.execute(
            """INSERT INTO blind_comments
               (id, post_id, topic_slug, parent_id, company, body, score, created_at, fetched_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
               ON CONFLICT(id) DO UPDATE SET
                 score      = excluded.score,
                 parent_id  = excluded.parent_id,
                 company    = excluded.company,
                 fetched_at = excluded.fetched_at""",
            (
                item["id"],
                item.get("post_id", ""),
                item.get("topic_slug", ""),
                item.get("parent_id"),
                item.get("company"),
                item.get("body"),
                item.get("score"),
                item.get("created_at"),
                now,
            ),
        )
        return cur.rowcount > 0

    def update_blind_topic_counts(self, slug: str) -> None:
        post_count = self.conn.execute(
            "SELECT COUNT(*) FROM blind_posts WHERE topic_slug = ?", (slug,)
        ).fetchone()[0]
        comment_count = self.conn.execute(
            "SELECT COUNT(*) FROM blind_comments WHERE topic_slug = ?", (slug,)
        ).fetchone()[0]
        self.conn.execute(
            """UPDATE blind_topics
               SET post_count = ?, comment_count = ?, last_scraped_at = ?
               WHERE slug = ?""",
            (post_count, comment_count, int(time.time()), slug),
        )

    # ---------------------------------------------------------------- Cleaner runs

    def save_flagged_posts(self, run_id: int, posts: list[dict]) -> None:
        """Store Qwen's delete recommendations for a run."""
        for p in posts:
            try:
                self.conn.execute(
                    """INSERT OR IGNORE INTO cleaner_run_flagged
                       (run_id, post_id, subreddit, title, body, qwen_reason)
                       VALUES (?,?,?,?,?,?)""",
                    (
                        run_id, p["id"], p.get("subreddit"), p.get("title"),
                        (p.get("body") or "")[:500], p.get("qwen_reason"),
                    ),
                )
            except Exception as e:
                log.debug("save_flagged_posts(%s): %s", p.get("id"), e)
        self.conn.commit()

    def get_flagged_posts_for_run(self, run_id: int) -> list[dict]:
        rows = self.conn.execute(
            "SELECT * FROM cleaner_run_flagged WHERE run_id = ? ORDER BY rowid",
            (run_id,),
        ).fetchall()
        return [dict(r) for r in rows]

    def get_all_flagged_posts(self) -> list[dict]:
        """All Qwen-flagged posts across all runs that are not yet in the recycle_bin."""
        rows = self.conn.execute(
            """SELECT f.run_id, f.post_id, f.subreddit, f.title, f.body, f.qwen_reason,
                      r.reason AS run_reason, r.started_at
               FROM cleaner_run_flagged f
               JOIN cleaner_runs r ON f.run_id = r.id
               WHERE f.post_id NOT IN (SELECT id FROM recycle_bin)
               ORDER BY r.started_at DESC"""
        ).fetchall()
        return [dict(r) for r in rows]

    def unflag_posts(self, post_ids: list[str]) -> int:
        """Remove posts from cleaner_run_flagged (un-stage from Qwen deletion queue)."""
        if not post_ids:
            return 0
        ph = ",".join("?" * len(post_ids))
        cur = self.conn.execute(
            f"DELETE FROM cleaner_run_flagged WHERE post_id IN ({ph})", post_ids
        )
        self.conn.commit()
        return cur.rowcount

    def start_cleaner_run(self, reason: str, scope: list[str], selected_count: int) -> int:
        cur = self.conn.execute(
            "INSERT INTO cleaner_runs (started_at, reason, scope, selected_count) VALUES (?,?,?,?)",
            (int(time.time()), reason, json.dumps(scope), selected_count),
        )
        self.conn.commit()
        return int(cur.lastrowid)

    def finish_cleaner_run(
        self,
        run_id: int,
        found_count: int,
        total_evaluated: int,
        status: str = "done",
        error: Optional[str] = None,
    ) -> None:
        self.conn.execute(
            """UPDATE cleaner_runs
               SET finished_at=?, found_count=?, total_evaluated=?, status=?, error=?
               WHERE id=?""",
            (int(time.time()), found_count, total_evaluated, status, error, run_id),
        )
        self.conn.commit()

    def list_cleaner_runs(self, limit: int = 50) -> list[dict]:
        rows = self.conn.execute(
            "SELECT * FROM cleaner_runs ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()
        return [dict(r) for r in rows]

    # ---------------------------------------------------------------- Cleaner / recycle bin

    def get_cleaner_run(self, run_id: int) -> Optional[dict]:
        row = self.conn.execute(
            "SELECT * FROM cleaner_runs WHERE id = ?", (run_id,)
        ).fetchone()
        return dict(row) if row else None

    def get_recycle_bin_for_run(self, run_id: int) -> list[dict]:
        rows = self.conn.execute(
            "SELECT * FROM recycle_bin WHERE run_id = ? ORDER BY deleted_at DESC", (run_id,)
        ).fetchall()
        return [dict(r) for r in rows]

    def move_posts_to_recycle_bin(
        self,
        post_ids: list[str],
        platform: str,
        reason: str,
        session_id: str,
        run_id: Optional[int] = None,
    ) -> int:
        """Copy posts/blind_posts to recycle_bin then delete from source. Returns moved count."""
        now = int(time.time())
        moved = 0
        source_table = "posts" if platform == "reddit" else "blind_posts"
        placeholders = ",".join("?" * len(post_ids))

        if platform == "reddit":
            rows = self.conn.execute(
                f"SELECT * FROM posts WHERE id IN ({placeholders})", post_ids
            ).fetchall()
            for row in rows:
                d = dict(row)
                self.conn.execute(
                    """INSERT OR REPLACE INTO recycle_bin
                       (id, source_table, platform, subreddit, title, body, author, score,
                        created_utc, url, record_json, deleted_at, deleted_reason, session_id, run_id)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        d["id"], source_table, platform,
                        d.get("subreddit"), d.get("title"), d.get("selftext"),
                        d.get("author"), d.get("score"),
                        d.get("created_utc"), d.get("url"),
                        json.dumps(d), now, reason, session_id, run_id,
                    ),
                )
                moved += 1
            self.conn.execute(
                f"DELETE FROM posts WHERE id IN ({placeholders})", post_ids
            )
        else:
            rows = self.conn.execute(
                f"SELECT * FROM blind_posts WHERE id IN ({placeholders})", post_ids
            ).fetchall()
            for row in rows:
                d = dict(row)
                self.conn.execute(
                    """INSERT OR REPLACE INTO recycle_bin
                       (id, source_table, platform, subreddit, title, body, author, score,
                        created_utc, url, record_json, deleted_at, deleted_reason, session_id, run_id)
                       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                    (
                        d["id"], source_table, platform,
                        d.get("topic_slug"), d.get("title"), d.get("body"),
                        None, d.get("score"),
                        d.get("created_at"), d.get("url"),
                        json.dumps(d), now, reason, session_id, run_id,
                    ),
                )
                moved += 1
            self.conn.execute(
                f"DELETE FROM blind_posts WHERE id IN ({placeholders})", post_ids
            )
        self.conn.commit()
        return moved

    def get_recycle_bin(self) -> list[dict]:
        """Return all recycle bin records ordered newest-first."""
        rows = self.conn.execute(
            "SELECT * FROM recycle_bin ORDER BY deleted_at DESC"
        ).fetchall()
        return [dict(r) for r in rows]

    def restore_from_recycle_bin(self, ids: list[str]) -> int:
        """Re-insert records back into their source table and remove from recycle_bin."""
        placeholders = ",".join("?" * len(ids))
        rows = self.conn.execute(
            f"SELECT * FROM recycle_bin WHERE id IN ({placeholders})", ids
        ).fetchall()
        restored = 0
        for row in rows:
            d = dict(row)
            record = json.loads(d["record_json"])
            table = d["source_table"]
            if table == "posts":
                self.conn.execute(
                    """INSERT OR IGNORE INTO posts
                       (id, subreddit, author, title, selftext, score, num_comments,
                        created_utc, permalink, url, is_self, over_18, fetched_at, source, media_json)
                       VALUES (:id,:subreddit,:author,:title,:selftext,:score,:num_comments,
                               :created_utc,:permalink,:url,:is_self,:over_18,:fetched_at,:source,:media_json)""",
                    record,
                )
            elif table == "blind_posts":
                self.conn.execute(
                    """INSERT OR IGNORE INTO blind_posts
                       (id, topic_slug, channel, title, body, company, score, comment_count,
                        created_at, url, fetched_at)
                       VALUES (:id,:topic_slug,:channel,:title,:body,:company,:score,:comment_count,
                               :created_at,:url,:fetched_at)""",
                    record,
                )
            restored += 1
        self.conn.execute(
            f"DELETE FROM recycle_bin WHERE id IN ({placeholders})", ids
        )
        self.conn.commit()
        return restored

    def purge_recycle_session(self, session_id: str) -> int:
        """Hard-delete all recycle bin records for a session (and orphaned comments). Returns count."""
        rows = self.conn.execute(
            "SELECT id, source_table FROM recycle_bin WHERE session_id = ?", (session_id,)
        ).fetchall()
        if not rows:
            return 0
        reddit_ids = [r["id"] for r in rows if r["source_table"] == "posts"]
        blind_ids = [r["id"] for r in rows if r["source_table"] == "blind_posts"]
        if reddit_ids:
            ph = ",".join("?" * len(reddit_ids))
            self.conn.execute(f"DELETE FROM comments WHERE post_id IN ({ph})", reddit_ids)
        if blind_ids:
            ph = ",".join("?" * len(blind_ids))
            self.conn.execute(f"DELETE FROM blind_comments WHERE post_id IN ({ph})", blind_ids)
        count = self.conn.execute(
            "SELECT COUNT(*) FROM recycle_bin WHERE session_id = ?", (session_id,)
        ).fetchone()[0]
        self.conn.execute("DELETE FROM recycle_bin WHERE session_id = ?", (session_id,))
        self.conn.commit()
        return count

    def purge_recycle_run(self, run_id: int) -> int:
        """Hard-delete all recycle bin records for a cleaner run (and orphaned comments). Returns count."""
        rows = self.conn.execute(
            "SELECT id, source_table FROM recycle_bin WHERE run_id = ?", (run_id,)
        ).fetchall()
        if not rows:
            return 0
        reddit_ids = [r["id"] for r in rows if r["source_table"] == "posts"]
        blind_ids = [r["id"] for r in rows if r["source_table"] == "blind_posts"]
        if reddit_ids:
            ph = ",".join("?" * len(reddit_ids))
            self.conn.execute(f"DELETE FROM comments WHERE post_id IN ({ph})", reddit_ids)
        if blind_ids:
            ph = ",".join("?" * len(blind_ids))
            self.conn.execute(f"DELETE FROM blind_comments WHERE post_id IN ({ph})", blind_ids)
        count = self.conn.execute(
            "SELECT COUNT(*) FROM recycle_bin WHERE run_id = ?", (run_id,)
        ).fetchone()[0]
        self.conn.execute("DELETE FROM recycle_bin WHERE run_id = ?", (run_id,))
        self.conn.commit()
        return count

    def commit(self) -> None:
        self.conn.commit()

    def close(self) -> None:
        self.conn.close()