diff --git a/src/api/errors.py b/src/api/errors.py index 7ec6fdc..dcbff82 100644 --- a/src/api/errors.py +++ b/src/api/errors.py @@ -17,7 +17,11 @@ from src.application.auth.errors import ( TenantInactiveError, ) from src.application.domains.errors import DomainAlreadyExistsError, UnknownDomainError -from src.application.files.errors import FileTooLargeError, InvalidUploadError +from src.application.files.errors import ( + FileTooLargeError, + InvalidUploadError, + SourceFileNotFoundError, +) from src.application.ingestion.errors import ( ChunkLimitExceededError, DocumentParseError, @@ -27,6 +31,7 @@ from src.application.ingestion.errors import ( PointIndexingError, UnsupportedSourceTypeError, ) +from src.application.points.errors import PointVersionConflictError from src.application.points.point import PointNotFoundError logger = structlog.get_logger(__name__) @@ -45,6 +50,10 @@ _MAPPING: tuple[tuple[type[Exception], int, str], ...] = ( # 404, never 403: a cross-tenant point id must be indistinguishable from a # nonexistent one, or the API becomes an existence oracle (ADR-0016). (PointNotFoundError, status.HTTP_404_NOT_FOUND, "not_found"), + (SourceFileNotFoundError, status.HTTP_404_NOT_FOUND, "not_found"), + # Not "the version guard fired once" — that is retried. This is the service + # giving up after repeated re-plans, i.e. a genuinely contended point. + (PointVersionConflictError, status.HTTP_409_CONFLICT, "conflict"), (UnknownDomainError, status.HTTP_400_BAD_REQUEST, "unknown_domain"), (DomainAlreadyExistsError, status.HTTP_409_CONFLICT, "conflict"), (DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"), diff --git a/src/api/routers/files.py b/src/api/routers/files.py index d873dbd..4aee87a 100644 --- a/src/api/routers/files.py +++ b/src/api/routers/files.py @@ -1,4 +1,4 @@ -"""`POST /v1/files`, `GET /v1/files/{file_id}` (ADR-0008). +"""`POST /v1/files`, `GET /v1/files/{file_id}`, `DELETE /v1/files/{file_id}` (ADR-0008). Routes adapt HTTP to `application/files` calls; they do not parse, hash, touch MinIO/Qdrant, or otherwise carry ingestion business logic (ADR-0015). @@ -13,9 +13,10 @@ from fastapi import APIRouter, Depends, Form, HTTPException, Response, UploadFil from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.api.dependencies.auth import require_scope -from src.api.schemas.files import FileStatusResponse, FileUploadResponse +from src.api.schemas.files import FileDeleteResponse, FileStatusResponse, FileUploadResponse from src.api.schemas.points import DEFAULT_PAGE_SIZE, LimitQuery, PointListResponse from src.application.auth.context import AuthContext +from src.application.files.deletion import delete_source_file from src.application.files.status import get_file_status from src.application.files.upload import upload_source_file from src.application.points.queries import list_file_points @@ -40,6 +41,7 @@ router = APIRouter(prefix="/files", tags=["files"]) _RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))] _RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))] +_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))] _SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)] _ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)] _PointStorageDep = Annotated[PointStorage, Depends(get_point_storage)] @@ -100,6 +102,35 @@ async def get_file( return FileStatusResponse.from_result(result) +@router.delete("/{file_id}") +async def delete_file( + file_id: uuid.UUID, + auth: _RequirePointsWrite, + sessionmaker: _SessionmakerDep, + repository: _PointRepositoryDep, +) -> FileDeleteResponse: + """Soft-delete a file: every active point, then the `source_files` row. + + Gated on `points:write` rather than `files:write` for the same reason as the + listing above — the data this destroys is points. Nothing is removed from + Qdrant (ADR-0002); the points are flagged inactive and the row is marked + `soft_deleted`, which is also what makes a later re-upload of the same bytes + ingest afresh instead of matching the duplicate path. + + Deleting an already-deleted file is a success reporting `0` points. + """ + points_soft_deleted = await delete_source_file( + sessionmaker, + repository, + tenant_id=auth.tenant_id, + source_file_id=file_id, + actor=f"api_key:{auth.api_key_id}", + ) + return FileDeleteResponse( + file_id=file_id, status="soft_deleted", points_soft_deleted=points_soft_deleted + ) + + @router.get("/{file_id}/points") async def list_points_for_file( file_id: uuid.UUID, diff --git a/src/api/routers/points.py b/src/api/routers/points.py index 9b9530b..d35425d 100644 --- a/src/api/routers/points.py +++ b/src/api/routers/points.py @@ -1,4 +1,4 @@ -"""`/v1/points` read paths (ADR-0002, ADR-0008). +"""`/v1/points` read and soft-delete paths (ADR-0002, ADR-0008). Routes adapt HTTP to `application/points` calls. They build no Qdrant filters and hold no CRUD semantics (ADR-0015), and they never read a tenant from the @@ -31,6 +31,7 @@ from src.api.schemas.points import ( PointSearchResponse, ) from src.application.auth.context import AuthContext +from src.application.points.deletion import soft_delete_point from src.application.points.queries import ( count_points, get_point, @@ -43,6 +44,7 @@ from src.bootstrap.dependencies import get_point_repository router = APIRouter(prefix="/points", tags=["points"]) _RequirePointsRead = Annotated[AuthContext, Depends(require_scope("points:read"))] +_RequirePointsWrite = Annotated[AuthContext, Depends(require_scope("points:write"))] _PointRepositoryDep = Annotated[PointRepository, Depends(get_point_repository)] @@ -132,3 +134,29 @@ async def get_tenant_point( repository, tenant_id=auth.tenant_id, point_id=point_id, with_vectors=with_vectors ) return PointResponse.from_point(point) + + +@router.delete("/{point_id}") +async def delete_tenant_point( + point_id: uuid.UUID, + auth: _RequirePointsWrite, + repository: _PointRepositoryDep, +) -> PointResponse: + """Soft-delete one point and relink its neighbours around the gap. + + The point is never removed from Qdrant (ADR-0002): it is flagged + `is_active=false` with `deleted_at` set, and its old neighbours are pointed + at each other in the same batch, so context-window expansion never walks + into it. + + Deleting an already-inactive point is a no-op success rather than a `404` — + the response is the point as it stands, so the resulting `version` and + `deleted_at` are visible either way. + """ + point = await soft_delete_point( + repository, + tenant_id=auth.tenant_id, + point_id=point_id, + actor=f"api_key:{auth.api_key_id}", + ) + return PointResponse.from_point(point) diff --git a/src/api/schemas/files.py b/src/api/schemas/files.py index a62c921..efc5146 100644 --- a/src/api/schemas/files.py +++ b/src/api/schemas/files.py @@ -28,6 +28,20 @@ class FileUploadResponse(BaseModel): ) +class FileDeleteResponse(BaseModel): + """What `DELETE /v1/files/{file_id}` did. + + `points_soft_deleted` is reported rather than left implicit because the + delete is a soft one: nothing is removed from Qdrant, and the count is the + only way a caller can tell "deactivated 40 points" from "the file was + already deleted" — both of which are successes. + """ + + file_id: uuid.UUID + status: str + points_soft_deleted: int + + class FileStatusResponse(BaseModel): file_id: uuid.UUID source_filename: str diff --git a/src/application/files/deletion.py b/src/application/files/deletion.py new file mode 100644 index 0000000..a774a0b --- /dev/null +++ b/src/application/files/deletion.py @@ -0,0 +1,81 @@ +"""`DELETE /v1/files/{file_id}` — retire a file and deactivate its points. + +Two stores have to agree here, and the phase boundaries are the same ones +ingestion uses (ADR-0017): a short Postgres transaction to authorize, then the +Qdrant work with **no session held**, then a short transaction to record the +outcome. Holding a session across the sweep would pin a pool connection for the +length of a multi-page delete. + +The order — points first, Postgres second — is deliberate. If the sweep dies +half way, the row stays `active` and a retried `DELETE` finishes the job, since +the sweep only ever looks at points that are still active. The reverse order +would leave a row marked deleted while its points are still live and still +retrievable by the agent, which is the failure that actually matters. +""" + +import uuid +from datetime import UTC, datetime +from time import perf_counter + +import structlog +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.files.errors import SourceFileNotFoundError +from src.application.points.deletion import soft_delete_file_points +from src.application.ports.point_repository import PointRepository +from src.infrastructure.postgres.repositories import source_files as source_files_repo + +logger = structlog.get_logger(__name__) + + +async def delete_source_file( + sessionmaker: async_sessionmaker[AsyncSession], + repository: PointRepository, + *, + tenant_id: uuid.UUID, + source_file_id: uuid.UUID, + actor: str, +) -> int: + """Soft-delete a file: every active point, then the `source_files` row. + + Returns how many points the sweep deactivated. Raises + `SourceFileNotFoundError` (`404`) when the file is not this tenant's — the + check happens before anything is written, so a probe for another tenant's + file id cannot deactivate a single point. + + Idempotent: a second call finds no active points and a row already marked + `soft_deleted`, and returns `0`. + """ + started = perf_counter() + async with sessionmaker() as session: + source_file = await source_files_repo.get_by_id( + session, tenant_id=tenant_id, source_file_id=source_file_id + ) + if source_file is None: + raise SourceFileNotFoundError(f"file {source_file_id} not found") + + points_soft_deleted = await soft_delete_file_points( + repository, tenant_id=tenant_id, file_id=source_file_id, actor=actor + ) + + async with sessionmaker() as session: + source_file = await source_files_repo.get_by_id( + session, tenant_id=tenant_id, source_file_id=source_file_id + ) + if source_file is None: + raise SourceFileNotFoundError(f"file {source_file_id} not found") + source_files_repo.mark_soft_deleted(source_file, deleted_at=datetime.now(UTC)) + await session.commit() + + logger.info( + "files.soft_deleted", + tenant_id=str(tenant_id), + file_id=str(source_file_id), + points_soft_deleted=points_soft_deleted, + actor=actor, + # End to end, including both Postgres transactions. Comparing it with + # the sweep's own `duration_ms` on `points.file_soft_deleted` is what + # separates a slow Qdrant from a slow database. + duration_ms=round((perf_counter() - started) * 1000, 2), + ) + return points_soft_deleted diff --git a/src/application/files/errors.py b/src/application/files/errors.py index b58631c..7153646 100644 --- a/src/application/files/errors.py +++ b/src/application/files/errors.py @@ -15,3 +15,11 @@ class InvalidUploadError(FilesError): class FileTooLargeError(FilesError): """The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`.""" + + +class SourceFileNotFoundError(FilesError): + """No such source file *within the requesting tenant*. Maps to `404`. + + Same non-disclosure rule as points (ADR-0016): a cross-tenant file id and a + nonexistent one are indistinguishable to the caller, so this is never `403`. + """ diff --git a/src/application/points/__init__.py b/src/application/points/__init__.py index 5cb1ca8..558623e 100644 --- a/src/application/points/__init__.py +++ b/src/application/points/__init__.py @@ -5,9 +5,11 @@ dispatches payload construction, batching, bounded-concurrency upserts, and the post-success soft-delete sweep. `build_chunk_payload` and the batching helpers stay internal, exported mainly for their own unit tests. -Direct `/v1/points` CRUD (single-point edits, reordering, keyword search) is -plan 002's surface, not this package's — plan 001 scopes it to "the reusable -service layer required by ingestion". +The `/v1/points` surface lives here too, in its own modules with their own +entry points: `queries.py` for the read paths and `deletion.py` for soft delete +with neighbour relinking. They share this package because they share ADR-0001's +payload schema, not because they share a caller — `index_chunks` writes a whole +file at once, while those serve one admin edit at a time. """ from src.application.points.indexing import IndexingResult, index_chunks diff --git a/src/application/points/deletion.py b/src/application/points/deletion.py new file mode 100644 index 0000000..e16f900 --- /dev/null +++ b/src/application/points/deletion.py @@ -0,0 +1,253 @@ +"""Soft delete for `/v1/points` and for a whole file's points (ADR-0002). + +The caller-facing entry points are `soft_delete_point` and +`soft_delete_file_points`. Routers call these; `patches_for_removal` and the +planning helpers stay internal, because getting a delete right is exactly the +composition a caller should not have to reassemble: read the point, load its +neighbours, compute the patches still missing, send them in **one** batch, +verify they landed, and retry against fresh versions if they did not. + +Why the retry exists. Qdrant has no multi-point transaction, so a batch whose +second operation loses a version race applies its first operation anyway — and +a filtered `set_payload` that matched nothing still reports success. That +combination means "did my write land?" is only answerable by reading back, and +a single-shot delete would be able to leave the deactivation applied and a +neighbour's pointer stale. Since `patches_for_removal` plans from current state +towards a fixed end state, simply re-planning emits precisely the patches that +did not land, so the loop converges instead of re-doing work. Only exhausting +the attempts raises `PointVersionConflictError` (`409`). + +Deleting an already-inactive point falls out of the same machinery rather than +needing a special case: its neighbours were relinked by the first delete, so the +plan is empty and the call is a no-op success — not a `404`, and not a second +relink. +""" + +import uuid +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from time import perf_counter + +import structlog + +from src.application.points.errors import PointVersionConflictError +from src.application.points.point import Point, PointNotFoundError +from src.application.points.relinking import ( + neighbour_ids, + patch_for_deactivation, + patches_for_removal, +) +from src.application.ports.point_repository import PayloadPatch, PointRepository + +logger = structlog.get_logger(__name__) + +# Three plan-apply rounds, then a final verifying plan. Each round only re-emits +# what a concurrent writer displaced, so a caller that legitimately needs more +# than this is contending on the same points continuously and deserves the +# `409` rather than an unbounded loop inside a request. +_MAX_ATTEMPTS = 3 + +# One sweep page. Matches ADR-0002's 100-operation batch cap, so a page of +# points is always expressible as a single `points/batch` request. +_SWEEP_BATCH_SIZE = 100 + +# A hard ceiling on sweep rounds, so a file being concurrently re-ingested while +# it is deleted cannot spin here for the life of the request. +_MAX_SWEEP_ROUNDS = 1_000 + + +def _elapsed_ms(started: float) -> float: + """Wall-clock milliseconds since `started` (ADR-0011's `duration_ms`). + + Worth carrying on these events even though the relinking itself is O(1): + what a delete actually spends is Qdrant round trips, and the whole-file + sweep spends a number of them proportional to the file's length. Timing the + operation is the only way to tell a slow store from a contended one, which + `rounds` on the same event then disambiguates. + """ + return round((perf_counter() - started) * 1000, 2) + + +async def soft_delete_point( + repository: PointRepository, + *, + tenant_id: uuid.UUID, + point_id: uuid.UUID, + actor: str, +) -> Point: + """Deactivate one point and relink its neighbours around the gap. + + Returns the point as it now stands. Raises `PointNotFoundError` (`404`) if + it is not this tenant's — the same non-disclosure rule the read paths + follow — or `PointVersionConflictError` (`409`) if concurrent writers keep + displacing the plan. + """ + started = perf_counter() + rounds = 0 + point, patches = await _plan_removal( + repository, tenant_id=tenant_id, point_id=point_id, actor=actor + ) + + for _ in range(_MAX_ATTEMPTS): + if not patches: + break + await repository.apply_patches(tenant_id=tenant_id, patches=patches) + rounds += 1 + # The next plan doubles as verification: anything that did not land is + # still missing from the end state and comes back as a patch. + point, patches = await _plan_removal( + repository, tenant_id=tenant_id, point_id=point_id, actor=actor + ) + + if patches: + logger.warning( + "points.soft_delete.conflict", + tenant_id=str(tenant_id), + point_id=str(point_id), + file_id=str(point.file_id), + unsettled_points=[str(patch.point_id) for patch in patches], + rounds=rounds, + duration_ms=_elapsed_ms(started), + ) + raise PointVersionConflictError( + f"point {point_id} could not be soft-deleted under concurrent modification" + ) + + if rounds: + logger.info( + "points.soft_deleted", + tenant_id=str(tenant_id), + point_id=str(point_id), + file_id=str(point.file_id), + version=point.version, + actor=actor, + # `rounds` is 1 unless a concurrent writer forced a re-plan, so a + # rising value here is contention, not slow relinking. + rounds=rounds, + duration_ms=_elapsed_ms(started), + ) + return point + + +async def soft_delete_file_points( + repository: PointRepository, + *, + tenant_id: uuid.UUID, + file_id: uuid.UUID, + actor: str, +) -> int: + """Deactivate every active point of one file, in batches. + + No relinking: the whole file leaves the sequence at once, so no surviving + active point can be left pointing at a deactivated one, and the chain is + preserved intact for whoever reads the deleted file later. + + Returns how many points were active when the sweep reached them. Each round + re-lists from the start rather than paging with a cursor — deactivated + points drop straight out of the default listing, so the listing itself is + the progress check, and a round that attempts the exact same ids as the one + before it made no progress and raises `PointVersionConflictError`. + """ + started = perf_counter() + swept: set[uuid.UUID] = set() + previous_attempt: frozenset[uuid.UUID] = frozenset() + rounds = 0 + + for _ in range(_MAX_SWEEP_ROUNDS): + page = await repository.list_by_file( + tenant_id=tenant_id, file_id=file_id, limit=_SWEEP_BATCH_SIZE + ) + if not page.points: + if swept: + logger.info( + "points.file_soft_deleted", + tenant_id=str(tenant_id), + file_id=str(file_id), + points_soft_deleted=len(swept), + actor=actor, + # Two Qdrant round trips per round, so this is the delete + # path whose cost tracks the size of the file. + rounds=rounds, + duration_ms=_elapsed_ms(started), + ) + return len(swept) + + attempt = frozenset(point.point_id for point in page.points) + if attempt == previous_attempt: + logger.warning( + "points.file_soft_delete.conflict", + tenant_id=str(tenant_id), + file_id=str(file_id), + unsettled_points=[str(point_id) for point_id in sorted(attempt, key=str)], + rounds=rounds, + duration_ms=_elapsed_ms(started), + ) + raise PointVersionConflictError( + f"file {file_id} could not be soft-deleted under concurrent modification" + ) + previous_attempt = attempt + + now = datetime.now(UTC) + await repository.apply_patches( + tenant_id=tenant_id, + patches=[patch_for_deactivation(point, actor=actor, now=now) for point in page.points], + ) + swept |= attempt + rounds += 1 + + logger.warning( + "points.file_soft_delete.conflict", + tenant_id=str(tenant_id), + file_id=str(file_id), + reason="sweep_rounds_exhausted", + rounds=rounds, + duration_ms=_elapsed_ms(started), + ) + raise PointVersionConflictError(f"file {file_id} still had active points after the sweep") + + +async def _plan_removal( + repository: PointRepository, + *, + tenant_id: uuid.UUID, + point_id: uuid.UUID, + actor: str, +) -> tuple[Point, tuple[PayloadPatch, ...]]: + point = await repository.get(tenant_id=tenant_id, point_id=point_id) + if point is None: + raise PointNotFoundError(f"point {point_id} not found") + + neighbours = await _load_neighbours(repository, tenant_id=tenant_id, point=point) + _warn_on_missing_neighbours(point, neighbours, tenant_id=tenant_id) + patches = patches_for_removal(point, neighbours, actor=actor, now=datetime.now(UTC)) + return point, patches + + +async def _load_neighbours( + repository: PointRepository, *, tenant_id: uuid.UUID, point: Point +) -> dict[uuid.UUID, Point]: + wanted: Sequence[uuid.UUID] = neighbour_ids(point) + if not wanted: + return {} + found = await repository.get_many(tenant_id=tenant_id, point_ids=wanted) + return {neighbour.point_id: neighbour for neighbour in found} + + +def _warn_on_missing_neighbours( + point: Point, neighbours: Mapping[uuid.UUID, Point], *, tenant_id: uuid.UUID +) -> None: + """A pointer naming a point that is not there means the chain is already broken. + + Worth a log line rather than an exception: the delete can still complete the + part of the relink that does exist, and refusing would leave the caller with + a point it cannot remove through any endpoint. + """ + missing = [pointer for pointer in neighbour_ids(point) if pointer not in neighbours] + if missing: + logger.warning( + "points.relink.neighbour_missing", + tenant_id=str(tenant_id), + point_id=str(point.point_id), + file_id=str(point.file_id), + missing_neighbours=[str(pointer) for pointer in missing], + ) diff --git a/src/application/points/errors.py b/src/application/points/errors.py new file mode 100644 index 0000000..011e868 --- /dev/null +++ b/src/application/points/errors.py @@ -0,0 +1,19 @@ +"""Mutation failures for the `/v1/points` write paths (ADR-0002). + +No HTTP knowledge here — `src/api/errors.py` owns the status mapping. Absence +lives on `PointNotFoundError` in `point.py`, next to the model whose read paths +raise it; this module is for the failures only a *write* can produce. +""" + + +class PointVersionConflictError(Exception): + """A version-guarded write could not be landed against a moving target. + + Raised when the service has re-read, recomputed, and re-applied its patches + the allowed number of times and the desired state still has not settled — + something else is writing the same points concurrently. Maps to `409`. + + This is not "the guard fired once": a single stale guard is expected and is + retried, because Qdrant reports success for a filtered `set_payload` that + matched nothing. It means the retries were exhausted. + """ diff --git a/src/application/points/relinking.py b/src/application/points/relinking.py new file mode 100644 index 0000000..2e09ee7 --- /dev/null +++ b/src/application/points/relinking.py @@ -0,0 +1,135 @@ +"""Adjacency-pointer maintenance for a point leaving a file's sequence. + +ADR-0001 keeps `previous_chunk_id`/`next_chunk_id` on every point so ADR-0003's +context-window expansion can walk a file in O(1) steps. ADR-0002 makes keeping +them correct an obligation of every operation that changes a point's position: +a partial relink is a defect, not a degraded-but-acceptable outcome. + +The function below is the primitive that obligation reduces to. It is pure, and +it is written as **"what is still missing between the state I just read and the +state I want"** rather than "the patches a delete implies". That framing is what +makes the caller's retry loop correct: re-planning after a partial apply emits +exactly the patches that did not land, and re-planning after a completed delete +emits nothing at all. The three cases the plan calls out — a normal delete, a +second delete of an already-inactive point, and recovery from a half-applied +batch — are then one code path instead of three. + +Note what is deliberately *not* patched: the departing point's own +`previous_chunk_id`/`next_chunk_id`. Nothing active points at it once its +neighbours are relinked, so those pointers are unreachable rather than stale, +and leaving them records where the point sat — which is what a later restore or +an audit reader would need. `src/application/points/deletion.py` relies on that +when it re-plans: the departing point's pointers are the only surviving record +of which two neighbours have to be joined. +""" + +import uuid +from collections.abc import Mapping +from datetime import datetime + +from src.application.points.point import Point +from src.application.ports.point_repository import PayloadPatch + + +def _optional_id(value: uuid.UUID | None) -> str | None: + return str(value) if value is not None else None + + +def _provenance(point: Point, *, actor: str, now: datetime) -> dict[str, object]: + """The fields every mutation writes: who, when, and the next version. + + Bumping `version` on a relinked *neighbour* is intentional. The neighbour's + payload really did change, so a concurrent editor holding the old version + must get a `409` rather than overwrite the pointer we just fixed. + """ + return { + "updated_at": now.isoformat(), + "updated_by": actor, + "version": point.version + 1, + } + + +def neighbour_ids(point: Point) -> tuple[uuid.UUID, ...]: + """The ids `patches_for_removal` needs loaded, skipping the nulls.""" + return tuple( + pointer for pointer in (point.previous_chunk_id, point.next_chunk_id) if pointer is not None + ) + + +def patches_for_removal( + point: Point, + neighbours: Mapping[uuid.UUID, Point], + *, + actor: str, + now: datetime, +) -> tuple[PayloadPatch, ...]: + """The patches still needed to remove `point` from its file's sequence. + + Returns an empty tuple when the removal is already complete, which the + caller reads as both "converged" and "this was a no-op". + + A neighbour absent from `neighbours` is skipped rather than patched blind: + its id came from the departing point's payload, so a missing one means the + chain was already broken, and inventing a patch for a point that is not + there would not fix it. The caller logs that case. + """ + patches: list[PayloadPatch] = [] + + if point.is_active: + patches.append( + PayloadPatch( + point_id=point.point_id, + payload={ + "is_active": False, + "deleted_at": now.isoformat(), + **_provenance(point, actor=actor, now=now), + }, + expected_version=point.version, + ) + ) + + previous = neighbours.get(point.previous_chunk_id) if point.previous_chunk_id else None + if previous is not None and previous.next_chunk_id != point.next_chunk_id: + patches.append( + PayloadPatch( + point_id=previous.point_id, + payload={ + "next_chunk_id": _optional_id(point.next_chunk_id), + **_provenance(previous, actor=actor, now=now), + }, + expected_version=previous.version, + ) + ) + + following = neighbours.get(point.next_chunk_id) if point.next_chunk_id else None + if following is not None and following.previous_chunk_id != point.previous_chunk_id: + patches.append( + PayloadPatch( + point_id=following.point_id, + payload={ + "previous_chunk_id": _optional_id(point.previous_chunk_id), + **_provenance(following, actor=actor, now=now), + }, + expected_version=following.version, + ) + ) + + return tuple(patches) + + +def patch_for_deactivation(point: Point, *, actor: str, now: datetime) -> PayloadPatch: + """Deactivate one point without touching any pointer. + + Used by the whole-file sweep, where every point in the file leaves at once: + no active point survives to dangle, so there is no neighbour to relink and + the chain stays intact for a later reader of the deactivated file. + """ + return PayloadPatch( + point_id=point.point_id, + payload={ + "is_active": False, + "deleted_at": now.isoformat(), + **_provenance(point, actor=actor, now=now), + }, + expected_version=point.version, + ) diff --git a/src/infrastructure/postgres/repositories/source_files.py b/src/infrastructure/postgres/repositories/source_files.py index d182acc..ca18964 100644 --- a/src/infrastructure/postgres/repositories/source_files.py +++ b/src/infrastructure/postgres/repositories/source_files.py @@ -7,6 +7,7 @@ signature error rather than a cross-tenant leak. """ import uuid +from datetime import datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -67,3 +68,19 @@ def create( ) session.add(source_file) return source_file + + +def mark_soft_deleted(source_file: SourceFile, *, deleted_at: datetime) -> None: + """Retire a file: `status='soft_deleted'` plus `deleted_at` (ADR-0009). + + Takes the already-loaded row rather than an id, because the caller fetched + it under its tenant filter and re-fetching here would be a second place + that could forget that filter. + + Retiring the row matters beyond bookkeeping: `find_active_by_content_hash` + matches only `active` files, so a re-upload of the same bytes after a delete + creates a fresh file and re-ingests it, instead of taking the duplicate path + and returning a file whose points have all been deactivated. + """ + source_file.status = "soft_deleted" + source_file.deleted_at = deleted_at