refactor(worker): add pydantic payload validation to all jobs

replace unsafe data.get() with pydantic model_validate() for runtime
type safety. routers now return 422 for invalid payloads instead of 500.

Author: gahyun-git

apps/worker/src/jobs/data_cleanup.py

@@ -6,6 +6,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import DataCleanupPayload
 from src.lib.config import settings
 from src.lib.retry import with_retry
 
@@ -32,13 +33,12 @@ async def _call_api(self, headers: dict[str, str], payload: dict[str, Any]) -> i
             return response.status_code
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        retention_days: int = data.get("retention_days", 90)
-        resource_type: str = data.get("resource_type", "all")
+        payload = DataCleanupPayload.model_validate(data)
 
         logger.info(
             "data_cleanup_start",
-            retention_days=retention_days,
-            resource_type=resource_type,
+            retention_days=payload.retention_days,
+            resource_type=payload.resource_type,
         )
 
         headers: dict[str, str] = {}
@@ -47,14 +47,17 @@ async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
 
         status_code = await self._call_api(
             headers,
-            {"retention_days": retention_days, "resource_type": resource_type},
+            {
+                "retention_days": payload.retention_days,
+                "resource_type": payload.resource_type,
+            },
         )
 
         logger.info(
             "data_cleanup_complete",
             api_status=status_code,
         )
-        return {"cleaned": True, "retention_days": retention_days}
+        return {"cleaned": True, "retention_days": payload.retention_days}
 
 
 register_job(DataCleanupJob())

apps/worker/src/jobs/medication_reminder.py

@@ -5,6 +5,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import MedicationReminderPayload
 from src.lib.dispatch import dispatch_local
 
 logger = structlog.get_logger(__name__)
@@ -18,30 +19,28 @@ def job_type(self) -> str:
         return "medication.reminder"
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        host_id: str = data.get("host_id", "")
-        pill_name: str = data.get("pill_name", "Unknown")
-        tokens: list[str] = data.get("tokens", [])
+        payload = MedicationReminderPayload.model_validate(data)
 
-        if not tokens:
-            logger.warning("medication_reminder_no_tokens", host_id=host_id)
+        if not payload.tokens:
+            logger.warning("medication_reminder_no_tokens", host_id=payload.host_id)
             return {"sent_count": 0, "skipped": True}
 
         await dispatch_local(
             "notification.send",
             {
-                "tokens": tokens,
+                "tokens": payload.tokens,
                 "title": "Medication Reminder",
-                "body": f"Time to take {pill_name}",
-                "data": {"host_id": host_id, "type": "medication_reminder"},
+                "body": f"Time to take {payload.pill_name}",
+                "data": {"host_id": payload.host_id, "type": "medication_reminder"},
             },
         )
 
         logger.info(
             "medication_reminder_dispatched",
-            host_id=host_id,
-            pill_name=pill_name,
+            host_id=payload.host_id,
+            pill_name=payload.pill_name,
         )
-        return {"dispatched": True, "pill_name": pill_name}
+        return {"dispatched": True, "pill_name": payload.pill_name}
 
 
 register_job(MedicationReminderJob())

apps/worker/src/jobs/notification_send.py

@@ -7,6 +7,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import NotificationSendPayload
 from src.lib.config import settings
 from src.lib.retry import with_retry
 
@@ -35,10 +36,11 @@ def job_type(self) -> str:
         return "notification.send"
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        tokens: list[str] = data.get("tokens", [])
-        title: str = data.get("title", "")
-        body: str = data.get("body", "")
-        extra: dict[str, str] = data.get("data", {})
+        payload = NotificationSendPayload.model_validate(data)
+        tokens = payload.tokens
+        title = payload.title
+        body = payload.body
+        extra = payload.data
 
         if not tokens:
             logger.warning("notification_send_no_tokens")
