--- title: 'Hard delete document endpoint in Airflow plugin' type: 'feature' created: '2026-04-09' status: 'done' baseline_commit: 'acd001b' context: - '_planning/docs/services/textiq-doc-extraction/architecture.md' --- ## Intent **Problem:** The Airflow plugin has an upload endpoint but no delete. APA needs to hard-delete documents and all associated data (chunks, vectors, S3 files) via the Airflow plugin API. **Approach:** Add a `DELETE /api/v1/documents/{document_id}` endpoint to the Airflow plugin that cascade-deletes from Milvus, PostgreSQL (`data_hierarchical_nodes`, `domain_term_dictionary`, `documents`), and S3 in a single request. Follow existing plugin patterns (sync FastAPI, HMAC auth, RLS). ## Boundaries & Constraints **Always:** - HMAC authentication + RLS tenant isolation (same as upload endpoint) - Verify document belongs to the requesting tenant before deleting - Delete from Milvus first, then PostgreSQL, then S3 (least recoverable last) - Return deleted counts in response **Ask First:** - Whether to delete `domain_term_dictionary` entries linked via `source_document_id` (currently included in cascade) **Never:** - Soft delete -- this is permanent hard delete - Delete documents belonging to other tenants - Skip Milvus/S3 cleanup even if DB delete succeeds ## I/O & Edge-Case Matrix | Scenario | Input / State | Expected Output / Behavior | Error Handling | |----------|--------------|---------------------------|----------------| | Happy path | Valid document_id owned by tenant | 200 with deletion counts | N/A | | Not found | document_id doesn't exist | 404 NOT_FOUND | Return error response | | Wrong tenant | document_id belongs to another tenant | 404 NOT_FOUND (no leak) | RLS blocks access | | Invalid UUID | Malformed document_id | 422 validation error | Pydantic/path validation | | Milvus failure | Milvus unavailable | 500, no DB/S3 changes | Fail fast, nothing deleted | | S3 file missing | Document row exists but S3 key gone | 200, log warning, continue | Don't fail on missing S3 object | ## Code Map - `airflow/plugins/api/upload.py` -- Reference pattern for FastAPI app, middleware, HMAC, RLS - `airflow/plugins/api/rls.py` -- `rls_connection()` context manager for tenant-scoped transactions - `airflow/plugins/api/milvus.py` -- Milvus client; needs new `delete_chunks_by_document_id()` method - `airflow/plugins/api/s3.py` -- S3 client with `delete_object` and `list_objects_v2` - `airflow/plugins/api/responses.py` -- `success_response()` / `error_response()` helpers - `airflow/plugins/__init__.py` -- Plugin registration of FastAPI sub-apps - `src/db/models.py` -- `Document` model (table: `documents`), `DomainTermDictionary` (table: `domain_term_dictionary`) ## Tasks & Acceptance **Execution:** - [x] `airflow/plugins/api/milvus.py` -- Add `delete_chunks_by_document_id(document_id, tenant_id, node_id)` using filter-based `client.delete(filter=...)` (NOT query+delete, to avoid default limit truncation). Filter must include `document_id`, `tenant_id`, AND `node_id`. - [x] `airflow/plugins/api/upload.py` -- Add `DELETE /{document_id}` route to existing upload_app. Sequence: validate UUID, HMAC auth, **verify document exists in PG first** (read-only RLS query), then delete Milvus vectors, then delete PG chunks (using JSONB key `document_id` NOT `source_document_id`) + domain terms + document row, then S3 files (with pagination for >1000 objects). - [x] `airflow/plugins/__init__.py` -- No change needed: upload_app already registered at `/api/v1/documents` **Acceptance Criteria:** - Given a valid document_id owned by the tenant, when DELETE is called, then the document, all chunks, domain terms, vectors, and S3 files are permanently removed - Given a document_id not owned by the tenant, when DELETE is called, then 404 is returned and no data is modified - Given Milvus is unavailable, when DELETE is called, then 500 is returned and no PostgreSQL/S3 data is deleted ## Spec Change Log **Iteration 1 (review round 1):** - **Trigger:** Three review findings — wrong JSONB key, existence check ordering, Milvus query limit. - **Amended:** Tasks section — (1) JSONB metadata key changed from `source_document_id` to `document_id` to match how DAGs actually store document references in `data_hierarchical_nodes`. (2) Existence check moved before Milvus delete to prevent orphaning vectors on 404. (3) Milvus delete changed to filter-based `client.delete(filter=...)` instead of query+delete to avoid default limit (~16384) truncation. (4) Added `node_id` to Milvus filter for consistency with PG queries. (5) S3 listing must paginate for >1000 objects. - **Known-bad state avoided:** Chunks never deleted (wrong JSONB key); Milvus vectors orphaned on non-existent document; large documents partially cleaned. - **KEEP:** Overall structure (HMAC + RLS + _EarlyReturn), S3 prefix-based cleanup, response format with deletion counts, adding delete to existing upload_app. ## Verification **Manual checks:** - Call DELETE endpoint with valid HMAC headers and confirm document + chunks + vectors + S3 files are gone - Verify RLS prevents cross-tenant deletion ## Suggested Review Order **Milvus deletion** - Filter-based delete scoped by document_id + tenant_id + node_id, avoids query limit [`milvus.py:76`](../../airflow/plugins/api/milvus.py#L76) **Delete endpoint** - Entry point: UUID validation, HMAC auth extraction, existence check before any mutation [`upload.py:214`](../../airflow/plugins/api/upload.py#L214) - Milvus fail-fast: abort with 500 if vectors can't be deleted, before touching PG [`upload.py:244`](../../airflow/plugins/api/upload.py#L244) - PG cascade: chunks by JSONB `document_id` key, domain terms, then document row [`upload.py:255`](../../airflow/plugins/api/upload.py#L255) - S3 paginated cleanup: handles >1000 objects with continuation tokens and batched deletes [`upload.py:297`](../../airflow/plugins/api/upload.py#L297) **Supporting** - Imports: UUID, get_docstore_namespace, delete_chunks_by_document_id, rls_connection [`upload.py:18`](../../airflow/plugins/api/upload.py#L18)