diff --git a/app/handlers/email.py b/app/handlers/email.py new file mode 100644 index 0000000..d50d414 --- /dev/null +++ b/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": "

HTML body

" # 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 diff --git a/app/handlers/log_processor.py b/app/handlers/log_processor.py new file mode 100644 index 0000000..a134b61 --- /dev/null +++ b/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: ' ' " + 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 diff --git a/app/handlers/utils.py b/app/handlers/utils.py new file mode 100644 index 0000000..cf8c964 --- /dev/null +++ b/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 diff --git a/app/handlers/webhook.py b/app/handlers/webhook.py new file mode 100644 index 0000000..967998c --- /dev/null +++ b/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 diff --git a/app/queues/base.py b/app/queues/base.py new file mode 100644 index 0000000..3516739 --- /dev/null +++ b/app/queues/base.py @@ -0,0 +1,57 @@ +from abc import ABC, abstractmethod + + +class BaseQueue(ABC): + @abstractmethod + async def push( + self, + job_id: str, + effective_priority: float, + scheduled_at: float, + created_at: float, + ) -> None: + """ + Add a job to the queue. + """ + ... + + @abstractmethod + async def pop(self) -> str | None: + """ + Remove and return the most urgent job_id from the queue. + Returns None if the queue is empty. + """ + ... + + @abstractmethod + async def remove(self, job_id: str) -> None: + """ + Remove a specific job from the queue without processing it. + Used when a job is cancelled while still pending in the queue. + No-op if the job is not in the queue. + """ + ... + + @abstractmethod + async def size(self) -> int: + """ + Return the number of valid (non-removed) jobs in the queue. + """ + ... + + async def update_priority( + self, + job_id: str, + new_priority: float, + scheduled_at: float, + created_at: float, + ) -> None: + """ + Re-score a job with a new effective_priority. + Called by the aging process when it decrements a job's priority. + + Default implementation: remove then re-push. + HeapQueue overrides this with lazy deletion for efficiency. + """ + await self.remove(job_id) + await self.push(job_id, new_priority, scheduled_at, created_at) diff --git a/app/queues/heapq.py b/app/queues/heapq.py new file mode 100644 index 0000000..0fabedd --- /dev/null +++ b/app/queues/heapq.py @@ -0,0 +1,148 @@ +import asyncio +import heapq +from dataclasses import dataclass, field + +from app.queues.base import BaseQueue + +_REMOVED = "__removed__" + + +@dataclass(order=True) +class HeapEntry: + """ + Sort order (left-to-right tuple comparison): + 1. effective_priority — lower value = more urgent + 2. scheduled_at — earlier time = more urgent + 3. created_at — older job = more urgent + 4. job_id — deterministic tiebreaker (UUID string sort) + """ + + effective_priority: float + scheduled_at: float + created_at: float + job_id: str = field(compare=True) + + +class HeapQueue(BaseQueue): + """ + In-memory min-heap priority queue. + + The heap is local to each worker process. On worker startup it is + populated from the Redis sorted set (flint:queue) which acts as the + persistent backing store. + """ + + def __init__(self) -> None: + self._heap: list[HeapEntry] = [] + self._entry_finder: dict[str, HeapEntry] = {} + self._lock = asyncio.Lock() + + async def push( + self, + job_id: str, + effective_priority: float, + scheduled_at: float, + created_at: float, + ) -> None: + """ + Push a job onto the heap. + + If the job_id already exists (e.g. being re-scored by aging), + the old entry is lazily marked REMOVED and a new one is pushed. + This avoids an O(n) heap rebuild. + """ + async with self._lock: + if job_id in self._entry_finder: + self._mark_removed(job_id) + + entry = HeapEntry( + effective_priority=effective_priority, + scheduled_at=scheduled_at, + created_at=created_at, + job_id=job_id, + ) + self._entry_finder[job_id] = entry + heapq.heappush(self._heap, entry) + + async def pop(self) -> str | None: + """ + Pop and return the most urgent job_id. + + Skips entries that have been marked REMOVED (lazy deletion). + Returns None if no valid entries remain. + """ + async with self._lock: + while self._heap: + entry = heapq.heappop(self._heap) + if entry.job_id != _REMOVED: + self._entry_finder.pop(entry.job_id, None) + return entry.job_id + return None + + async def remove(self, job_id: str) -> None: + """ + Mark a job as removed. It will be silently skipped on the next pop(). + No-op if the job is not in the queue. + """ + async with self._lock: + if job_id in self._entry_finder: + self._mark_removed(job_id) + + async def update_priority( + self, + job_id: str, + new_priority: float, + scheduled_at: float, + created_at: float, + ) -> None: + """ + Re-score an existing job with a new effective_priority. + + Called by the aging process every AGING_INTERVAL seconds. + Uses lazy deletion: marks the old entry removed, pushes a new one. + O(log n) — no heap rebuild required. + """ + async with self._lock: + if job_id in self._entry_finder: + self._mark_removed(job_id) + + entry = HeapEntry( + effective_priority=new_priority, + scheduled_at=scheduled_at, + created_at=created_at, + job_id=job_id, + ) + self._entry_finder[job_id] = entry + heapq.heappush(self._heap, entry) + + async def size(self) -> int: + """ + Returns the number of valid (non-removed) jobs in the queue. + O(1) — reads from _entry_finder, not the heap list. + """ + async with self._lock: + return len(self._entry_finder) + + async def peek(self) -> str | None: + """ + Return the most urgent job_id without removing it. + Skips REMOVED entries but does not pop them from the heap. + """ + async with self._lock: + for entry in self._heap: + if entry.job_id != _REMOVED: + return entry.job_id + return None + + async def contains(self, job_id: str) -> bool: + """Check if a job_id is currently in the queue.""" + async with self._lock: + return job_id in self._entry_finder + + def _mark_removed(self, job_id: str) -> None: + """ + Mark an entry as removed in-place. + The entry remains in the heap list but will be skipped on pop(). + """ + entry = self._entry_finder.pop(job_id) + entry.job_id = _REMOVED diff --git a/app/queues/timing_wheel.py b/app/queues/timing_wheel.py new file mode 100644 index 0000000..b21c882 --- /dev/null +++ b/app/queues/timing_wheel.py @@ -0,0 +1,143 @@ +import asyncio +import time +from collections import defaultdict + +from app.queues.base import BaseQueue + +WHEEL_SIZE = 3600 + + +class TimingWheel(BaseQueue): + def __init__(self) -> None: + self._wheel: list[list[tuple[float, str]]] = [[] for _ in range(WHEEL_SIZE)] + self._overflow: dict[float, list[tuple[float, str]]] = defaultdict(list) + self._current_slot: int = 0 + self._start_time: float = time.monotonic() + self._lock = asyncio.Lock() + + def _seconds_until(self, scheduled_at_unix: float) -> float: + """ + Return how many seconds until scheduled_at_unix from now. + Negative or zero means the job is due immediately (slot 0 offset). + """ + delta = scheduled_at_unix - time.time() + return max(0.0, delta) + + def _slot_for(self, seconds_from_now: float) -> int | None: + """ + Map a delay in seconds to a wheel slot index. + Returns None if the delay exceeds the wheel's range (goes to overflow). + """ + if seconds_from_now >= WHEEL_SIZE: + return None + offset = int(seconds_from_now) + return (self._current_slot + offset) % WHEEL_SIZE + + def _insert_into_slot( + self, + slot: int, + priority: float, + job_id: str, + ) -> None: + """ + Insert a job into a wheel slot, keeping the slot sorted by priority. + """ + self._wheel[slot].append((priority, job_id)) + # Keep ascending sort so index 0 = highest priority (lowest score) + self._wheel[slot].sort(key=lambda x: x[0]) + + def _drain_overflow(self) -> None: + """ + Move overflow jobs that are now within the wheel's range into slots. + """ + now = time.time() + to_insert = [ts for ts in self._overflow if ts <= now + WHEEL_SIZE] + for ts in to_insert: + entries = self._overflow.pop(ts) + seconds = max(0.0, ts - now) + slot = self._slot_for(seconds) + if slot is not None: + for priority, job_id in entries: + self._insert_into_slot(slot, priority, job_id) + + async def push( + self, + job_id: str, + effective_priority: float, + scheduled_at: float, + created_at: float, + ) -> None: + """ + Place a job in the appropriate wheel slot or overflow. + created_at is accepted for interface compatibility but not used — + the timing wheel sorts within a slot by priority only. + """ + async with self._lock: + seconds = self._seconds_until(scheduled_at) + slot = self._slot_for(seconds) + if slot is None: + self._overflow[scheduled_at].append((effective_priority, job_id)) + else: + self._insert_into_slot(slot, effective_priority, job_id) + + async def pop(self) -> str | None: + """ + Advance one tick and return the highest-priority job due in this slot. + Also drains overflow jobs that have come within the wheel's range. + """ + due = await self.tick() + return due[0] if due else None + + async def tick(self) -> list[str]: + """ + Advance the wheel pointer by one slot. + + Returns the list of job_ids due in the current slot (ordered by + effective_priority within the slot, most urgent first). + + Also drains overflow jobs that are now within wheel range. + """ + async with self._lock: + due_entries = self._wheel[self._current_slot] + due_job_ids = [job_id for _, job_id in due_entries] + self._wheel[self._current_slot] = [] + + self._current_slot = (self._current_slot + 1) % WHEEL_SIZE + + self._drain_overflow() + + return due_job_ids + + async def remove(self, job_id: str) -> None: + """ + Remove a job from the wheel by scanning all slots and overflow. + """ + async with self._lock: + for slot in self._wheel: + slot[:] = [(p, jid) for p, jid in slot if jid != job_id] + for ts in list(self._overflow.keys()): + self._overflow[ts] = [ + (p, jid) for p, jid in self._overflow[ts] if jid != job_id + ] + if not self._overflow[ts]: + del self._overflow[ts] + + async def size(self) -> int: + """ + Return total number of jobs across all wheel slots and overflow. + """ + async with self._lock: + wheel_count = sum(len(slot) for slot in self._wheel) + overflow_count = sum(len(v) for v in self._overflow.values()) + return wheel_count + overflow_count + + async def contains(self, job_id: str) -> bool: + """Check whether a job_id exists anywhere in the wheel or overflow.""" + async with self._lock: + for slot in self._wheel: + if any(jid == job_id for _, jid in slot): + return True + for entries in self._overflow.values(): + if any(jid == job_id for _, jid in entries): + return True + return False diff --git a/app/schemas/job.py b/app/schemas/job.py new file mode 100644 index 0000000..3e6b682 --- /dev/null +++ b/app/schemas/job.py @@ -0,0 +1,179 @@ +import re +from datetime import datetime +from typing import Any +from uuid import UUID + +from pydantic import BaseModel, Field, field_validator + +from app.models.enum import JobType +from app.models.job import JobPriority, JobStatus + +INTERVAL_MULTIPLIERS: dict[str, int] = { + "s": 1, + "m": 60, + "h": 3600, + "d": 86400, + "mo": 2592000, + "y": 31536000, +} + +INTERVAL_PATTERN = re.compile(r"^(\d+)(s|mo|m|h|d|y)$") + + +def parse_interval(interval_str: str) -> int: + """ + Parse a flexible interval string into total seconds. + + Accepted formats: + "30s" → 30 + "5m" → 300 + "2h" → 7200 + "1d" → 86400 + "1mo" → 2592000 + "1y" → 31536000 + """ + match = INTERVAL_PATTERN.match(interval_str.strip().lower()) + if not match: + raise ValueError( + f"Invalid interval '{interval_str}'. " + "Use where unit is one of: s, m, h, d, mo, y. " + "Examples: '30s', '5m', '2h', '1d', '1mo', '1y'." + ) + value = int(match.group(1)) + unit = match.group(2) + if value <= 0: + raise ValueError("Interval value must be greater than 0.") + return value * INTERVAL_MULTIPLIERS[unit] + + +def format_interval_seconds(seconds: int) -> str: + """ + Convert interval_seconds back to a human-readable string. + """ + if seconds < 60: + return f"{seconds}s" + if seconds < 3600: + return f"{seconds // 60}m" + if seconds < 86400: + return f"{seconds // 3600}h" + if seconds < 2592000: + return f"{seconds // 86400}d" + if seconds < 31536000: + return f"{seconds // 2592000}mo" + return f"{seconds // 31536000}y" + + +class JobCreate(BaseModel): + type: JobType + payload: dict[str, Any] = Field(default_factory=dict) + priority: JobPriority = JobPriority.MEDIUM + scheduled_at: datetime | None = Field( + default=None, + description="ISO 8601 datetime. Defaults to now if not provided.", + ) + interval: str | None = Field( + default=None, + description="Recurring interval e.g. '5m', '1h'. Null means non-recurring.", + ) + dependency_ids: list[UUID] | None = Field( + default=None, + description="List of job IDs that must complete before this job runs.", + ) + max_retries: int = Field( + default=3, + ge=1, + le=10, + description="Maximum retry attempts before the job moves to DLQ.", + ) + + @field_validator("interval") + @classmethod + def validate_interval(cls, v: str | None) -> str | None: + if v is not None: + parse_interval(v) + return v + + @field_validator("dependency_ids") + @classmethod + def validate_dependency_ids(cls, v: list[UUID] | None) -> list[UUID] | None: + if v is not None and len(v) != len(set(v)): + raise ValueError("dependency_ids must not contain duplicates.") + return v + + model_config = { + "json_schema_extra": { + "examples": [ + { + "type": "webhook_delivery", + "payload": { + "url": "https://webhook.site/abc", + "body": {"event": "user.created"}, + }, + "priority": 1, + "scheduled_at": "2026-06-10T10:00:00Z", + "interval": "1h", + "dependency_ids": [], + "max_retries": 3, + } + ] + } + } + + +class JobLogResponse(BaseModel): + id: UUID + job_id: UUID + event: str + message: str + metadata_: dict[str, Any] | None = Field(None, alias="metadata") + created_at: datetime + + model_config = { + "from_attributes": True, + "populate_by_name": True, + } + + +class JobResponse(BaseModel): + id: UUID + type: str + payload: dict[str, Any] + priority: int + status: str + scheduled_at: datetime + interval_seconds: int | None + retry_count: int + max_retries: int + last_error: str | None + effective_priority: float + cancellation_requested: bool + worker_id: str | None + started_at: datetime | None + completed_at: datetime | None + is_dlq: bool + created_at: datetime + updated_at: datetime + + # Populated on detail endpoint only + dependencies: list[UUID] | None = None + logs: list[JobLogResponse] | None = None + + model_config = {"from_attributes": True} + + +class JobListResponse(BaseModel): + jobs: list[JobResponse] + + +class JobFilterParams(BaseModel): + page: int = Field(default=1, ge=1, description="Page number (1-indexed).") + limit: int = Field(default=20, ge=1, le=100, description="Items per page.") + status: JobStatus | None = Field(default=None, description="Filter by status.") + type: JobType | None = Field(default=None, description="Filter by job type.") + priority: JobPriority | None = Field( + default=None, description="Filter by priority." + ) + search: str | None = Field( + default=None, + description="Search across job ID (partial) and type.", + ) diff --git a/app/services/alert.py b/app/services/alert.py new file mode 100644 index 0000000..9a4759c --- /dev/null +++ b/app/services/alert.py @@ -0,0 +1,147 @@ +import os +from email.mime.multipart import MIMEMultipart +from email.mime.text import MIMEText + +import aiosmtplib +from jinja2 import Environment, FileSystemLoader, select_autoescape +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.logger import get_logger + +logger = get_logger(__name__) + +TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "..", "templates", "emails") + +jinja_env = Environment( + loader=FileSystemLoader(TEMPLATE_DIR), + autoescape=select_autoescape(["html"]), +) + + +async def send_dlq_alert( + dlq_count: int, + session: AsyncSession, +) -> None: + """ + Send a DLQ threshold alert email to all configured recipients. + """ + from app.core.config import settings as app_settings + from app.services.dlq import get_recent_dlq_jobs + from app.services.settings import get_alert_emails, get_dlq_threshold + + recipients = await get_alert_emails(session) + if not recipients: + logger.info( + "dlq_alert_skipped", + reason="no_recipients_configured", + dlq_count=dlq_count, + ) + return + + threshold = await get_dlq_threshold(session) + recent_jobs = await get_recent_dlq_jobs(session, limit=10) + + try: + template = jinja_env.get_template("dlq_alert.html") + html_body = template.render( + dlq_count=dlq_count, + threshold=threshold, + jobs=recent_jobs, + dashboard_url="https://app.flint.local/dlq", + app_name="Flint", + tagline="Quietly igniting your payload, every job has a spark.", + ) + except Exception as exc: + logger.error( + "dlq_alert_template_error", + error=str(exc), + ) + html_body = _plain_text_fallback(dlq_count, threshold, recent_jobs) + + subject = ( + f"[Flint Alert] DLQ threshold reached — " + f"{dlq_count} failed job{'s' if dlq_count != 1 else ''}" + ) + + for recipient in recipients: + await _send_email( + to=recipient, + subject=subject, + html_body=html_body, + smtp_host=app_settings.SMTP_HOST, + smtp_port=app_settings.SMTP_PORT, + smtp_from=app_settings.SMTP_FROM, + ) + + logger.warning( + "dlq_alert_sent", + dlq_count=dlq_count, + threshold=threshold, + recipient_count=len(recipients), + ) + + +async def _send_email( + to: str, + subject: str, + html_body: str, + smtp_host: str, + smtp_port: int, + smtp_from: str, +) -> None: + """Send a single HTML email via aiosmtplib.""" + msg = MIMEMultipart("alternative") + msg["Subject"] = subject + msg["From"] = smtp_from + msg["To"] = to + + # Attach plain text fallback first, then HTML + plain = "Flint DLQ Alert: threshold reached. Visit your dashboard for details." + msg.attach(MIMEText(plain, "plain", "utf-8")) + msg.attach(MIMEText(html_body, "html", "utf-8")) + + try: + await aiosmtplib.send( + msg, + hostname=smtp_host, + port=smtp_port, + use_tls=False, + start_tls=False, + ) + logger.info("dlq_alert_email_delivered", to=to) + except Exception as exc: + logger.error( + "dlq_alert_email_failed", + to=to, + error=str(exc), + ) + + +def _plain_text_fallback( + dlq_count: int, + threshold: int, + jobs: list[dict], +) -> str: + """ + Plain HTML fallback when the Jinja template fails to render. + """ + rows = "".join( + f"" + f"{j['short_id']}" + f"{j['type']}" + f"{j['last_error'][:80]}" + f"{j['retry_count']}/{j['max_retries']}" + f"{j['updated_at']}" + f"" + for j in jobs + ) + return f""" + +

