---
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'
---

<frozen-after-approval reason="human-owned intent -- do not modify unless human renegotiates">

## 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 |

</frozen-after-approval>

## 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)
