Added webhooks subapp for instagram webhooks and celery app and task to pass comment text to an LLM for further processing

README.md

@@ -16,3 +16,43 @@ Python project initialized with uv.
 4. Run tests:
 
    uv run pytest
+
+## Webhooks
+
+Instagram comments webhook endpoint:
+
+- POST /webhooks/instagram/comments
+
+The route validates incoming payloads and enqueues a Celery task.
+
+## LLM task configuration
+
+The Celery task that handles webhook payload forwarding reads these environment variables:
+
+- LLM_ENDPOINT_URL (required): Full HTTP(S) URL for the downstream LLM endpoint.
+- LLM_ENDPOINT_API_KEY (optional): Bearer token sent as Authorization header.
+- LLM_ENDPOINT_TIMEOUT_SECONDS (optional): Request timeout in seconds. Defaults to 10.
+
+Retry behavior is built into the task:
+
+- Retries transient failures (network/timeouts and HTTP 429/500/502/503/504).
+- Exponential backoff enabled.
+- Retry delay is jittered.
+- Maximum retries is 7.
+
+## Running a Celery worker
+
+The Celery entrypoint is in src/comment_automation/celery_app.py.
+
+It reads these environment variables:
+
+- CELERY_BROKER_URL (optional): Message broker URL. Defaults to redis://localhost:6379/0.
+- CELERY_RESULT_BACKEND (optional): Result backend URL. Defaults to redis://localhost:6379/0.
+
+With uv:
+
+uv run celery -A comment_automation.celery_app:celery_app worker --loglevel=info
+
+Without uv (venv already created):
+
+./.venv/bin/celery -A comment_automation.celery_app:celery_app worker --loglevel=info

pyproject.toml

@@ -5,15 +5,23 @@ description = "Comment automation service"
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
+  "celery>=5.4.0",
   "fastapi[standard]>=0.116.0",
+  "httpx>=0.28.1",
   "pydantic>=2.11.0",
   "uplink>=0.9.7",
 ]
 
 [dependency-groups]
 dev = [
-  "pytest>=8.3.0",
+  "bandit>=1.7.9",
+  "black>=24.10.0",
+  "isort>=5.13.2",
   "pre-commit>=4.2.0",
+  "pytest>=8.3.0",
+  "ruff>=0.8.0",
+  "ty>=0.0.1a13",
+  "uv>=0.8.15",
 ]
 
 [build-system]
@@ -26,3 +34,6 @@ packages = ["src/comment_automation"]
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 addopts = "-q"
+
+[tool.isort]
+profile = "black"

src/comment_automation/celery_app.py

@@ -0,0 +1,18 @@
+import os
+
+from celery import Celery
+
+celery_app = Celery(
+    "comment_automation",
+    broker=os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0"),
+    backend=os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/0"),
+    include=["comment_automation.webhooks.tasks"],
+)
+
+celery_app.conf.update(
+    task_serializer="json",
+    accept_content=["json"],
+    result_serializer="json",
+    timezone="UTC",
+    enable_utc=True,
+)

src/comment_automation/main.py

@@ -1,8 +1,10 @@
 from fastapi import FastAPI
 from pydantic import BaseModel
 
+from comment_automation.webhooks.app import webhooks_app
 
 app = FastAPI(title="CommentAutomation")
+app.mount("/webhooks", webhooks_app)
 
 
 class HealthResponse(BaseModel):

src/comment_automation/webhooks/__init__.py

@@ -0,0 +1 @@
+"""Webhooks subapp."""

src/comment_automation/webhooks/app.py

@@ -0,0 +1,6 @@
+from fastapi import FastAPI
+
+from comment_automation.webhooks.router import router as webhooks_router
+
+webhooks_app = FastAPI(title="CommentAutomation Webhooks")
+webhooks_app.include_router(webhooks_router)

src/comment_automation/webhooks/models.py