Flint — DLQ Threshold Reached

+

{dlq_count} failed jobs in the DLQ (threshold: {threshold}).

+ + + {rows} +
IDTypeErrorRetriesFailed At
+ + """ diff --git a/app/services/dag.py b/app/services/dag.py new file mode 100644 index 0000000..3a1e1cc --- /dev/null +++ b/app/services/dag.py @@ -0,0 +1,379 @@ +import uuid +from typing import TYPE_CHECKING + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.exceptions import DependencyCycleException, DependencyNotFoundException +from app.core.logger import get_logger +from app.models.job import Job, JobStatus +from app.models.job_depedencies import JobDependency +from app.models.job_log import JobLog, LogEvent + +if TYPE_CHECKING: + from app.queues.base import BaseQueue + +logger = get_logger(__name__) + + +async def check_cycle( + new_job_id: uuid.UUID, + dependency_ids: list[uuid.UUID], + db: AsyncSession, +) -> None: + """ + Verify that adding dependency_ids to new_job_id does not create a cycle. + + Performs a depth-first search starting from each dependency_id, + traversing its own dependencies recursively. If new_job_id is + encountered during the traversal, a cycle would be created. + + Raises: + DependencyNotFoundException: If any dependency_id does not exist. + DependencyCycleException: If a cycle would be introduced. + """ + for dep_id in dependency_ids: + result = await db.execute( + select(Job.id).where( + Job.id == dep_id, + Job.deleted_at.is_(None), + ) + ) + if result.scalar_one_or_none() is None: + raise DependencyNotFoundException(str(dep_id)) + + for dep_id in dependency_ids: + visited: set[uuid.UUID] = set() + await _dfs_cycle_check( + start=dep_id, + target=new_job_id, + visited=visited, + db=db, + ) + + +async def _dfs_cycle_check( + start: uuid.UUID, + target: uuid.UUID, + visited: set[uuid.UUID], + db: AsyncSession, +) -> None: + """ + Recursive DFS. Raises DependencyCycleException if target is found. + """ + if start == target: + raise DependencyCycleException(str(target), str(start)) + + if start in visited: + return + visited.add(start) + + result = await db.execute( + select(JobDependency.depends_on_id).where(JobDependency.job_id == start) + ) + upstream_ids = [row[0] for row in result.fetchall()] + + for upstream_id in upstream_ids: + await _dfs_cycle_check( + start=upstream_id, + target=target, + visited=visited, + db=db, + ) + + +async def create_dependencies( + job_id: uuid.UUID, + dependency_ids: list[uuid.UUID], + db: AsyncSession, +) -> None: + """ + Insert rows into job_dependencies for all dependency_ids. + Assumes cycle check has already been performed. + """ + for dep_id in dependency_ids: + dependency = JobDependency( + job_id=job_id, + depends_on_id=dep_id, + ) + db.add(dependency) + + await db.flush() + logger.info( + "job_dependencies_created", + job_id=str(job_id), + dependency_count=len(dependency_ids), + dependency_ids=[str(d) for d in dependency_ids], + ) + + +async def on_job_completed( + job_id: uuid.UUID, + db: AsyncSession, + queue: "BaseQueue", +) -> None: + """ + Called after every successful job completion. + """ + result = await db.execute( + select(JobDependency.job_id).where(JobDependency.depends_on_id == job_id) + ) + dependent_job_ids = [row[0] for row in result.fetchall()] + + for dependent_id in dependent_job_ids: + await _maybe_unblock_job(dependent_id, db, queue) + + +async def _maybe_unblock_job( + job_id: uuid.UUID, + db: AsyncSession, + queue: "BaseQueue", +) -> None: + """ + Check if all dependencies of job_id are completed. + If yes, push the job onto the queue. + """ + result = await db.execute( + select(Job).where( + Job.id == job_id, + Job.deleted_at.is_(None), + ) + ) + job = result.scalar_one_or_none() + if not job or job.status != JobStatus.PENDING: + return + + total_result = await db.execute( + select(func.count(JobDependency.id)).where(JobDependency.job_id == job_id) + ) + total_deps = total_result.scalar() or 0 + + if total_deps == 0: + return + + # Count completed dependencies + completed_result = await db.execute( + select(func.count(Job.id)) + .join(JobDependency, Job.id == JobDependency.depends_on_id) + .where( + JobDependency.job_id == job_id, + Job.status == JobStatus.COMPLETED, + ) + ) + completed_count = completed_result.scalar() or 0 + + if completed_count == total_deps: + logger.info( + "job_unblocked", + job_id=str(job_id), + completed_deps=completed_count, + ) + await queue.push( + job_id=str(job_id), + effective_priority=job.effective_priority, + scheduled_at=job.scheduled_at.timestamp(), + created_at=job.created_at.timestamp(), + ) + + +async def on_job_failed( + job_id: uuid.UUID, + db: AsyncSession, +) -> None: + """ + Called when a job exhausts its retries and moves to the DLQ. + """ + await _cascade_cancel(job_id, db) + + +async def _cascade_cancel( + failed_job_id: uuid.UUID, + db: AsyncSession, +) -> None: + """ + BFS through the dependency graph cancelling all downstream jobs. + Only cancels jobs that are still 'pending'. + """ + queue: list[uuid.UUID] = [failed_job_id] + visited: set[uuid.UUID] = {failed_job_id} + + while queue: + current_id = queue.pop(0) + + # Find all jobs that directly depend on current_id + result = await db.execute( + select(JobDependency.job_id).where( + JobDependency.depends_on_id == current_id + ) + ) + dependent_ids = [row[0] for row in result.fetchall()] + + for dep_id in dependent_ids: + if dep_id in visited: + continue + visited.add(dep_id) + + update_result = await db.execute( + update(Job) + .where( + Job.id == dep_id, + Job.status == JobStatus.PENDING, + Job.deleted_at.is_(None), + ) + .values( + status=JobStatus.CANCELLED, + last_error=( + f"Cancelled: dependency job {failed_job_id} " + f"failed permanently and was moved to DLQ." + ), + updated_at=func.now(), + ) + .returning(Job.id) + ) + cancelled_id = update_result.scalar_one_or_none() + + if cancelled_id: + log_entry = JobLog( + job_id=dep_id, + event=LogEvent.JOB_CANCELLED, + message=( + f"Job automatically cancelled because upstream " + f"dependency {failed_job_id} failed permanently." + ), + metadata_={ + "failed_dependency_id": str(failed_job_id), + "reason": "upstream_dependency_failed", + }, + ) + db.add(log_entry) + + logger.info( + "job_cascade_cancelled", + job_id=str(dep_id), + failed_dependency=str(failed_job_id), + ) + + queue.append(dep_id) + + await db.commit() + + +async def on_dag_root_retried( + job_id: uuid.UUID, + db: AsyncSession, +) -> None: + """ + Called when manually retries a DLQ job that has downstream dependents. + """ + await _cascade_reset(job_id, db) + + +async def _cascade_reset( + retried_job_id: uuid.UUID, + session: AsyncSession, +) -> None: + """ + BFS through the dependency graph resetting auto-cancelled downstream jobs. + """ + queue: list[uuid.UUID] = [retried_job_id] + visited: set[uuid.UUID] = {retried_job_id} + + while queue: + current_id = queue.pop(0) + + result = await session.execute( + select(JobDependency.job_id).where( + JobDependency.depends_on_id == current_id + ) + ) + dependent_ids = [row[0] for row in result.fetchall()] + + for dep_id in dependent_ids: + if dep_id in visited: + continue + visited.add(dep_id) + + # Only reset jobs that were auto-cancelled due to dependency failure + dep_result = await session.execute( + select(Job).where( + Job.id == dep_id, + Job.deleted_at.is_(None), + ) + ) + dep_job = dep_result.scalar_one_or_none() + + if not dep_job: + continue + + is_auto_cancelled = ( + dep_job.status == JobStatus.CANCELLED + and dep_job.last_error is not None + and "dependency" in dep_job.last_error.lower() + ) + + if is_auto_cancelled: + await session.execute( + update(Job) + .where(Job.id == dep_id) + .values( + status=JobStatus.PENDING, + retry_count=0, + last_error=None, + worker_id=None, + updated_at=func.now(), + ) + ) + + log_entry = JobLog( + job_id=dep_id, + event=LogEvent.JOB_CREATED, + message=( + f"Job reset to pending: upstream dependency " + f"{retried_job_id} was manually retried." + ), + metadata_={ + "retried_dependency_id": str(retried_job_id), + "reason": "upstream_dependency_retried", + }, + ) + session.add(log_entry) + + logger.info( + "job_cascade_reset", + job_id=str(dep_id), + retried_dependency=str(retried_job_id), + ) + + queue.append(dep_id) + + await session.commit() + + +async def get_dependency_ids( + job_id: uuid.UUID, + db: AsyncSession, +) -> list[uuid.UUID]: + """Return the list of job IDs that job_id depends on.""" + result = await db.execute( + select(JobDependency.depends_on_id).where(JobDependency.job_id == job_id) + ) + return [row[0] for row in result.fetchall()] + + +async def has_unmet_dependencies( + job_id: uuid.UUID, + session: AsyncSession, +) -> bool: + """ + Return True if job_id has at least one dependency that is not yet completed. + """ + result = await session.execute( + select(func.count(JobDependency.id)) + .join(Job, Job.id == JobDependency.depends_on_id) + .where( + JobDependency.job_id == job_id, + Job.status != JobStatus.COMPLETED, + ) + ) + unmet_count = result.scalar() or 0 + return unmet_count > 0 diff --git a/app/services/dlq.py b/app/services/dlq.py new file mode 100644 index 0000000..b8baa20 --- /dev/null +++ b/app/services/dlq.py @@ -0,0 +1,246 @@ +import uuid + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.exceptions import JobNotInDLQException +from app.core.logger import get_logger +from app.models.job import Job, JobStatus +from app.models.job_log import JobLog, LogEvent +from app.queues.base import BaseQueue +from app.services.settings import get_dlq_threshold + +logger = get_logger(__name__) + + +async def send_to_dlq( + job_id: uuid.UUID, + error: str, + db: AsyncSession, +) -> None: + """ + Move a job to the dead letter queue after exhausting all retries. + """ + await db.execute( + update(Job) + .where(Job.id == job_id) + .values( + status=JobStatus.FAILED, + is_dlq=True, + last_error=error, + worker_id=None, + updated_at=func.now(), + ) + ) + + log_entry = JobLog( + job_id=job_id, + event=LogEvent.JOB_FAILED, + message=f"Job moved to DLQ after exhausting all retries. Error: {error}", + metadata_={"error": error, "reason": "max_retries_exhausted"}, + ) + db.add(log_entry) + await db.flush() + + logger.error( + "job_failed", + job_id=str(job_id), + error=error[:200], + reason="max_retries_exhausted", + ) + + from app.services import dag + + await dag.on_job_failed(job_id, db) + + await _check_and_alert(db) + + +async def _check_and_alert(session: AsyncSession) -> None: + """ + Count current DLQ jobs. If count >= threshold, fire alert email. + """ + count = await get_dlq_count(session) + threshold = await get_dlq_threshold(session) + + logger.info("dlq_count_check", count=count, threshold=threshold) + + if count >= threshold: + logger.warning( + "dlq_threshold_reached", + count=count, + threshold=threshold, + ) + from app.services import alert + + await alert.send_dlq_alert(count, session) + + +async def get_dlq_jobs( + page: int, + limit: int, + db: AsyncSession, +) -> tuple[list[Job], int]: + """ + Return paginated DLQ jobs with total count. + """ + base_filter = [ + Job.is_dlq.is_(True), + Job.deleted_at.is_(None), + ] + + total_result = await db.execute(select(func.count(Job.id)).where(*base_filter)) + total = total_result.scalar() or 0 + + offset = (page - 1) * limit + jobs_result = await db.execute( + select(Job) + .where(*base_filter) + .order_by(Job.updated_at.desc()) + .offset(offset) + .limit(limit) + ) + jobs = list(jobs_result.scalars().all()) + + return jobs, total + + +async def get_dlq_count(db: AsyncSession) -> int: + """Return the current number of jobs in the DLQ.""" + result = await db.execute( + select(func.count(Job.id)).where( + Job.is_dlq.is_(True), + Job.deleted_at.is_(None), + ) + ) + return result.scalar() or 0 + + +async def get_recent_dlq_jobs( + db: AsyncSession, + limit: int = 10, +) -> list[dict]: + """ + Return the most recent DLQ jobs as plain dicts. + """ + result = await db.execute( + select(Job) + .where( + Job.is_dlq.is_(True), + Job.deleted_at.is_(None), + ) + .order_by(Job.updated_at.desc()) + .limit(limit) + ) + jobs = result.scalars().all() + return [ + { + "id": str(job.id), + "short_id": str(job.id)[:8], + "type": job.type, + "last_error": job.last_error or "Unknown error", + "retry_count": job.retry_count, + "max_retries": job.max_retries, + "updated_at": job.updated_at.strftime("%Y-%m-%d %H:%M:%S UTC"), + } + for job in jobs + ] + + +async def retry_dlq_job( + job_id: uuid.UUID, + db: AsyncSession, + queue: BaseQueue, +) -> Job: + """ + Manually retry a job from the DLQ. + """ + result = await db.execute( + select(Job).where( + Job.id == job_id, + Job.deleted_at.is_(None), + ) + ) + job = result.scalar_one_or_none() + + if not job or not job.is_dlq: + raise JobNotInDLQException(str(job_id)) + + # Reset the job + await db.execute( + update(Job) + .where(Job.id == job_id) + .values( + status=JobStatus.PENDING, + is_dlq=False, + retry_count=0, + last_error=None, + worker_id=None, + effective_priority=float(job.priority), + updated_at=func.now(), + ) + ) + + log_entry = JobLog( + job_id=job_id, + event=LogEvent.JOB_CREATED, + message="Job manually retried from DLQ by engineer.", + metadata_={"reason": "manual_dlq_retry"}, + ) + db.add(log_entry) + await db.flush() + + logger.info("dlq_job_retried", job_id=str(job_id)) + + # Cascade reset downstream auto-cancelled dependents + from app.services import dag + + await dag.on_dag_root_retried(job_id, db) + + # Re-evaluate DAG: push to queue only if all deps are met + has_unmet = await dag.has_unmet_dependencies(job_id, db) + if not has_unmet: + await queue.push( + job_id=str(job_id), + effective_priority=float(job.priority), + scheduled_at=job.scheduled_at.timestamp(), + created_at=job.created_at.timestamp(), + ) + # Also sync to Redis sorted set + from app.services.job_service import _sync_job_to_redis + + await _sync_job_to_redis(str(job_id), float(job.priority)) + + await db.commit() + + refreshed = await db.get(Job, job_id) + return refreshed + + +async def remove_from_dlq( + job_id: uuid.UUID, + session: AsyncSession, +) -> None: + """ + Soft-delete a job from the DLQ. + The job moves to the bin and can be hard-deleted from there. + """ + result = await session.execute( + select(Job).where( + Job.id == job_id, + Job.deleted_at.is_(None), + ) + ) + job = result.scalar_one_or_none() + + if not job or not job.is_dlq: + raise JobNotInDLQException(str(job_id)) + + await session.execute( + update(Job) + .where(Job.id == job_id) + .values(deleted_at=func.now(), updated_at=func.now()) + ) + await session.commit() + + logger.info("dlq_job_removed", job_id=str(job_id)) diff --git a/app/services/job.py b/app/services/job.py new file mode 100644 index 0000000..67bbf6e --- /dev/null +++ b/app/services/job.py @@ -0,0 +1,456 @@ +import json +import uuid +from datetime import UTC, datetime + +import redis.asyncio as aioredis +from sqlalchemy import func, or_, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.config import settings as app_settings +from app.core.exceptions import ( + JobNotCancellableException, + JobNotDeletableException, + JobNotFoundException, + JobNotInBinException, +) +from app.core.logger import get_logger +from app.models.job import Job, JobStatus +from app.models.job_log import JobLog, LogEvent +from app.schemas.job import JobCreate, JobFilterParams, parse_interval +from app.services import dag + +logger = get_logger(__name__) + +QUEUE_KEY = "flint:queue" +EVENTS_CHANNEL = "flint:events" + + +async def _get_redis() -> aioredis.Redis: + """Get a Redis client. Used internally by this service.""" + return aioredis.from_url( + app_settings.REDIS_URL, + decode_responses=True, + encoding="utf-8", + ) + + +async def _sync_job_to_redis(job_id: str, effective_priority: float) -> None: + """ + Add a job to the Redis sorted set (flint:queue). + Score = effective_priority. Lower score = higher urgency. + Workers read from this set to know which job to process next. + """ + redis = await _get_redis() + try: + await redis.zadd(QUEUE_KEY, {job_id: effective_priority}) + finally: + await redis.aclose() + + +async def _remove_job_from_redis(job_id: str) -> None: + """Remove a job from the Redis sorted set.""" + redis = await _get_redis() + try: + await redis.zrem(QUEUE_KEY, job_id) + finally: + await redis.aclose() + + +async def _publish_sse_event(event: dict) -> None: + """ + Publish a job status change event to the Redis pub/sub channel. + The SSE endpoint subscribes to this channel and forwards events + to connected browser clients. + """ + redis = await _get_redis() + try: + await redis.publish(EVENTS_CHANNEL, json.dumps(event)) + finally: + await redis.aclose() + + +async def create_job( + data: JobCreate, + db: AsyncSession, +) -> Job: + """ + Create a new job and optionally push it to the queue. + """ + # Parse interval + interval_seconds = None + if data.interval: + interval_seconds = parse_interval(data.interval) + + dependency_ids = data.dependency_ids or [] + if dependency_ids: + await dag.check_cycle( + new_job_id=uuid.uuid4(), # placeholder — job not yet in DB + dependency_ids=dependency_ids, + db=db, + ) + + scheduled_at = data.scheduled_at or datetime.now(UTC) + + job = Job( + type=data.type, + payload=data.payload, + priority=int(data.priority), + effective_priority=float(data.priority), + status=JobStatus.PENDING, + scheduled_at=scheduled_at, + interval_seconds=interval_seconds, + max_retries=data.max_retries, + retry_count=0, + ) + db.add(job) + await db.flush() + + # Now run cycle check with the real job ID + if dependency_ids: + await dag.check_cycle( + new_job_id=job.id, + dependency_ids=dependency_ids, + db=db, + ) + await dag.create_dependencies(job.id, dependency_ids, db) + + log_entry = JobLog( + job_id=job.id, + event=LogEvent.JOB_CREATED, + message=f"Job created with type '{job.type}' and priority {job.priority}.", + metadata_={ + "type": job.type, + "priority": job.priority, + "scheduled_at": scheduled_at.isoformat(), + "interval_seconds": interval_seconds, + "dependency_count": len(dependency_ids), + }, + ) + db.add(log_entry) + await db.commit() + + logger.info( + "job_created", + job_id=str(job.id), + type=job.type, + priority=job.priority, + scheduled_at=scheduled_at.isoformat(), + has_dependencies=bool(dependency_ids), + ) + + has_unmet = bool(dependency_ids) + is_due = scheduled_at <= datetime.now(UTC) + + if not has_unmet and is_due: + await _sync_job_to_redis(str(job.id), job.effective_priority) + + await _publish_sse_event( + { + "job_id": str(job.id), + "status": JobStatus.PENDING, + "type": job.type, + } + ) + + return job + + +async def get_jobs( + filters: JobFilterParams, + db: AsyncSession, +) -> tuple[list[Job], int]: + """ + Return paginated jobs with total count. + """ + conditions: list = [ + Job.deleted_at.is_(None), + Job.is_dlq.is_(False), + ] + + if filters.status: + conditions.append(Job.status == filters.status) + if filters.type: + conditions.append(Job.type == filters.type) + if filters.priority: + conditions.append(Job.priority == int(filters.priority)) + if filters.search: + search_term = f"%{filters.search}%" + conditions.append( + or_( + Job.type.ilike(search_term), + Job.id.cast(db_string_type()).ilike(search_term), + ) + ) + + total_result = await db.execute(select(func.count(Job.id)).where(*conditions)) + total = total_result.scalar() or 0 + + offset = (filters.page - 1) * filters.limit + jobs_result = await db.execute( + select(Job) + .where(*conditions) + .order_by(Job.created_at.desc()) + .offset(offset) + .limit(filters.limit) + ) + jobs = list(jobs_result.scalars().all()) + + return jobs, total + + +def db_string_type(): + """SQLAlchemy string type for UUID casting in search.""" + from sqlalchemy import String + + return String + + +async def get_job_by_id( + job_id: uuid.UUID, + db: AsyncSession, + include_logs: bool = False, + include_dependencies: bool = False, +) -> Job: + """ + Fetch a single job by ID. + """ + result = await db.execute( + select(Job).where( + Job.id == job_id, + Job.deleted_at.is_(None), + ) + ) + job = result.scalar_one_or_none() + if not job: + raise JobNotFoundException(str(job_id)) + return job + + +async def get_job_with_details( + job_id: uuid.UUID, + db: AsyncSession, +) -> tuple[Job, list[uuid.UUID], list]: + """ + Fetch a job with its dependency IDs and log entries. + """ + job = await get_job_by_id(job_id, db) + + dependency_ids = await dag.get_dependency_ids(job_id, db) + + from sqlalchemy import asc + + from app.models.job_log import JobLog + + logs_result = await db.execute( + select(JobLog).where(JobLog.job_id == job_id).order_by(asc(JobLog.created_at)) + ) + logs = list(logs_result.scalars().all()) + + return job, dependency_ids, logs + + +async def cancel_job( + job_id: uuid.UUID, + db: AsyncSession, +) -> Job: + """ + Request cancellation of a job. + """ + job = await get_job_by_id(job_id, db) + + if not job.is_cancellable: + raise JobNotCancellableException(str(job_id), job.status) + + if job.status == JobStatus.PENDING: + # Immediate cancellation + await db.execute( + update(Job) + .where(Job.id == job_id) + .values( + status=JobStatus.CANCELLED, + cancellation_requested=True, + updated_at=func.now(), + ) + ) + + await _remove_job_from_redis(str(job_id)) + + # Cascade cancel downstream dependents + await dag.on_job_failed(job_id, db) + + log_entry = JobLog( + job_id=job_id, + event=LogEvent.JOB_CANCELLED, + message="Job cancelled by user request while pending.", + metadata_={"reason": "user_requested", "was_status": "pending"}, + ) + db.add(log_entry) + + logger.info("job_cancelled", job_id=str(job_id), was_status="pending") + + elif job.status == JobStatus.PROCESSING: + await db.execute( + update(Job) + .where(Job.id == job_id) + .values( + cancellation_requested=True, + updated_at=func.now(), + ) + ) + + log_entry = JobLog( + job_id=job_id, + event=LogEvent.JOB_CANCELLED, + message=( + "Cancellation requested while job is processing. " + "Worker will honour at next checkpoint." + ), + metadata_={"reason": "user_requested", "was_status": "processing"}, + ) + db.add(log_entry) + + logger.info( + "job_cancellation_requested", + job_id=str(job_id), + was_status="processing", + ) + + await db.commit() + + await _publish_sse_event( + { + "job_id": str(job_id), + "status": JobStatus.CANCELLED, + } + ) + + job = await db.get(Job, job_id) + assert job is not None, f"Job {job_id} vanished after cancellation" + return job + + +async def soft_delete_job( + job_id: uuid.UUID, + db: AsyncSession, +) -> None: + """ + Move a job to the bin (sets deleted_at). + """ + job = await get_job_by_id(job_id, db) + + if not job.is_terminal: + raise JobNotDeletableException(str(job_id), job.status) + + await db.execute( + update(Job) + .where(Job.id == job_id) + .values(deleted_at=func.now(), updated_at=func.now()) + ) + await db.commit() + + logger.info("job_soft_deleted", job_id=str(job_id)) + + +async def get_bin_jobs( + page: int, + limit: int, + db: AsyncSession, +) -> tuple[list[Job], int]: + """ + Return soft-deleted jobs (the bin). Paginated, most recently deleted first. + """ + conditions = [Job.deleted_at.isnot(None)] + + total_result = await db.execute(select(func.count(Job.id)).where(*conditions)) + total = total_result.scalar() or 0 + + offset = (page - 1) * limit + jobs_result = await db.execute( + select(Job) + .where(*conditions) + .order_by(Job.deleted_at.desc()) + .offset(offset) + .limit(limit) + ) + jobs = list(jobs_result.scalars().all()) + + return jobs, total + + +async def restore_job( + job_id: uuid.UUID, + db: AsyncSession, +) -> Job: + """ + Restore a soft-deleted job from the bin. + Clears deleted_at. Does not re-queue the job. + Raises JobNotInBinException if the job is not soft-deleted. + """ + result = await db.execute(select(Job).where(Job.id == job_id)) + job = result.scalar_one_or_none() + + if not job or job.deleted_at is None: + raise JobNotInBinException(str(job_id)) + + await db.execute( + update(Job) + .where(Job.id == job_id) + .values(deleted_at=None, updated_at=func.now()) + ) + await db.commit() + + logger.info("job_restored", job_id=str(job_id)) + job = await db.get(Job, job_id) + assert job is not None, f"Job {job_id} vanished after cancellation" + return job + + +async def hard_delete_job( + job_id: uuid.UUID, + db: AsyncSession, +) -> None: + """ + Permanently delete a job from the database. + """ + result = await db.execute(select(Job).where(Job.id == job_id)) + job = result.scalar_one_or_none() + + if not job or job.deleted_at is None: + raise JobNotInBinException(str(job_id)) + + await db.delete(job) + await db.commit() + + logger.info("job_hard_deleted", job_id=str(job_id)) + + +async def get_job_counts_by_status( + db: AsyncSession, +) -> dict[str, int]: + """ + Return a dict of job counts per status for the dashboard. + Excludes soft-deleted jobs. Includes DLQ count separately. + """ + + from app.services.dlq import get_dlq_count + + results = await db.execute( + select(Job.status, func.count(Job.id)) + .where( + Job.deleted_at.is_(None), + Job.is_dlq.is_(False), + ) + .group_by(Job.status) + ) + counts = {status: count for status, count in results.fetchall()} + + dlq_count = await get_dlq_count(db) + + return { + "pending": counts.get(JobStatus.PENDING, 0), + "processing": counts.get(JobStatus.PROCESSING, 0), + "completed": counts.get(JobStatus.COMPLETED, 0), + "failed": counts.get(JobStatus.FAILED, 0), + "cancelled": counts.get(JobStatus.CANCELLED, 0), + "dlq": dlq_count, + } diff --git a/app/services/settings.py b/app/services/settings.py new file mode 100644 index 0000000..6efb8f9 --- /dev/null +++ b/app/services/settings.py @@ -0,0 +1,111 @@ +import json + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.exceptions import SettingNotFoundException +from app.core.logger import get_logger +from app.models.settings import Setting, SettingKey + +logger = get_logger(__name__) + + +async def get_setting(key: str, session: AsyncSession) -> str: + """ + Return the value string for the given settings key. + """ + result = await session.execute(select(Setting.value).where(Setting.key == key)) + value = result.scalar_one_or_none() + if value is None: + raise SettingNotFoundException(key) + return value + + +async def get_setting_with_default( + key: str, + default: str, + db: AsyncSession, +) -> str: + """ + Return the value for a settings key, falling back to default if not found. + Useful in worker/scheduler code where a missing key should not crash. + """ + result = await db.execute(select(Setting.value).where(Setting.key == key)) + value = result.scalar_one_or_none() + return value if value is not None else default + + +async def get_all_settings(db: AsyncSession) -> dict[str, str]: + """ + Return all settings as a flat {key: value} dict. + """ + result = await db.execute(select(Setting)) + settings = result.scalars().all() + return {s.key: s.value for s in settings} + + +async def update_settings( + updates: dict[str, str], + db: AsyncSession, +) -> dict[str, str]: + """ + Upsert one or more settings by key. + """ + for key, value in updates.items(): + result = await db.execute(select(Setting).where(Setting.key == key)) + existing = result.scalar_one_or_none() + + if existing: + existing.value = value + logger.info( + "setting_updated", + key=key, + new_value=value if key != "alert_emails" else "***", + ) + else: + new_setting = Setting(key=key, value=value) + db.add(new_setting) + logger.info("setting_created", key=key) + + await db.commit() + return await get_all_settings(db) + + +async def get_dlq_threshold(db: AsyncSession) -> int: + """ + Return the DLQ threshold as an integer. + Falls back to 5 if the setting is missing or unparseable. + """ + raw = await get_setting_with_default(SettingKey.DLQ_THRESHOLD, "5", db) + try: + return int(raw) + except (ValueError, TypeError): + logger.warning("dlq_threshold_parse_error", raw_value=raw) + return 5 + + +async def get_alert_emails(db: AsyncSession) -> list[str]: + """ + Return the alert email list. + """ + raw = await get_setting_with_default(SettingKey.ALERT_EMAILS, "[]", db) + try: + parsed = json.loads(raw) + if isinstance(parsed, list): + return [str(e) for e in parsed if e] + return [] + except (ValueError, TypeError, json.JSONDecodeError): + logger.warning("alert_emails_parse_error", raw_value=raw) + return [] + + +async def get_scheduler_strategy(db: AsyncSession) -> str: + """ + Return the active scheduler strategy: 'heap' or 'timing_wheel'. + Falls back to 'heap' if the setting is missing or invalid. + """ + raw = await get_setting_with_default(SettingKey.SCHEDULER_STRATEGY, "heap", db) + if raw not in ("heap", "timing_wheel"): + logger.warning("scheduler_strategy_invalid", raw_value=raw) + return "heap" + return raw diff --git a/app/templates/email/base.html b/app/templates/email/base.html new file mode 100644 index 0000000..b2022b5 --- /dev/null +++ b/app/templates/email/base.html @@ -0,0 +1,163 @@ + + + + + + {% block title %}Flint Notification{% endblock %} + + + +
+
+
+
+ ◆ Flint +
+
{{ tagline }}
+
+
+
{% block content %}{% endblock %}
+
+ Flint — {{ tagline }}
+ This is an automated alert. Do not reply to this email. +
+
+ + diff --git a/app/templates/email/dlq_alert.html b/app/templates/email/dlq_alert.html new file mode 100644 index 0000000..1c4c658 --- /dev/null +++ b/app/templates/email/dlq_alert.html @@ -0,0 +1,61 @@ +{% extends "base.html" %} {% block title %}Flint DLQ Alert — {{ dlq_count }} +Failed Jobs{% endblock %} {% block content %} +

⚠️ Dead Letter Queue Threshold Reached

+ +

+ {{ dlq_count }} + failed job{{ 's' if dlq_count != 1 else '' }} have accumulated in the Dead + Letter Queue. Your configured threshold is + {{ threshold }}. +

+ +

+ These jobs exhausted all retry attempts and require manual inspection. Review + the error details below and retry each job once the underlying issue has been + resolved. +

+ +{% if jobs %} + + + + + + + + + + + + {% for job in jobs %} + + + + + + + + {% endfor %} + +
Job IDTypeErrorRetriesFailed At
{{ job.short_id }} + {{ job.type }} + +
+ {{ job.last_error | truncate(80, true, '...') }} +
+ 		+ {{ job.retry_count }}/{{ job.max_retries }} + {{ job.updated_at }}
+{% else %} +

No job details available.

+{% endif %} + +

+ Use the Flint dashboard to inspect error details, fix the underlying issue, + and manually retry each job. +

+ + + View Dead Letter Queue → + +{% endblock %}