containers/talkbot: serialize webhook processing through a persistent queue

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 `<talk-username>.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
 

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",
         }
     )

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: