Feat/queues and handlers

app/handlers/email.py

@@ -0,0 +1,94 @@
+"""
+Payload shape:
+    {
+        "to":      "user@example.com",    # required
+        "subject": "Welcome to Flint",    # required
+        "body":    "Plain text body",     # optional
+        "html":    "<p>HTML body</p>"     # optional
+    }
+
+Result shape (on success):
+    {
+        "to": "user@example.com",
+        "subject": "Welcome to Flint",
+        "delivered": true
+    }
+"""
+
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import Any
+
+import aiosmtplib
+from pydantic import BaseModel
+
+from app.core.logger import get_logger
+from app.handlers.base import BaseHandler
+from app.handlers.utils import parse_payload
+
+logger = get_logger(__name__)
+
+
+class EmailPayload(BaseModel):
+    to: str
+    subject: str
+    body: str | None = ""
+    html: str | None = None
+
+
+class EmailHandler(BaseHandler):
+    async def execute(self, payload: dict[str, Any]) -> dict[str, Any]:
+        from app.core.config import settings
+
+        p = parse_payload(payload, EmailPayload)
+
+        to = p.to
+        subject = p.subject
+
+        body = p.body or ""
+        html = p.html
+
+        msg = MIMEMultipart("alternative")
+        msg["Subject"] = subject
+        msg["From"] = settings.SMTP_FROM
+        msg["To"] = to
+
+        msg.attach(MIMEText(body, "plain", "utf-8"))
+
+        if html:
+            msg.attach(MIMEText(html, "html", "utf-8"))
+
+        logger.info(
+            "email_attempt",
+            to=to,
+            subject=subject,
+            has_html=bool(html),
+            smtp_host=settings.SMTP_HOST,
+            smtp_port=settings.SMTP_PORT,
+        )
+
+        try:
+            await aiosmtplib.send(
+                msg,
+                hostname=settings.SMTP_HOST,
+                port=settings.SMTP_PORT,
+                use_tls=False,
+                start_tls=False,
+            )
+        except aiosmtplib.SMTPException as exc:
+            raise Exception(f"SMTP delivery failed to '{to}': {str(exc)}") from exc
+        except OSError as exc:
+            raise Exception(
+                f"Cannot connect to SMTP server "
+                f"{settings.SMTP_HOST}:{settings.SMTP_PORT} — {str(exc)}"
+            ) from exc
+
+        result = {
+            "to": to,
+            "subject": subject,
+            "delivered": True,
+        }
+
+        logger.info("email_success", to=to, subject=subject)
+
+        return result

app/handlers/log_processor.py

@@ -0,0 +1,135 @@
+"""
+Payload shape:
+    {
+        "lines": [
+            "2026-06-09T10:00:00Z INFO  Request received",
+            "2026-06-09T10:00:01Z ERROR Database timeout",
+            "2026-06-09T10:00:02Z ERROR Database timeout"
+        ],
+        "source": "app-server-01"   # optional label
+    }
+
+Result shape (on success):
+    {
+        "source": "app-server-01",
+        "total_lines": 3,
+        "parsed_lines": 3,
+        "parse_errors": 0,
+        "level_counts": {"INFO": 1, "ERROR": 2},
+        "top_errors": [
+            {"message": "Database timeout", "count": 2}
+        ],
+        "has_critical": false,
+        "error_rate_pct": 66.67
+    }
+"""
+
+from collections import Counter
+from typing import Any
+
+from pydantic import BaseModel, field_validator
+
+from app.core.logger import get_logger
+from app.handlers.base import BaseHandler
+from app.handlers.utils import parse_payload
+
+logger = get_logger(__name__)
+
+VALID_LEVELS = frozenset({"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"})
+TOP_ERRORS_LIMIT = 5
+
+
+class LogPayload(BaseModel):
+    lines: list[str]
+    source: str = "unknown"
+
+    @field_validator("lines")
+    @classmethod
+    def must_not_be_empty(cls, v):
+        if not v:
+            raise ValueError("'lines' must be a non-empty list.")
+        return v
+
+
+class LogProcessorHandler(BaseHandler):
+    async def execute(self, payload: dict[str, Any]) -> dict[str, Any]:
+        p = parse_payload(payload, LogPayload)
+        lines: list = p.lines
+        source: str = p.source
+
+        parsed = []
+        parse_errors = []
+
+        for raw_line in lines:
+            line = str(raw_line).strip()
+            if not line:
+                continue
+
+            parts = line.split(None, 2)
+            if len(parts) < 2:
+                parse_errors.append(line)
+                continue
+
+            # Parts: [timestamp, level] or [timestamp, level, message]
+            timestamp = parts[0]
+            level = parts[1].upper()
+            message = parts[2] if len(parts) == 3 else ""
+
+            if level not in VALID_LEVELS:
+                parse_errors.append(line)
+                continue
+
+            parsed.append(
+                {
+                    "timestamp": timestamp,
+                    "level": level,
+                    "message": message,
+                }
+            )
+
+        if not parsed:
+            raise ValueError(
+                f"No parseable log lines found in payload. "
+                f"Expected format: '<timestamp> <LEVEL> <message>' "
+                f"where LEVEL is one of: {', '.join(sorted(VALID_LEVELS))}. "
+                f"{len(parse_errors)} line(s) could not be parsed."
+            )
+
+        # Aggregate
+        level_counts = Counter(entry["level"] for entry in parsed)
+
+        error_messages = [
+            entry["message"]
+            for entry in parsed
+            if entry["level"] in ("ERROR", "CRITICAL")
+        ]
+        top_errors = [
+            {"message": msg, "count": count}
+            for msg, count in Counter(error_messages).most_common(TOP_ERRORS_LIMIT)
+        ]
+
+        error_count = level_counts.get("ERROR", 0) + level_counts.get("CRITICAL", 0)
+        error_rate = round((error_count / len(parsed)) * 100, 2) if parsed else 0.0
+
+        summary = {
+            "source": source,
+            "total_lines": len(lines),
+            "parsed_lines": len(parsed),
+            "parse_errors": len(parse_errors),
+            "level_counts": dict(level_counts),
+            "top_errors": top_errors,
+            "has_critical": level_counts.get("CRITICAL", 0) > 0,
+            "error_rate_pct": error_rate,
+        }
+
+        logger.info(
+            "log_processing_complete",
+            source=source,
+            total_lines=len(lines),
+            parsed_lines=len(parsed),
+            parse_errors=len(parse_errors),
+            has_critical=summary["has_critical"],
+            error_rate_pct=error_rate,
+        )
+
+        return summary

app/handlers/utils.py

@@ -0,0 +1,22 @@
+from typing import Any, TypeVar
+
+from pydantic import BaseModel, ValidationError
+
+T = TypeVar("T", bound=BaseModel)
+
+
+def parse_payload[T](
+    payload: dict[str, Any],
+    schema: type[T],
+) -> T:
+    """
+    Validate *payload* against a Pydantic BaseModel subclass.
+    """
+    try:
+        return schema(**payload)
+    except ValidationError as exc:
+        lines = [f"Payload validation failed for {schema.__name__}:"]
+        for err in exc.errors():
+            field = " → ".join(str(p) for p in err["loc"]) or "(root)"
+            lines.append(f"  • {field}: {err['msg']}  [input={err.get('input')!r}]")
+        raise ValueError("\n".join(lines)) from exc

app/handlers/webhook.py

@@ -0,0 +1,112 @@
+"""
+Payload shape:
+    {
+        "url":     "https://webhook.site/abc",   # required
+        "method":  "POST",                        # optional, default POST
+        "headers": {"X-Custom": "value"},         # optional
+        "body":    {"event": "user.created"}      # optional
+    }
+
+Result shape (on success):
+    {
+        "status_code": 200,
+        "response_body": "...",
+        "url": "https://...",
+        "method": "POST"
+    }
+"""
+
+from typing import Any
+
+import httpx
+from pydantic import BaseModel, field_validator
+
+from app.core.logger import get_logger
+from app.handlers.base import BaseHandler
+from app.handlers.utils import parse_payload
+
+logger = get_logger(__name__)
+
+TIMEOUT_SECONDS = 10
+MAX_REDIRECTS = 3
+ALLOWED_METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE"}
+
+
+class WebhookPayload(BaseModel):
+    url: str
+    method: str | None = "POST"
+    headers: dict[str, str] = {}
+    body: dict[str, Any] = {}
+
+    @field_validator("method")
+    @classmethod
+    def validate_method(cls, v):
+        if v is None:
+            return "POST"
+
+        if v.upper() not in ALLOWED_METHODS:
+            raise ValueError(
+                f"Invalid HTTP method '{v}'. "
+                f"Must be one of: {', '.join(sorted(ALLOWED_METHODS))}."
+            )
+        return v.upper()
+
+
+class WebhookHandler(BaseHandler):
+    async def execute(self, payload: dict[str, Any]) -> dict[str, Any]:
+        p = parse_payload(payload, WebhookPayload)
+        url = p.url
+        method = p.method or "POST"
+        headers = p.headers or {}
+        body = p.body or {}
+
+        logger.info(
+            "webhook_attempt",
+            url=url,
+            method=method,
+            body_keys=list(body.keys()),
+        )
+
+        try:
+            async with httpx.AsyncClient(
+                timeout=httpx.Timeout(TIMEOUT_SECONDS),
+                follow_redirects=True,
+                max_redirects=MAX_REDIRECTS,
+            ) as client:
+                response = await client.request(
+                    method=method,
+                    url=url,
+                    json=body if body else None,
+                    headers=headers,
+                )
+        except httpx.TimeoutException as exc:
+            raise Exception(
+                f"Webhook timed out after {TIMEOUT_SECONDS}s: {url}. Detail: {str(exc)}"
+            ) from exc
+        except httpx.TooManyRedirects as exc:
+            raise Exception(
+                f"Webhook exceeded max redirects ({MAX_REDIRECTS}): {url}."
+            ) from exc
+        except httpx.RequestError as exc:
+            raise Exception(f"Webhook connection error to {url}: {str(exc)}") from exc
+
+        if response.status_code >= 400:
+            raise Exception(
+                f"Webhook failed: HTTP {response.status_code} from {url}. "
+                f"Body: {response.text[:500]}"
+            )
+
+        result = {
+            "status_code": response.status_code,
+            "response_body": response.text[:1000],
+            "url": url,
+            "method": method,
+        }
+
+        logger.info(
+            "webhook_success",
+            url=url,
+            status_code=response.status_code,
+        )
+
+        return result