feat: process jobs; upload_pdf to sftp; status callbacks

sftp_store.py

@@ -80,3 +80,10 @@ def upload_bytes(remote_path: str, data: bytes) -> None:
             sftp.close()
         finally:
             transport.close()
+
+
+def upload_pdf(pdf_id: str, data: bytes) -> str:
+    # Stores under: pdfs/{pdf_id}.pdf (relative to SFTP_ROOT)
+    remote_path = f"pdfs/{pdf_id}.pdf"
+    upload_bytes(remote_path, data)
+    return remote_path

worker.py

@@ -1,8 +1,11 @@
 import os
 import time
+import base64
 import threading
 import requests
 from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from sftp_store import upload_pdf
 from sftp_store import sftp_test
 
 import sftp_store
@@ -46,6 +49,53 @@ def serve_http():
     HTTPServer(("0.0.0.0", PORT), Handler).serve_forever()
 
 
+def _api_headers() -> dict:
+    # API expects X-Worker-Token (NOT Authorization Bearer)
+    return {"X-Worker-Token": WORKER_TOKEN}
+
+def _job_status(job_id: str, status: str, message: str = "") -> None:
+    try:
+        requests.post(
+            f"{API_BASE}/api/jobs/{job_id}/status",
+            headers=_api_headers(),
+            json={"status": status, "message": message},
+            timeout=10,
+        )
+    except Exception as e:
+        print(f"status update failed: {e}", flush=True)
+
+def process_job(job: dict) -> None:
+    # API returns job dict (asdict(Job)) => id field is the canonical one
+    jid = job.get("id") or job.get("job_id")
+    if not jid:
+        raise RuntimeError("job missing id/job_id")
+
+    payload = job.get("payload") or {}
+    action = payload.get("action")
+
+    if action == "upload_pdf":
+        pdf_id = payload.get("pdf_id")
+        pdf_b64 = payload.get("pdf_b64")
+        if not pdf_id or not pdf_b64:
+            raise RuntimeError("upload_pdf requires payload.pdf_id and payload.pdf_b64")
+
+        # decode
+        try:
+            pdf_bytes = base64.b64decode(pdf_b64)
+        except Exception as e:
+            raise RuntimeError(f"invalid base64 in pdf_b64: {e}")
+
+        # upload
+        remote_path = upload_pdf(pdf_id, pdf_bytes)
+        _job_status(jid, "done", f"uploaded:{remote_path}")
+        print(f"job {jid} done -> {remote_path}", flush=True)
+        return
+
+    # unknown action -> fail but don't crash loop
+    _job_status(jid, "failed", f"unknown action: {action}")
+    print(f"job {jid} failed: unknown action {action}", flush=True)
+
+
 def poll_loop():
     print("Worker polling started", flush=True)