--- title: 'Document approval pre-extraction gate (approve/reject endpoints)' type: 'feature' created: '2026-04-13' status: 'done' baseline_commit: '889ec3fb6a99aad15e59627b9e93bfe838d7d49b' context: - _planning/docs/services/textiq-doc-extraction/architecture.md - _planning/docs/services/textiq-admin-portal-api/architecture.md --- ## Intent **Problem:** Uploaded documents immediately trigger extraction, consuming compute before a human reviewer decides whether the file should be processed. There is no gate between upload and extraction. **Approach:** Make approval a pre-extraction gate. Upload stores the file and creates a `pending` document but does **not** trigger the DAG. Two new endpoints (`POST /{document_id}/approve` and `POST /{document_id}/reject`) let a reviewer decide. Approve dispatches the extraction DAG; reject deletes the file from SeaweedFS and marks the document rejected. Existing documents are backfilled to `approved` so in-flight/completed work is unaffected. ## Boundaries & Constraints **Always:** - HMAC auth via `HMACVerificationMiddleware` + `rls_connection(tenant_id)` on every new endpoint (same pattern as `domain_terms.py`, `tags.py`). - `tenant_id` and `user_id` from HMAC context headers (`request.state.hmac_ctx`) — not from request body. Body contains only `note` (approve) or `rejection_reason` (reject). - Raw SQL inside `rls_connection` context manager — no ORM writes in plugin layer. - Backfill all existing documents to `approval_status='approved'` (grandfathered). - New uploads start at `approval_status='pending'` with no DAG trigger. - Approve endpoint triggers DAG only when `status='pending'` (skip if already processing/completed). - Reject endpoint deletes file from SeaweedFS (using existing `_delete_s3_folder()` pattern). - Transitions are reversible: approve <-> reject. Each call overwrites all 5 approval columns atomically. **Ask First:** - Any change to the DAG task dependency graph beyond adding the defensive `approval_status` check. - Adding new columns beyond the 5 specified (`approval_status`, `reviewed_by`, `reviewed_at`, `rejection_reason`, `reviewed_note`). **Never:** - Import from `src.*` in the plugin layer (Airflow api-server process uses a separate Python environment). - Put `tenant_id` or `reviewed_by` in the request body — these come from HMAC context. - Add presigned URL endpoints (Feature 3 deferred). - Break backward compatibility for existing completed/in-flight documents. ## I/O & Edge-Case Matrix | Scenario | Input / State | Expected Output / Behavior | Error Handling | |----------|--------------|---------------------------|----------------| | Approve pending doc | `POST /{id}/approve`, body `{note?}` | 200, `approval_status=approved`, DAG triggered | N/A | | Reject pending doc | `POST /{id}/reject`, body `{rejection_reason}` 10-500 chars | 200, `approval_status=rejected`, S3 file deleted, no DAG | N/A | | Approve already-completed (backfilled) | doc at `status=completed, approval_status=approved` | 200, approval cols updated, DAG **not** re-triggered | N/A | | Reverse: approved -> rejected | `POST /{id}/reject` on approved doc | 200, cols overwritten, S3 deleted (if status still pending) | N/A | | Reverse: rejected -> approved | `POST /{id}/approve` on rejected doc (status=pending) | 200, cols overwritten, DAG triggered | N/A | | Doc not found / cross-tenant | Valid UUID, wrong tenant | RLS hides row | 404 `NOT_FOUND` | | Invalid UUID | `POST /not-a-uuid/approve` | Validation | 422 | | Reject reason too short | `rejection_reason` < 10 chars | Pydantic | 422 | | Reject reason too long | `rejection_reason` > 500 chars | Pydantic | 422 | | Missing HMAC | No signature headers | Middleware | 401 | | DAG task on non-approved doc | Worker starts but `approval_status != approved` | Log skip, return (no-op) | N/A | ## Code Map - `airflow/plugins/api/upload.py` -- Add approve/reject routes to existing `upload_app`; remove `trigger_dag()` from upload handler; add `approval_status='pending'` to INSERT - `airflow/plugins/api/rls.py` -- Existing `rls_connection()` (reuse) - `airflow/plugins/api/responses.py` -- Existing `error_response()`/`success_response()` (reuse) - `airflow/plugins/api/dag.py` -- Existing `trigger_dag()` (reuse from approve handler) - `airflow/plugins/api/s3.py` -- Existing S3 client (reuse for reject cleanup) - `src/db/migrations/versions/20260413_100000_add_document_approval_fields.py` -- **NEW** migration - `src/db/models.py:68-109` -- Add 5 approval columns to `Document` ORM ## Tasks & Acceptance **Execution:** - [x] `src/db/migrations/versions/20260413_100000_add_document_approval_fields.py` -- NEW: 5 columns (`approval_status`, `reviewed_by`, `reviewed_at`, `rejection_reason`, `reviewed_note`), index, backfill all existing docs to `approved` - [x] `src/db/models.py` -- ADD 5 mapped columns to `Document` class (needed by DAG tasks using ORM) - [x] `airflow/plugins/api/upload.py` -- ADD `POST /{document_id}/approve` and `POST /{document_id}/reject` routes to existing `upload_app`. Pydantic models: `ApproveRequest(note?)`, `RejectRequest(rejection_reason)`. `user_id`/`tenant_id` from `hmac_ctx`. Approve: update cols + `trigger_dag` when `status='pending'`. Reject: update cols + `_delete_s3_folder()`. REMOVE `trigger_dag()` + DAG conf from upload handler. ADD `approval_status='pending'` to INSERT. - [x] `tests/api/test_document_approval.py` -- NEW: happy paths, 404, 422, cross-tenant, reversals, HMAC, S3 cleanup on reject **Acceptance Criteria:** - Given a newly uploaded document, when checking its DB row, then `approval_status='pending'` and no DAG was triggered - Given a pending document, when `POST /{id}/approve` with valid HMAC, then `approval_status='approved'`, DAG triggered, response 200 - Given a pending document, when `POST /{id}/reject` with valid reason, then `approval_status='rejected'`, S3 file deleted, no DAG, response 200 - Given a rejected document, when `POST /{id}/approve`, then approval cols overwritten, DAG triggered - Given tenant B's document, when tenant A calls approve, then 404 (RLS) - Given rejection reason < 10 chars, when `POST /{id}/reject`, then 422 - Given no HMAC headers, when calling any approval endpoint, then 401 - Given existing completed docs after migration, then `approval_status='approved'` (backfilled) ## Design Notes **Request body is minimal.** `tenant_id` and `user_id` (used as `reviewed_by`) come from `request.state.hmac_ctx` — same as `tags.py` and `domain_terms.py`. Approve body: `{"note": "optional"}`. Reject body: `{"rejection_reason": "required 10-500 chars"}`. **Routes on `upload_app`.** Approve/reject routes are added directly to `upload_app` (already at `/api/v1/documents`). No new app registration needed. Routes: `/upload` (POST), `/{id}` (DELETE), `/{id}/approve` (POST), `/{id}/reject` (POST). **DAG conf reconstruction.** Approve handler reads `file_path` from the documents row to build `conf = {document_id, bucket, object_key, tenant_id, node_id, correlation_id, request_id}` — identical payload to current upload flow. **S3 cleanup on reject.** Reuse `_delete_s3_folder(tenant_id, node_id, document_id)` from `upload.py`. Only delete when file hasn't been extracted yet (`status='pending'`). If doc is already `completed` (reversal scenario), skip S3 deletion — extracted data is in Milvus/PG. ## Verification **Commands:** - `cd /home/nguyen/code/textiq-doc-extraction && alembic upgrade head` -- expected: migration applies, `approval_status` column exists with index - `python -m pytest tests/api/test_document_approval.py -v` -- expected: all tests pass **Manual checks:** - `SELECT approval_status FROM documents WHERE status='completed' LIMIT 5` — all show `approved` - `SELECT approval_status FROM documents WHERE status='pending' LIMIT 5` — all show `approved` (backfilled)