Minz
ef90ee6a0b
Stats panel: a persistent spinner marks whichever target is currently being scanned/reacted to, and jittered waits show a live-ticking countdown bar instead of a static "waiting Xs" line. Both the status line and countdown row are always rendered (blank when idle) so their appearing/disappearing no longer shifts the table's height. Backlog: channel_scan_state now tracks oldest_covered_at alongside newest_id_seen. If days_back is widened after a channel already has a resume checkpoint, the forward-only diff would never look back far enough to notice — now the newly-exposed gap gets backfilled first.
275 lines
12 KiB
Python
275 lines
12 KiB
Python
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()
|