@@ -57,7 +59,7 @@ async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
             message = messaging.MulticastMessage(
                 tokens=tokens,
                 notification=notification,
-                data={k: str(v) for k, v in extra.items()},
+                data=extra,
             )
             response: messaging.BatchResponse = await asyncio.wait_for(
                 asyncio.to_thread(messaging.send_each_for_multicast, message),

apps/worker/src/jobs/relation_inactive.py

@@ -6,6 +6,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import RelationInactiveCheckPayload
 from src.lib.config import settings
 from src.lib.retry import with_retry
 
@@ -32,24 +33,26 @@ async def _call_api(self, headers: dict[str, str], params: dict[str, Any]) -> in
             return response.status_code
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        threshold_days: int = data.get("threshold_days", 30)
+        payload = RelationInactiveCheckPayload.model_validate(data)
 
         logger.info(
             "relation_inactive_check_start",
-            threshold_days=threshold_days,
+            threshold_days=payload.threshold_days,
         )
 
         headers: dict[str, str] = {}
         if settings.INTERNAL_API_KEY:
             headers["X-Internal-Key"] = settings.INTERNAL_API_KEY
 
-        status_code = await self._call_api(headers, {"threshold_days": threshold_days})
+        status_code = await self._call_api(
+            headers, {"threshold_days": payload.threshold_days}
+        )
 
         logger.info(
             "relation_inactive_check_complete",
             api_status=status_code,
         )
-        return {"checked": True, "threshold_days": threshold_days}
+        return {"checked": True, "threshold_days": payload.threshold_days}
 
 
 register_job(RelationInactiveCheckJob())

apps/worker/src/jobs/schemas.py

@@ -0,0 +1,50 @@
+"""Pydantic payload schemas for worker jobs."""
+
+from pydantic import BaseModel, Field
+
+
+class NotificationSendPayload(BaseModel):
+    """Payload for notification.send job."""
+
+    tokens: list[str]
+    title: str = ""
+    body: str = ""
+    data: dict[str, str] = Field(default_factory=dict)
+
+
+class WellnessEscalationPayload(BaseModel):
+    """Payload for wellness.escalation job."""
+
+    log_id: str = ""
+    host_id: str = ""
+    status: str = ""
+    summary: str = ""
+    contact_tokens: list[str] = Field(default_factory=list)
+
+
+class MedicationReminderPayload(BaseModel):
+    """Payload for medication.reminder job."""
+
+    host_id: str = ""
+    pill_name: str = "Unknown"
+    tokens: list[str] = Field(default_factory=list)
+
+
+class DataCleanupPayload(BaseModel):
+    """Payload for data.cleanup job."""
+
+    retention_days: int = 90
+    resource_type: str = "all"
+
+
+class RelationInactiveCheckPayload(BaseModel):
+    """Payload for relation.inactive_check job."""
+
+    threshold_days: int = 30
+
+
+class WellnessAggregatePayload(BaseModel):
+    """Payload for wellness.aggregate job."""
+
+    host_id: str = Field(min_length=1)
+    date: str = Field(min_length=1)

apps/worker/src/jobs/wellness_aggregate.py

@@ -6,6 +6,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import WellnessAggregatePayload
 from src.lib.config import settings
 from src.lib.retry import with_retry
 
@@ -32,32 +33,33 @@ async def _call_api(self, headers: dict[str, str], params: dict[str, Any]) -> in
             return response.status_code
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        host_id: str = data.get("host_id", "")
-        date: str = data.get("date", "")
-
-        if not host_id or not date:
-            msg = "host_id and date are required"
-            raise ValueError(msg)
+        payload = WellnessAggregatePayload.model_validate(data)
 
         logger.info(
             "wellness_aggregate_start",
-            host_id=host_id,
-            date=date,
+            host_id=payload.host_id,
+            date=payload.date,
         )
 
         headers: dict[str, str] = {}
         if settings.INTERNAL_API_KEY:
             headers["X-Internal-Key"] = settings.INTERNAL_API_KEY
 
-        status_code = await self._call_api(headers, {"host_id": host_id, "date": date})
+        status_code = await self._call_api(
+            headers, {"host_id": payload.host_id, "date": payload.date}
+        )
 
         logger.info(
             "wellness_aggregate_complete",
-            host_id=host_id,
-            date=date,
+            host_id=payload.host_id,
+            date=payload.date,
             api_status=status_code,
         )
-        return {"host_id": host_id, "date": date, "aggregated": True}
+        return {
+            "host_id": payload.host_id,
+            "date": payload.date,
+            "aggregated": True,
+        }
 
 
 register_job(WellnessAggregateJob())

apps/worker/src/jobs/wellness_escalation.py

@@ -5,6 +5,7 @@
 import structlog
 
 from src.jobs.base import BaseJob, register_job
+from src.jobs.schemas import WellnessEscalationPayload
 from src.lib.dispatch import dispatch_local
 
 logger = structlog.get_logger(__name__)
@@ -18,43 +19,39 @@ def job_type(self) -> str:
         return "wellness.escalation"
 
     async def execute(self, data: dict[str, Any]) -> dict[str, Any]:
-        log_id: str = data.get("log_id", "")
-        host_id: str = data.get("host_id", "")
-        status: str = data.get("status", "")
-        summary: str = data.get("summary", "")
-        contact_tokens: list[str] = data.get("contact_tokens", [])
+        payload = WellnessEscalationPayload.model_validate(data)
 
-        if not contact_tokens:
+        if not payload.contact_tokens:
             logger.warning(
                 "wellness_escalation_no_contacts",
-                log_id=log_id,
-                host_id=host_id,
+                log_id=payload.log_id,
+                host_id=payload.host_id,
             )
             return {"escalated": False, "reason": "no_contacts"}
 
         await dispatch_local(
             "notification.send",
             {
-                "tokens": contact_tokens,
-                "title": f"URGENT: Wellness {status.upper()}",
-                "body": summary or "Immediate attention required",
+                "tokens": payload.contact_tokens,
+                "title": f"URGENT: Wellness {payload.status.upper()}",
+                "body": payload.summary or "Immediate attention required",
                 "data": {
-                    "log_id": log_id,
-                    "host_id": host_id,
+                    "log_id": payload.log_id,
+                    "host_id": payload.host_id,
                     "type": "wellness_escalation",
                 },
             },
         )
 
         logger.info(
             "wellness_escalation_dispatched",
-            log_id=log_id,
-            host_id=host_id,
-            contact_count=len(contact_tokens),
+            log_id=payload.log_id,
+            host_id=payload.host_id,
+            contact_count=len(payload.contact_tokens),
         )
         return {
             "escalated": True,
-            "contact_count": len(contact_tokens),
+            "contact_count": len(payload.contact_tokens),
         }
 
 

apps/worker/src/routers/pubsub.py

@@ -6,7 +6,7 @@
 
 import structlog
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from src.jobs.base import get_job
 from src.lib.idempotency import release_claim, try_claim
@@ -54,6 +54,14 @@ async def handle_pubsub_push(envelope: PubSubEnvelope) -> dict[str, Any]:
     logger.info("pubsub_job_execute_start", job_type=task_type)
     try:
         result = await job.execute(data)
+    except (ValueError, ValidationError) as exc:
+        release_claim(task_type, data, idempotency_key=idempotency_key)
+        logger.warning(
+            "pubsub_job_payload_invalid",
+            job_type=task_type,
+            error=str(exc),
+        )
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
     except Exception:
         release_claim(task_type, data, idempotency_key=idempotency_key)
         raise

apps/worker/src/routers/tasks.py

@@ -5,7 +5,7 @@
 import structlog
 from fastapi import APIRouter, HTTPException
 from opentelemetry import trace
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from src.jobs.base import get_job, list_jobs
 from src.lib.idempotency import release_claim, try_claim
@@ -43,6 +43,14 @@ async def process_task(payload: TaskPayload) -> dict[str, Any]:
             attributes={"job.type": payload.task_type},
         ):
             result = await job.execute(payload.data)
+    except (ValueError, ValidationError) as exc:
+        release_claim(payload.task_type, payload.data)
+        logger.warning(
+            "job_payload_invalid",
+            job_type=payload.task_type,
+            error=str(exc),
+        )
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
     except Exception:
         release_claim(payload.task_type, payload.data)
         raise