@@ -0,0 +1,38 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class InstagramCommentWebhookValue(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    from_id: str | None = None
+    media_id: str | None = None
+    comment_id: str | None = None
+    text: str | None = None
+    timestamp: int | None = None
+
+
+class InstagramCommentWebhookChange(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    field: str
+    value: InstagramCommentWebhookValue
+
+
+class InstagramCommentWebhookEntry(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    id: str
+    time: int
+    changes: list[InstagramCommentWebhookChange]
+
+
+class InstagramCommentWebhookPayload(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    object: str
+    entry: list[InstagramCommentWebhookEntry]
+
+
+class InstagramCommentWebhookAccepted(BaseModel):
+    status: str
+    task_id: str | None = None

src/comment_automation/webhooks/router.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter, status
+
+from comment_automation.webhooks.models import (
+    InstagramCommentWebhookAccepted,
+    InstagramCommentWebhookPayload,
+)
+from comment_automation.webhooks.services import handle_instagram_comment_webhook
+
+router = APIRouter(tags=["webhooks"])
+
+
+@router.post(
+    "/instagram/comments",
+    response_model=InstagramCommentWebhookAccepted,
+    status_code=status.HTTP_202_ACCEPTED,
+)
+def instagram_comments_webhook(
+    payload: InstagramCommentWebhookPayload,
+) -> InstagramCommentWebhookAccepted:
+    return handle_instagram_comment_webhook(payload)

src/comment_automation/webhooks/services.py

@@ -0,0 +1,15 @@
+from comment_automation.webhooks.models import (
+    InstagramCommentWebhookAccepted,
+    InstagramCommentWebhookPayload,
+)
+from comment_automation.webhooks.tasks import send_instagram_comment_to_llm
+
+
+def handle_instagram_comment_webhook(
+    payload: InstagramCommentWebhookPayload,
+) -> InstagramCommentWebhookAccepted:
+    task_result = send_instagram_comment_to_llm.delay(payload.model_dump())
+
+    return InstagramCommentWebhookAccepted(
+        status="accepted", task_id=str(task_result.id)
+    )

src/comment_automation/webhooks/tasks.py

@@ -0,0 +1,64 @@
+import os
+from typing import Any
+
+import httpx
+from celery import shared_task
+
+
+class LLMEndpointTemporaryError(Exception):
+    """Raised when the LLM endpoint fails with a transient/retryable error."""
+
+
+class LLMEndpointConfigurationError(Exception):
+    """Raised when required LLM endpoint configuration is missing."""
+
+
+@shared_task(
+    bind=True,
+    name="comment_automation.tasks.send_instagram_comment_to_llm",
+    autoretry_for=(LLMEndpointTemporaryError,),
+    retry_backoff=True,
+    retry_backoff_max=300,
+    retry_jitter=True,
+    retry_kwargs={"max_retries": 7},
+)
+def send_instagram_comment_to_llm(self, payload: dict[str, Any]) -> None:
+    """Forward the Instagram webhook payload to the LLM endpoint.
+
+    Raise LLMEndpointTemporaryError for transient failures (timeouts/5xx/network)
+    so Celery re-schedules this task with exponential backoff.
+    """
+    _forward_payload_to_llm(payload)
+
+
+def _forward_payload_to_llm(payload: dict[str, Any]) -> None:
+    """Send payload to LLM endpoint, mapping transient failures to retries."""
+    endpoint_url = os.getenv("LLM_ENDPOINT_URL")
+    if not endpoint_url:
+        raise LLMEndpointConfigurationError("LLM_ENDPOINT_URL is not configured")
+
+    api_key = os.getenv("LLM_ENDPOINT_API_KEY")
+    timeout = float(os.getenv("LLM_ENDPOINT_TIMEOUT_SECONDS", "10"))
+
+    headers = {"Content-Type": "application/json"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+
+    try:
+        response = httpx.post(
+            endpoint_url,
+            json=payload,
+            headers=headers,
+            timeout=timeout,
+        )
+    except (httpx.TimeoutException, httpx.NetworkError) as exc:
+        raise LLMEndpointTemporaryError(
+            "Temporary network issue while calling LLM endpoint"
+        ) from exc
+
+    if response.status_code in {429, 500, 502, 503, 504}:
+        raise LLMEndpointTemporaryError(
+            f"Retryable LLM endpoint response status: {response.status_code}"
+        )
+
+    response.raise_for_status()

tests/test_health.py

@@ -2,7 +2,6 @@ from fastapi.testclient import TestClient
 
 from comment_automation.main import app
 
-
 client = TestClient(app)
 
 

tests/test_instagram_webhooks.py

@@ -0,0 +1,49 @@
+from fastapi.testclient import TestClient
+
+from comment_automation.main import app
+
+client = TestClient(app)
+
+
+def test_instagram_comment_webhook_enqueues_celery_task(monkeypatch) -> None:
+    captured_payload: dict = {}
+
+    class FakeAsyncResult:
+        id = "task-123"
+
+    def fake_delay(payload: dict) -> FakeAsyncResult:
+        captured_payload.update(payload)
+        return FakeAsyncResult()
+
+    monkeypatch.setattr(
+        "comment_automation.webhooks.services.send_instagram_comment_to_llm.delay",
+        fake_delay,
+    )
+
+    body = {
+        "object": "instagram",
+        "entry": [
+            {
+                "id": "17841400000000000",
+                "time": 1711799299,
+                "changes": [
+                    {
+                        "field": "comments",
+                        "value": {
+                            "from_id": "123456",
+                            "media_id": "17895695668004550",
+                            "comment_id": "17900000000000000",
+                            "text": "Nice post!",
+                            "timestamp": 1711799299,
+                        },
+                    }
+                ],
+            }
+        ],
+    }
+
+    response = client.post("/webhooks/instagram/comments", json=body)
+
+    assert response.status_code == 202
+    assert response.json() == {"status": "accepted", "task_id": "task-123"}
+    assert captured_payload == body

tests/test_webhook_tasks.py

@@ -0,0 +1,52 @@
+import pytest
+
+from comment_automation.webhooks.tasks import (
+    LLMEndpointConfigurationError,
+    LLMEndpointTemporaryError,
+    _forward_payload_to_llm,
+)
+
+
+def test_forward_payload_raises_when_endpoint_missing(monkeypatch) -> None:
+    monkeypatch.delenv("LLM_ENDPOINT_URL", raising=False)
+
+    with pytest.raises(LLMEndpointConfigurationError):
+        _forward_payload_to_llm({"hello": "world"})
+
+
+def test_forward_payload_retries_on_retryable_status(monkeypatch) -> None:
+    class FakeResponse:
+        status_code = 503
+
+        def raise_for_status(self) -> None:
+            return None
+
+    def fake_post(*args, **kwargs):
+        return FakeResponse()
+
+    monkeypatch.setenv("LLM_ENDPOINT_URL", "https://example.org/llm")
+    monkeypatch.setattr("comment_automation.webhooks.tasks.httpx.post", fake_post)
+
+    with pytest.raises(LLMEndpointTemporaryError):
+        _forward_payload_to_llm({"hello": "world"})
+
+
+def test_forward_payload_success(monkeypatch) -> None:
+    called = {"value": False}
+
+    class FakeResponse:
+        status_code = 200
+
+        def raise_for_status(self) -> None:
+            return None
+
+    def fake_post(*args, **kwargs):
+        called["value"] = True
+        return FakeResponse()
+
+    monkeypatch.setenv("LLM_ENDPOINT_URL", "https://example.org/llm")
+    monkeypatch.setattr("comment_automation.webhooks.tasks.httpx.post", fake_post)
+
+    _forward_payload_to_llm({"hello": "world"})
+
+    assert called["value"] is True