From cd52d6612c58b1c5db6bde38f8f6ce41b66677bd Mon Sep 17 00:00:00 2001 From: BadStorm Date: Thu, 3 Sep 2026 18:45:09 +0200 Subject: [PATCH] containers/talkbot: serialize webhook processing through a persistent queue --- containers/talkbot/README.md | 16 ++ containers/talkbot/server.py | 289 ++++++++++++++++++++------- containers/talkbot/talkbot.container | 5 + 3 files changed, 238 insertions(+), 72 deletions(-) diff --git a/containers/talkbot/README.md b/containers/talkbot/README.md index dfeb8c7..145d5ba 100644 --- a/containers/talkbot/README.md +++ b/containers/talkbot/README.md @@ -135,6 +135,22 @@ conversazione Talk dove il bot è abilitato: vedi sezione "Stato dei test". | `VOICE_SAMPLES_DIR` | no | `/app/voice-samples` | Cartella con i campioni `.wav` per il cloning | | `DEFAULT_TARGET_LANGUAGE` | no | `en` | Lingua di fallback se non si riesce a leggere quella dell'utente | | `QWEN_TTS_SPEAKER` | no | `Ryan` | Voce preset usata quando non c'è un campione utente | +| `QUEUE_DB_PATH` | no | `/app/data/queue.db` | File SQLite della coda dei messaggi in attesa (vedi sotto) | + +## Coda dei messaggi + +Il webhook non elabora i messaggi in modo sincrono: fa solo il parsing/verifica +della richiesta e mette il job in coda, rispondendo subito a Nextcloud. Un +singolo worker in background consuma la coda **in sequenza** (un messaggio +alla volta, per non sovraccaricare whisper/llama.cpp/qwen-tts con richieste +in parallelo su hardware limitato). + +La coda è persistita su SQLite (`QUEUE_DB_PATH`, di default +`/app/data/queue.db`, montato come volume in `talkbot.container`) invece che +in memoria: un job viene rimosso dalla tabella solo a elaborazione completata, +quindi se il container si blocca o viene riavviato mentre ci sono messaggi in +coda (anche quello attualmente in lavorazione), al riavvio successivo vengono +ripresi ed elaborati normalmente, senza perdita. ## Stato dei test / cosa manca ancora diff --git a/containers/talkbot/server.py b/containers/talkbot/server.py index 9bdbe1b..c68edbe 100644 --- a/containers/talkbot/server.py +++ b/containers/talkbot/server.py @@ -2,7 +2,9 @@ import json import logging import os import re -from datetime import datetime +import sqlite3 +import threading +from datetime import datetime, timezone from typing import Optional import httpx @@ -45,10 +47,82 @@ VOICE_SAMPLES_DIR = os.environ.get( "/app/voice-samples", ) +QUEUE_DB_PATH = os.environ.get( + "QUEUE_DB_PATH", + "/app/data/queue.db", +) + MANAGER_USER = NC_USER app = FastAPI(title="Nextcloud Talk voice translator") +# Events are processed one at a time by a single background worker: the +# pipeline (whisper -> llama.cpp -> qwen-tts) is heavy and the host has +# limited resources, so we serialize instead of processing concurrently. +# The webhook handler only validates/parses the payload (no blocking I/O) +# and enqueues the job, so Nextcloud always gets an immediate response +# and never times out waiting for a translation to finish. +# +# The queue itself lives in a SQLite file (QUEUE_DB_PATH) instead of +# memory: a job row is only deleted after it has been fully processed, +# so if the container is killed/restarted mid-queue, whatever is still +# in the table gets picked up and processed again on the next startup. +new_job_signal = threading.Event() + + +def init_queue_db() -> None: + os.makedirs(os.path.dirname(QUEUE_DB_PATH), exist_ok=True) + + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + payload TEXT NOT NULL, + created_at TEXT NOT NULL + ) + """ + ) + conn.commit() + + +def enqueue_event(job: dict) -> None: + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + conn.execute( + "INSERT INTO queue (payload, created_at) VALUES (?, ?)", + ( + json.dumps(job), + datetime.now(timezone.utc).isoformat(), + ), + ) + conn.commit() + + new_job_signal.set() + + +def dequeue_next_event() -> Optional[tuple[int, dict]]: + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + row = conn.execute( + "SELECT id, payload FROM queue ORDER BY id LIMIT 1" + ).fetchone() + + if not row: + return None + + row_id, payload = row + return row_id, json.loads(payload) + + +def remove_queued_event(row_id: int) -> None: + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + conn.execute("DELETE FROM queue WHERE id = ?", (row_id,)) + conn.commit() + + +def queue_size() -> int: + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + return conn.execute("SELECT COUNT(*) FROM queue").fetchone()[0] + LANGUAGE_NAMES = { "en": "English", "it": "Italian", @@ -990,6 +1064,128 @@ def process_voice_message( ) +def process_event(job: dict) -> None: + token = job["token"] + message_id = job["message_id"] + sender_user_id = job["sender_user_id"] + sender_display_name = job["sender_display_name"] + file_param = job.get("file_param") + message_text = job.get("message_text", "") + + try: + recipients = get_target_user_ids( + token, + sender_user_id, + ) + except Exception as exc: + log.exception( + "Could not load participants for conversation %s: %s", + token, + exc, + ) + recipients = [] + + if file_param: + try: + process_voice_message( + token=token, + message_id=message_id, + sender_user_id=sender_user_id, + sender_display_name=sender_display_name, + file_param=file_param, + recipients=recipients, + ) + + except Exception as exc: + log.exception( + "Voice processing failed for message %s: %s", + message_id, + exc, + ) + + return + + if message_text: + log.info( + "Text message %s from %s", + message_id, + sender_user_id, + ) + + for recipient in recipients: + target_language = normalize_language_code( + get_user_language(recipient), + ) + + translated = translate( + message_text, + "auto", + target_language, + ) + + if translated.strip().lower() == message_text.lower(): + continue + + post_manager_message( + token, + translated, + ) + + log.info( + "Posted text translation for %s: %s", + recipient, + target_language, + ) + + return + + log.info( + "Message %s has no text or attachment", + message_id, + ) + + +def event_worker() -> None: + while True: + job_row = dequeue_next_event() + + if job_row is None: + new_job_signal.wait(timeout=5) + new_job_signal.clear() + continue + + row_id, job = job_row + + try: + process_event(job) + except Exception as exc: + log.exception( + "Unhandled error processing message %s: %s", + job.get("message_id"), + exc, + ) + finally: + remove_queued_event(row_id) + + +@app.on_event("startup") +def start_event_worker() -> None: + init_queue_db() + + pending = queue_size() + if pending: + log.info( + "Resuming %s queued message(s) left over from a previous run", + pending, + ) + + threading.Thread( + target=event_worker, + name="talkbot-event-worker", + daemon=True, + ).start() + + @app.get("/health") def health(): return { @@ -998,6 +1194,7 @@ def health(): "llamacpp_timeout": LLAMACPP_TIMEOUT, "llamacpp_thinking": LLAMACPP_ENABLE_THINKING, "qwen_tts_timeout": QWEN_TTS_TIMEOUT, + "queue_size": queue_size(), } @@ -1101,91 +1298,39 @@ async def webhook( parameters, ) - try: - recipients = get_target_user_ids( - token, - sender_user_id, - ) - except Exception as exc: - log.exception( - "Could not load participants for conversation %s: %s", - token, - exc, - ) - recipients = [] - - if file_param: - try: - process_voice_message( - token=token, - message_id=int(message_id), - sender_user_id=sender_user_id, - sender_display_name=sender_display_name, - file_param=file_param, - recipients=recipients, - ) - - except Exception as exc: - log.exception( - "Voice processing failed for message %s: %s", - message_id, - exc, - ) - - return JSONResponse( - { - "status": "ok", - "type": "voice", - } - ) - - if message_text: + if not file_param and not message_text: log.info( - "Text message %s from %s", + "Message %s has no text or attachment", message_id, - sender_user_id, ) - for recipient in recipients: - target_language = normalize_language_code( - get_user_language(recipient), - ) - - translated = translate( - message_text, - "auto", - target_language, - ) - - if translated.strip().lower() == message_text.lower(): - continue - - post_manager_message( - token, - translated, - ) - - log.info( - "Posted text translation for %s: %s", - recipient, - target_language, - ) - return JSONResponse( { - "status": "ok", - "type": "text", + "status": "ignored", + "reason": "empty message", } ) + enqueue_event( + { + "token": token, + "message_id": int(message_id), + "sender_user_id": sender_user_id, + "sender_display_name": sender_display_name, + "file_param": file_param, + "message_text": message_text, + } + ) + log.info( - "Message %s has no text or attachment", + "Queued message %s (%s pending)", message_id, + queue_size(), ) return JSONResponse( { - "status": "ignored", - "reason": "empty message", + "status": "queued", + "type": "voice" if file_param else "text", } ) diff --git a/containers/talkbot/talkbot.container b/containers/talkbot/talkbot.container index 591b116..503997d 100644 --- a/containers/talkbot/talkbot.container +++ b/containers/talkbot/talkbot.container @@ -16,6 +16,11 @@ SecurityLabelType=container_runtime_t Volume=/srv/containers/talkbot/voices:/app/voice-samples:Z,ro Environment=VOICE_SAMPLES_DIR=/app/voice-samples +# Persistent job queue (SQLite): survives container restarts, so messages +# that were queued but not yet processed are picked up again on startup. +Volume=/srv/containers/talkbot/data:/app/data:Z +#Environment=QUEUE_DB_PATH=/app/data/queue.db + # --- Nextcloud connection --- Environment=NC_URL=https://your-nextcloud.example.tld # Shared secret used when registering the bot via: