diff --git a/src/bot/bot.py b/src/bot/bot.py index 6c63c8b..fa1a1af 100644 --- a/src/bot/bot.py +++ b/src/bot/bot.py @@ -18,7 +18,7 @@ from telegram.ext import ( from telegram.constants import ParseMode from src.config import settings -from src.db.database import get_db, ResearchDB, ResearchStatus, OutputType +from src.db.database import get_db, close_db, ResearchDB, ResearchStatus, OutputType from src.scraper.exhaustive import ExhaustiveScraper from src.processor.processor import OllamaClient, ContentProcessor from src.generator.generator import OutputGenerator @@ -913,6 +913,10 @@ async def _on_startup(app: Application) -> None: await _start_scheduler(app) +async def _on_shutdown(app: Application) -> None: + await close_db() + + async def cmd_export(update: Update, ctx: ContextTypes.DEFAULT_TYPE): if not is_authorized(update.effective_user.id): return @@ -1259,6 +1263,7 @@ def create_bot() -> Application: Application.builder() .token(settings.telegram_bot_token) .post_init(_on_startup) + .post_shutdown(_on_shutdown) .build() ) diff --git a/src/db/database.py b/src/db/database.py index a59b472..084b798 100644 --- a/src/db/database.py +++ b/src/db/database.py @@ -1,3 +1,4 @@ +import asyncio import aiosqlite import json import time @@ -117,19 +118,65 @@ CREATE TABLE IF NOT EXISTS watched_topics ( """ +# Conexión única compartida por proceso. aiosqlite serializa todas las +# operaciones en el hilo de la conexión, así que compartirla entre coroutines +# es seguro y además evita la contención del lock de escritura a nivel de +# fichero que sufrían las conexiones múltiples (scheduler, /compare). +_shared_conn: Optional[aiosqlite.Connection] = None +_init_lock = asyncio.Lock() + + +class _SharedConnection: + """Proxy sobre la conexión compartida cuyo close() es no-op. + + Permite que los handlers sigan usando el patrón `db = await get_db()` … + `finally: await db.close()` sin cambios, mientras por debajo se reutiliza + una sola conexión real durante toda la vida del proceso. + """ + def __init__(self, conn: aiosqlite.Connection): + self._conn = conn + + def __getattr__(self, name): + return getattr(self._conn, name) + + async def close(self): + # No cerrar: la conexión es compartida por todo el proceso. + pass + + +async def _init_shared() -> aiosqlite.Connection: + global _shared_conn + if _shared_conn is None: + async with _init_lock: + if _shared_conn is None: # doble check tras adquirir el lock + Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True) + conn = await aiosqlite.connect(settings.db_path) + conn.row_factory = aiosqlite.Row + await conn.execute("PRAGMA journal_mode=WAL") + await conn.execute("PRAGMA synchronous=NORMAL") + # Espera hasta 5s si OTRO proceso tiene el lock de escritura + # antes de fallar con "database is locked". + await conn.execute("PRAGMA busy_timeout=5000") + await conn.executescript(SCHEMA) + await conn.commit() + _shared_conn = conn + logger.info("Shared DB connection initialized", path=settings.db_path) + return _shared_conn + + async def get_db() -> aiosqlite.Connection: - Path(settings.db_path).parent.mkdir(parents=True, exist_ok=True) - db = await aiosqlite.connect(settings.db_path) - db.row_factory = aiosqlite.Row - await db.execute("PRAGMA journal_mode=WAL") - await db.execute("PRAGMA synchronous=NORMAL") - # Espera hasta 5s si otra conexión tiene el lock de escritura (scheduler - # solapado con research activa, o las 2 sesiones de /compare) en vez de - # fallar al instante con "database is locked". - await db.execute("PRAGMA busy_timeout=5000") - await db.executescript(SCHEMA) - await db.commit() - return db + conn = await _init_shared() + return _SharedConnection(conn) + + +async def close_db() -> None: + """Cierra la conexión compartida (checkpoint WAL). Llamar al apagar el bot.""" + global _shared_conn + if _shared_conn is not None: + try: + await _shared_conn.close() + finally: + _shared_conn = None class ResearchDB: