diff --git a/backend/app/worker/background/channel.py b/backend/app/worker/background/channel.py new file mode 100644 index 0000000..dbd9077 --- /dev/null +++ b/backend/app/worker/background/channel.py @@ -0,0 +1,27 @@ +"""In-process handler for channel-bot webhook events (Telegram, Slack). + +Dispatched via FastAPI ``BackgroundTasks`` from the webhook routes so the platform +gets a fast 200 OK while routing/AI happens in the background. +""" + +import logging +from typing import Any + +from app.db.session import get_db_context +from app.services.channels.router import ChannelMessageRouter + +logger = logging.getLogger(__name__) + + +async def process_channel_event(incoming: Any) -> None: + """Route an incoming channel event to its handler. + + Errors are logged with full traceback but never re-raised — the response to the + platform has already been sent, and an unhandled exception here would just become + a noisy stack trace in the API process logs. + """ + try: + async with get_db_context() as db: + await ChannelMessageRouter().route(incoming, db) + except Exception: + logger.exception("channel_event_processing_failed") diff --git a/backend/app/worker/broker.py b/backend/app/worker/broker.py deleted file mode 100644 index 2afcde7..0000000 --- a/backend/app/worker/broker.py +++ /dev/null @@ -1,14 +0,0 @@ -"""Taskiq broker — shared across all task modules.""" - -from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend - -from app.core.config import settings - -# Create Taskiq broker with Redis -broker = ListQueueBroker( - url=settings.TASKIQ_BROKER_URL, -).with_result_backend( - RedisAsyncResultBackend( - redis_url=settings.TASKIQ_RESULT_BACKEND, - ) -) diff --git a/backend/app/worker/taskiq_app.py b/backend/app/worker/taskiq_app.py index b948014..1a525cf 100644 --- a/backend/app/worker/taskiq_app.py +++ b/backend/app/worker/taskiq_app.py @@ -2,13 +2,20 @@ from taskiq import TaskiqScheduler from taskiq.schedule_sources import LabelScheduleSource +from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend -from app.worker.broker import broker -import app.worker.tasks # noqa: F401 — register @broker.task decorated functions -import app.worker.tasks.schedules # noqa: F401 — register scheduled tasks +from app.core.config import settings + +# Create Taskiq broker with Redis +broker = ListQueueBroker( + url=settings.TASKIQ_BROKER_URL, +).with_result_backend( + RedisAsyncResultBackend( + redis_url=settings.TASKIQ_RESULT_BACKEND, + ) +) # Create scheduler for periodic tasks -# LabelScheduleSource auto-discovers @broker.task(schedule=[...]) decorated functions scheduler = TaskiqScheduler( broker=broker, sources=[LabelScheduleSource(broker)], @@ -26,3 +33,6 @@ async def startup() -> None: async def shutdown() -> None: """Cleanup on shutdown.""" pass + + +import app.worker.tasks.schedules # noqa: F401 diff --git a/backend/app/worker/tasks/rag_tasks.py b/backend/app/worker/tasks/rag_tasks.py index 034a32f..c05908b 100644 --- a/backend/app/worker/tasks/rag_tasks.py +++ b/backend/app/worker/tasks/rag_tasks.py @@ -6,7 +6,7 @@ import tempfile from pathlib import Path from typing import Any -from app.worker.broker import broker +from app.worker.taskiq_app import broker logger = logging.getLogger(__name__) diff --git a/backend/app/worker/tasks/schedules.py b/backend/app/worker/tasks/schedules.py index 5d429a1..aad0a7c 100644 --- a/backend/app/worker/tasks/schedules.py +++ b/backend/app/worker/tasks/schedules.py @@ -1,6 +1,6 @@ """Taskiq scheduled tasks (cron-like).""" -from app.worker.broker import broker +from app.worker.taskiq_app import broker from app.worker.tasks.rag_tasks import check_scheduled_syncs