# REFACTOR.md — Worker-to-API Refactor Plan ## Context VideoDetect workers currently run as long-lived processes (`main.py` → `WorkerPool`) that query the database directly via `DBConnector`. The re-architecture moves to **independent, externally-invoked task processes** that communicate through a REST API provided by the Perl `api/app.pl` module. This enables finer-grained scaling, resilience to individual process crashes, and separation of concerns between scanning (discovery) and processing (AI inference). --- ## Target Architecture ``` ┌───────────┐ /api/v1/nexttask/AISCAN ┌─────────────┐ │ Scanner │ ──────────────────────────────> │ Perl API │ │ (long- │ <────────────────────────────── │ (Dancer2) │ │ lived) │ task reservation + assign_key ├─────────────┤ └───────────┘ │ DB │ │ │ ┌───────────────────────────────────────────┼──────────┐ │ ▼ │ │ ┌──────────────┐ │ │ POST /api/v1/task/:id/complete │ │ │ ◄─────────────────────────────────────────► │ │ │ │ │ │ ┌───────────────────────────────┘ │ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ ┌──────────────┐ │worker-1│ │worker-2│ │worker-N│ │ MariaDB │ │(short- │ │(short- │ │(short- │ │ │ │ lived) │ │ lived) │ │ lived) │ └──────────────┘ └────────┘ └────────┘ └────────┘ ``` Each worker process: 1. Starts → accepts `--tasks N` (default 1) via CLI argument 2. Loops up to N times: **claim task** → **fetch video data** → **process** → **submit results** 3. Exits cleanly after completing the requested count --- ## API Reference (from `api/app.pl`) ### GET `/api/v1/nexttask/:type` Claims the next pending task of the given type and reserves it. | Field | Type | Example | |-----------|---------|----------------------------------------------| | task | object | Task record from DB | | assign_key| string | `"worker_338"` (server-generated) | **On success:** HTTP 200 with JSON body: ```json { "task": { "id": 1, "video_id": 42, "task_type": "AISCAN", "status": "PENDING", "created_at": "2026-09-09T13:05:56", "updated_at": "2026-09-09T14:53:57", "assign_key": null, "results": null, "assigned_at": "2026-09-09T14:43:31" }, "assign_key": "worker_338" } ``` **On no tasks available:** HTTP 404 with `{"message":"No task"}`. > **NOTE:** The Perl API marks the task `IN_PROGRESS` and sets `assign_key` + `assigned_at` atomically (`UPDATE ... WHERE id=? AND status='PENDING'`). This is safe for multiple workers racing to claim the same task. --- ### GET `/api/v1/video/:id` Returns full video metadata by ID. | Field | Type | Example | |---------------|---------|------------------------------------------------------| | id | integer | `1` | | file_path | string | `"/data/The Fappening/Sextape - Alyson Hannigan (American actress - American pie).wmv"` | | file_size | integer | `19358016` | | file_hash | string | `"1ffd178e9ee23039aebffd79ddcbc88e983633edcb75062e6bd4c269f4d7bf94"` | | resolution_w | integer | `640` | | resolution_h | integer | `480` | | codec | string | `"wmv1"` | | duration | float | `102.499` | | last_scan_time| datetime| `"2026-09-09T13:05:56"` | | created_at | datetime| `"2026-09-09T13:05:56"` | | updated_at | datetime| `"2026-09-09T13:05:56"` | **On not found:** HTTP 404 with `{"message":"No video"}`. --- ### POST `/api/v1/task/:task/complete` Submits processing results for a claimed task. The API verifies the task is `IN_PROGRESS` and assigned to the given `assign_key`. **Required parameters (form-encoded or JSON body):** | Field | Type | Description | |-------------|--------|------------------------------------------------| | assign_key | string | Must match the key returned by `/nexttask` | | results | object | JSON-serializable dict to store in DB `results` column | **On success:** HTTP 200 with `{"message":"Task completed successfully"}`. **On mismatch or task not IN_PROGRESS:** HTTP 403 with error message. > **NOTE:** The Perl API currently uses `body_parameters->get("results")`, which reads form-encoded data. For JSON body submission, the Dancer2 config may need `serializer: JSON` (which is already set in `api/config.yml`). However, `body_parameters` only parses form-encoded fields — JSON body content goes to `$app->request->body`. **This needs verification before implementation.** --- ## Refactor Steps ### Phase 1 — Create the Task Worker Module **Goal:** A single entry-point script that can run as a standalone process. **File:** `src/task_worker.py` (new) **Acceptance Criteria:** - Starts with `python3 src/task_worker.py --tasks N` (default 1 task) - Prints startup log message showing number of tasks requested and model version - Exits cleanly after processing the requested count ```bash # Default: process 1 task and exit python3 -m src.task_worker # Explicit: process 5 tasks and exit python3 -m src.task_worker --tasks 5 # Also supported via direct script invocation python3 src/task_worker.py --tasks 10 ``` **CLI Implementation:** ```python import argparse parser = argparse.ArgumentParser(description="VideoDetect task worker") parser.add_argument("--tasks", type=int, default=1, help="Number of tasks to process before exiting (default: 1)") args = parser.parse_args() ``` --- ### Phase 2 — API Client Module **Goal:** A thin HTTP client layer for communicating with the Perl API. **File:** `src/api_client.py` (new) **Acceptance Criteria:** - Encapsulates all API calls in one class: `ApiClient(base_url)` - Handles JSON serialization/deserialization automatically - Raises a custom `ApiError` on HTTP errors (with status code and response body) - Returns `None` for 404 responses (no task remaining) from `/nexttask` **API Error handling:** ```python class ApiError(Exception): def __init__(self, status_code: int, message: str): self.status_code = status_code self.message = message super().__init__(f"API error {status_code}: {message}") # Usage patterns: try: response = client.get_next_task("AISCAN") except ApiError as e: if e.status_code == 404: logger.info("No more tasks available") break # exit processing loop raise result = client.submit_results(task_id, assign_key, results_dict) ``` **Methods to implement:** | Method | Endpoint | Returns | Special handling | |--------|----------|---------|-----------------| | `get_next_task(task_type)` | `GET /api/v1/nexttask/{type}` | `{"task": {...}, "assign_key": str} \| None` | Returns None on 404 | | `get_video(video_id)` | `GET /api/v1/video/{id}` | dict with video metadata | Raises ApiError on 404/5xx | | `submit_results(task_id, assign_key, results)` | `POST /api/v1/task/{task}/complete` | dict with response | Passes assign_key + serialized results | **Notes for implementation:** - Use Python's standard library only (`urllib.request`) if possible to avoid adding dependencies. If JSON body submission is needed for the Perl API, check whether `body_parameters->get('results')` in Dancer2 handles raw JSON (it typically does not — it expects form-encoded data). - If form-encoding is required: `requests.post(url, data={'assign_key': key, 'results': json.dumps(results)})` - If JSON body works: `requests.post(url, json={'assign_key': key, 'results': results})` - **Recommendation:** Use the `requests` library (already in worker requirements.txt likely). Document both approaches and implement the one that matches your Dancer2 config. --- ### Phase 3 — Implement the AI Processing Pipeline **Goal:** Extract the core AI processing logic from `orchestrator.py` into a reusable function callable by the task worker. **File:** New or updated module in `src/` (tentatively `src/task_processor.py`) **Acceptance Criteria:** - Takes video metadata dict + file path as input - Returns a results dict matching what will be stored in the DB `results` column - Handles all error cases gracefully (returns errors instead of crashing) - Logs all significant decisions at INFO level **Input:** ```python video = { "id": 1, "file_path": "/data/input/...", "codec": "wmv1", "duration": 102.499, "resolution_w": 640, "resolution_h": 480, "file_size": 19358016, "file_hash": "abc...", } ``` **Output (results dict stored in DB):** ```python { "status": "COMPLETED", # or "FAILED" on error "confidence": 0.87, # video-level confidence score "routing_decision": "MATCH", # MATCH | REVIEW | SKIP "face_count": 142, # total faces detected across all frames "frame_count": 23, # frames extracted and processed "model_version": "v0.0.0-placeholder", "processing_time_seconds": 12.4, "error": None, # error message if FAILED } ``` **Processing flow (extracted from orchestrator.py):** ```python def process_video(video: dict) -> dict: """Run the full AI scan pipeline on a single video. Returns results dict for submission via API. """ start_time = time.time() file_path = video["file_path"] video_id = video["id"] try: # 1. Probe (if metadata is incomplete) prober = VideoProber(timeout=10) metadata = prober.probe(file_path) # 2. Extract frames to scratch scratch = ScratchManager(base_path="/scratch", video_id=str(video_id)) frame_dir = scratch.ensure_frame_dir() sampler = FrameSampler(interval_seconds=30, quality=2) frame_paths = sampler.extract_frames( video_path=file_path, output_dir=str(frame_dir), duration=metadata.duration, resolution=(metadata.resolution_w, metadata.resolution_h) ) # 3. Detect faces detector = FaceDetector(engine_path="/models/face_detector/face_detector.trt") detections_per_frame = detector.detect_faces(frame_paths) total_faces = sum(len(dets) for dets in detections_per_frame) # 4. Crop and classify (aggregate confidences from face crops) classifier = FaceClassifier(engine_path="/models/classifier/classifier.trt") # ... crop paths → classify → get confidence list # 5. Aggregate confidence = aggregate(confidences, strategy="max") # 6. Route routing = router.route(confidence) processing_time = time.time() - start_time return { "status": "COMPLETED", "confidence": round(confidence, 4), "routing_decision": routing, "face_count": total_faces, "frame_count": len(frame_paths), "model_version": get_model_version(), "processing_time_seconds": round(processing_time, 2), "error": None, } except Exception as exc: logger.error("Processing failed for video %d: %s", video_id, exc) return { "status": "FAILED", "confidence": 0.0, "routing_decision": "REVIEW", "face_count": 0, "frame_count": 0, "model_version": get_model_version(), "processing_time_seconds": round(time.time() - start_time, 2), "error": str(exc), } ``` **Notes for implementation:** - Reuse existing classes from `orchestrator.py` (`FaceDetector`, `FaceClassifier`, `FrameSampler`, `ScratchManager`, `VideoProber`, `aggregate`, `router`) without refactoring them — just import and use them. - GPU setup (torch CUDA, GPUMemoryManager) can be done once at module level or lazily inside `process_video()` to avoid overhead per task invocation. - Model loading is expensive (~seconds). Consider lazy initialization or process-level caching via a singleton pattern. --- ### Phase 4 — Wire It Together in task_worker.py **Goal:** Combine the API client, processing pipeline, and CLI entry point into a working script. **Acceptance Criteria:** - Full end-to-end: claim → fetch → process → submit → repeat N times - Exits with code 0 on success, non-zero on unrecoverable errors - Logs startup/shutdown counts (tasks attempted, tasks succeeded, tasks failed) - Graceful shutdown on SIGTERM/SIGINT (finish current task, then exit) **Expected lifecycle log output:** ```json {"timestamp": "...", "level": "INFO", "message": "Task worker starting. Will process 3 task(s)."} {"timestamp": "...", "level": "INFO", "message": "API client configured: base_url=http://localhost:8890"} {"timestamp": "...", "level": "INFO", "message": "Claiming task 1 (video_id=42) for AISCAN"} {"timestamp": "...", "level": "INFO", "message": "Processing video 42: /data/input/..."} {"timestamp": "...", "level": "INFO", "message": "Video 42 complete: C=0.87 routing=MATCH faces=142 frames=23 time=12.4s"} {"timestamp": "...", "level": "INFO", "message": "Submitting results for task 1 via API"} {"timestamp": "...", "level": "INFO", "message": "Claiming task 2 (video_id=99) for AISCAN"} {"timestamp": "...", "level": "INFO", "message": "Processing video 99: /data/input/..."} {"timestamp": "...", "level": "INFO", "message": "Video 99 complete: C=0.31 routing=SKIP faces=0 frames=18 time=8.7s"} {"timestamp": "...", "level": "INFO", "message": "Submitting results for task 2 via API"} {"timestamp": "...", "level": "INFO", "message": "Claiming task 3 (video_id=157) for AISCAN"} {"timestamp": "...", "level": "WARNING", "message": "Processing video 157 failed: ffprobe error: cannot decode stream"} {"timestamp": "...", "level": "INFO", "message": "Submitting FAILED results for task 3 via API"} {"timestamp": "...", "level": "INFO", "message": "No more tasks available (API returned 404)"} {"timestamp": "...", "level": "INFO", "message": "Task worker finished: 3 attempted, 2 succeeded, 1 failed"} ``` **Core loop:** ```python for i in range(1, args.tasks + 1): logger.info("=== Processing task %d/%d ===", i, args.tasks) # Claim a task response = api.get_next_task("AISCAN") if response is None: logger.info("No more tasks available. Done.") break task = response["task"] assign_key = response["assign_key"] video_id = task["video_id"] # Fetch video metadata video = api.get_video(video_id) # Process results = process_video(video) # Submit api.submit_results(task["id"], assign_key, results) logger.info("Finished: %d attempted, %d succeeded, %d failed", total_attempted, total_succeeded, total_failed) ``` --- ### Phase 5 — Update docker-compose.yml **Goal:** Change the worker service from a long-lived `WorkerPool` to a short-lived task process. **Changes to `docker-compose.yml` worker section:** 1. **Update command** to invoke the task worker instead of `main.py`: ```yaml command: python3 -m src.task_worker --tasks 50 ``` 2. **Add restart policy considerations** — since workers are now short-lived, you have two options: - **Option A (simpler):** Run a single worker container with a large `--tasks` count (e.g., 100) so it processes many videos before exiting, then rely on Kubernetes/Cron/external scheduler to restart. - **Option B (more flexible):** Make `--tasks` configurable via environment variable: ```yaml environment: - TASK_COUNT=${TASK_COUNT:-50} command: > python3 -m src.task_worker --tasks ${TASK_COUNT} ``` 3. **Remove the WorkerPool initialization** from `main.py` — it becomes unused (or is removed entirely in a later cleanup). --- ### Phase 6 — Verify API Compatibility **Goal:** Confirm the Dancer2 Perl API correctly handles JSON body submission from Python's `requests` library on `/api/v1/task/:task/complete`. **Acceptance Criteria:** - POST with `Content-Type: application/json` body containing `assign_key` + `results` (as a dict, not a string) returns 200 - POST with form-encoded data also works as fallback - The Perl API correctly stores the results JSON in the database **Test procedure:** ```bash # Step 1: Create a test task manually (via MySQL or via scanner) mysql videodetect -e "INSERT INTO tasks (task_type, video_id, status) VALUES ('AISCAN', 1, 'PENDING');" # Step 2: Claim the task via API curl -s http://localhost:8890/api/v1/nexttask/AISCAN | python3 -m json.tool # Step 3: Submit results (adjust video_id to a real one) curl -X POST http://localhost:8890/api/v1/task/1/complete \ -H "Content-Type: application/json" \ -d '{"assign_key":"worker_999","results":{"confidence":0.87,"routing_decision":"MATCH"}}' | python3 -m json.tool # Step 4: Verify results stored in DB mysql videodetect -e "SELECT id, status, results FROM tasks WHERE id=1;" ``` **If JSON body does NOT work**, update the Dancer2 API to handle JSON explicitly: ```perl post '/api/v1/task/:task/complete' => sub { my $task = route_parameters->get("task"); # Handle both JSON body and form-encoded data my $body; if (request_content_type eq 'application/json') { use Dancer2::Core::Request::Entity; $body = decode_json(request_body); } else { $body = body_parameters->to_hash; } my $assign_key = $body->{assign_key}; my $results = $body->{results}; # ... rest of the handler unchanged }; ``` --- ## Files Modified / Created Summary | File | Action | Description | |------|--------|-------------| | `src/task_worker.py` | **NEW** | CLI entry point: `--tasks N`, loop claim→process→submit | | `src/api_client.py` | **NEW** | HTTP client for the Perl API (`ApiClient`) | | `src/task_processor.py` | **NEW** | `process_video(video_dict)` — core AI pipeline callable standalone | | `docker-compose.yml` | **MODIFY** | Change worker command to invoke task_worker; add TASK_COUNT env var | | `api/app.pl` | **MAYBE MODIFY** | Update `/task/:task/complete` to handle JSON body if Dancer2 doesn't support it natively | | `src/main.py` | **MODIFY (cleanup later)** | Remove or deprecate WorkerPool usage (long-lived process no longer needed) | | `REFACTOR.md` | **CURRENT FILE** | This plan document | --- ## Risks & Considerations 1. **GPU model loading per invocation:** Each new Python process loads the ONNX/TensorRT engine into GPU memory. With `--tasks 1`, this is wasteful. Mitigation: use larger `--tasks` values (e.g., 20-100) so amortization is favorable, or implement a local cache server pattern later. 2. **Permanence of scratch space:** The `ScratchManager` creates per-video temp files under `/scratch`. Since workers are now short-lived, ensure `cleanup=True` is always set (it is by default in the existing code). 3. **API JSON body compatibility:** Dancer2's `body_parameters` may not parse JSON bodies — it expects form-encoded data. This is the highest-risk item. Test Phase 6 early. 4. **Task idempotency:** If a worker crashes between processing and submitting results, the task remains `IN_PROGRESS`. The Perl API prevents re-claiming (the WHERE clause checks `status='PENDING'`). A manual SQL update or a `/api/v1/task/:task/reset` endpoint may be needed for recovery. This is outside the scope of this refactor but worth noting. --- ## Suggested Order of Execution 1. **Phase 6 first** — verify API JSON compatibility (takes 5 minutes, unblocks everything) 2. **Phase 2** — create `api_client.py` with minimal methods 3. **Phase 3** — create `task_processor.py` by extracting from `orchestrator.py` 4. **Phase 1** — create `task_worker.py` with CLI + loop 5. **Phase 4** — wire together and test locally (`python3 -m src.task_worker --tasks 3`) 6. **Phase 5** — update docker-compose.yml and deploy