shoehorning graphql, step functions and lamdas. aws deployment scripts

api/routes/jobs.py

@@ -2,17 +2,20 @@
 Job endpoints - transcode/trim job management.
 """
 
-import json
+import os
+from pathlib import Path
 from typing import Optional
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, Header, HTTPException, Query
 
 from api.deps import get_asset, get_job, get_preset
 from api.schemas import JobCreate, JobResponse
 
 router = APIRouter(prefix="/jobs", tags=["jobs"])
 
+CALLBACK_API_KEY = os.environ.get("CALLBACK_API_KEY", "")
+
 
 @router.post("/", response_model=JobResponse, status_code=201)
 def create_job(data: JobCreate):
@@ -36,7 +39,6 @@ def create_job(data: JobCreate):
     if data.preset_id:
         try:
             preset = TranscodePreset.objects.get(id=data.preset_id)
-            # Snapshot preset at job creation time
             preset_snapshot = {
                 "name": preset.name,
                 "container": preset.container,
@@ -61,22 +63,13 @@ def create_job(data: JobCreate):
             status_code=400, detail="Must specify preset_id or trim_start/trim_end"
         )
 
-    # Generate output filename and path
-    import os
-    from pathlib import Path
-
+    # Generate output filename - stored as S3 key in output bucket
     output_filename = data.output_filename
     if not output_filename:
         stem = Path(source.filename).stem
         ext = preset_snapshot.get("container", "mp4") if preset else "mp4"
         output_filename = f"{stem}_output.{ext}"
 
-    media_out = os.environ.get("MEDIA_OUT", "/app/media/out")
-    output_path = str(Path(media_out) / output_filename)
-
-    media_in = os.environ.get("MEDIA_IN", "/app/media/in")
-    source_path = str(Path(media_in) / source.file_path)
-
     # Create job
     job = TranscodeJob.objects.create(
         source_asset_id=source.id,
@@ -85,26 +78,95 @@ def create_job(data: JobCreate):
         trim_start=data.trim_start,
         trim_end=data.trim_end,
         output_filename=output_filename,
-        output_path=output_path,
+        output_path=output_filename,  # S3 key in output bucket
         priority=data.priority or 0,
     )
 
-    # Dispatch to Celery
+    # Dispatch based on executor mode
+    executor_mode = os.environ.get("MPR_EXECUTOR", "local")
+
+    if executor_mode == "lambda":
+        _dispatch_lambda(job, source, preset_snapshot)
+    else:
+        _dispatch_celery(job, source, preset_snapshot)
+
+    return job
+
+
+def _dispatch_celery(job, source, preset_snapshot):
+    """Dispatch job to Celery worker."""
     from task.tasks import run_transcode_job
 
     result = run_transcode_job.delay(
         job_id=str(job.id),
-        source_path=source_path,
-        output_path=output_path,
+        source_key=source.file_path,
+        output_key=job.output_filename,
         preset=preset_snapshot or None,
-        trim_start=data.trim_start,
-        trim_end=data.trim_end,
+        trim_start=job.trim_start,
+        trim_end=job.trim_end,
         duration=source.duration,
     )
     job.celery_task_id = result.id
     job.save(update_fields=["celery_task_id"])
 
-    return job
+
+def _dispatch_lambda(job, source, preset_snapshot):
+    """Dispatch job to AWS Step Functions."""
+    from task.executor import get_executor
+
+    executor = get_executor()
+    executor.run(
+        job_id=str(job.id),
+        source_path=source.file_path,
+        output_path=job.output_filename,
+        preset=preset_snapshot or None,
+        trim_start=job.trim_start,
+        trim_end=job.trim_end,
+        duration=source.duration,
+    )
+
+
+@router.post("/{job_id}/callback")
+def job_callback(
+    job_id: UUID,
+    payload: dict,
+    x_api_key: Optional[str] = Header(None),
+):
+    """
+    Callback endpoint for Lambda to report job completion.
+    Protected by API key.
+    """
+    if CALLBACK_API_KEY and x_api_key != CALLBACK_API_KEY:
+        raise HTTPException(status_code=403, detail="Invalid API key")
+
+    from django.utils import timezone
+
+    from mpr.media_assets.models import TranscodeJob
+
+    try:
+        job = TranscodeJob.objects.get(id=job_id)
+    except TranscodeJob.DoesNotExist:
+        raise HTTPException(status_code=404, detail="Job not found")
+
+    status = payload.get("status", "failed")
+    job.status = status
+    job.progress = 100.0 if status == "completed" else job.progress
+    update_fields = ["status", "progress"]
+
+    if payload.get("error"):
+        job.error_message = payload["error"]
+        update_fields.append("error_message")
+
+    if status == "completed":
+        job.completed_at = timezone.now()
+        update_fields.append("completed_at")
+    elif status == "failed":
+        job.completed_at = timezone.now()
+        update_fields.append("completed_at")
+
+    job.save(update_fields=update_fields)
+
+    return {"ok": True}
 
 
 @router.get("/", response_model=list[JobResponse])
@@ -118,12 +180,10 @@ def list_jobs(
     from mpr.media_assets.models import TranscodeJob
 
     qs = TranscodeJob.objects.all()
-
     if status:
         qs = qs.filter(status=status)
     if source_asset_id:
         qs = qs.filter(source_asset_id=source_asset_id)
-
     return list(qs[offset : offset + limit])
 
 
@@ -154,11 +214,8 @@ def cancel_job(job_id: UUID, job=Depends(get_job)):
             status_code=400, detail=f"Cannot cancel job with status: {job.status}"
         )
 
-    # TODO: Cancel via gRPC
-
     job.status = "cancelled"
     job.save(update_fields=["status"])
-
     return job
 
 
@@ -173,6 +230,4 @@ def retry_job(job_id: UUID, job=Depends(get_job)):
     job.error_message = None
     job.save(update_fields=["status", "progress", "error_message"])
 
-    # TODO: Resubmit via gRPC
-
     return job