From 734ba404c8e453dd771fa6725ba428f45f054f7c Mon Sep 17 00:00:00 2001 From: Lucas Winther Date: Sat, 26 Sep 2026 11:23:00 +0200 Subject: [PATCH] feat: expose membership and job APIs --- docs/ARCHITECTURE.md | 9 +- docs/TODO.md | 10 +- docs/plans/001-search-engine-api.md | 8 +- src/recoll_modern/api/collections.py | 12 +- src/recoll_modern/api/jobs.py | 78 +++++++++++ src/recoll_modern/api/memberships.py | 115 +++++++++++++++ src/recoll_modern/api/schemas.py | 72 +++++++++- src/recoll_modern/app.py | 61 +++++++- src/recoll_modern/storage/jobs.py | 13 +- src/recoll_modern/storage/membership.py | 52 +++++++ tests/api/test_jobs.py | 34 +++++ tests/api/test_memberships.py | 179 ++++++++++++++++++++++++ 12 files changed, 624 insertions(+), 19 deletions(-) create mode 100644 src/recoll_modern/api/jobs.py create mode 100644 src/recoll_modern/api/memberships.py create mode 100644 tests/api/test_jobs.py create mode 100644 tests/api/test_memberships.py diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ce81845..24415a8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -6,7 +6,8 @@ How `recoll-modern` is built today. Update in the same commit as any structural The Python service scaffold, SQLite schema, collection and membership repositories, document discovery/identity modules, durable job worker, engine adapter, authentication service, and initial -HTTP shell are built. Membership, search and job resource routes remain to be added. Recoll behaviour +HTTP shell, membership routes, and job routes are built. Search and document retrieval remain to be +added. Recoll behaviour below was verified against a real container. The goal: a self-hosted service that wraps Recoll's indexing and full-text search behind (a) a modern @@ -31,7 +32,7 @@ web UI and (b) an MCP server, running in Docker, with several independent collec | `helper/` | C reader run by Recoll's `metadatacmds` for each indexed file | | `src/recoll_modern/engine/` | Shared Protocol, deterministic fake, and isolated Recoll Python/indexer adapter | | `src/recoll_modern/domain/auth.py`, `storage/auth.py` | Argon2 password checks, seeded administrator, hashed server-side sessions and CSRF verification | -| `src/recoll_modern/api/` | HTTP auth, health, file browsing, collection metadata, prefix/proxy handling and JSON errors | +| `src/recoll_modern/api/` | HTTP auth, health, file browsing, collection/membership/rule/job resources, prefix/proxy handling and JSON errors | | `tests/` | Automated service tests | Discovery produces content hashes and root-relative paths without writing to the document volume. @@ -104,6 +105,10 @@ Core routes use a server-side session cookie and CSRF token; `GET /auth/session` the database only stores hashes. Forwarded scheme, host and prefix are applied only for configured proxy networks. +Membership, rule, and reindex writes create versioned, durable jobs. The worker updates SQLite +provenance, exports the membership map, and forces affected paths through the engine. Document listing +derives effective membership from SQLite and includes direct and rule provenance. + The resource groups are auth, health, read-only file browsing, collections and membership sources, directory rules, scoped search, document text/original retrieval, and indexing jobs. Search callers select returned fields; only `document_id` is unconditional, and inline full text is bounded. Every diff --git a/docs/TODO.md b/docs/TODO.md index e7f61b2..9b16126 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -15,12 +15,12 @@ Where the workflow stands (`AGENTS.md` -> Workflow: brainstorm -> spec -> plan - selectable search fields, subpath hosting and the packaging seam for deferred desktop spec 005. - Plan `docs/plans/001-search-engine-api.md`: **approved 2026-09-18.** Its listed Python runtime and development dependencies are approved. Execution follows the plan task by task. -- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–10 add discovery, content-hash +- Plan Tasks 1–3 are complete and independently reviewed. Tasks 4–11 add discovery, content-hash identity, additive membership, durable jobs, the atomic membership map, the engine boundary, - session authentication and the HTTP shell. Their checkboxes are current in the plan. The next task - is membership, rule and job routes. + session authentication, the HTTP shell, and membership/rule/job routes. Their checkboxes are current + in the plan. The next task is scoped search and document retrieval. -The next action is Task 11, "Expose membership, rule, and job APIs." Do not repeat Tasks 1–10. +The next action is Task 12, "Expose scoped search and document retrieval." Do not repeat Tasks 1–11. Things a resuming agent will otherwise get wrong: @@ -49,7 +49,7 @@ Things a resuming agent will otherwise get wrong: then record the verified ones in `AGENTS.md`. - [ ] Dockerfile + compose: Recoll in the image, document volume mounted `:ro`, writable volumes for the index and SQLite. -- [ ] Write the `metadatacmds` membership helper as a fast-starting non-Python program, fed by a flat +- [x] Write the `metadatacmds` membership helper as a fast-starting non-Python program, fed by a flat map exported from SQLite. A `python3` helper made a 400-file index 9x slower; Recoll spawns it once per file per index run. - [ ] Re-measure indexing overhead against real documents (PDF, EPUB) rather than small text files — diff --git a/docs/plans/001-search-engine-api.md b/docs/plans/001-search-engine-api.md index 1e2052f..083266b 100644 --- a/docs/plans/001-search-engine-api.md +++ b/docs/plans/001-search-engine-api.md @@ -787,7 +787,7 @@ git commit -m "feat: add authenticated HTTP service" - Consumes: Task 5 membership plans and Task 6 job repository - Produces: spec endpoints under `/collections/{id}/memberships`, `/directory-rules`, and `/jobs` -- [ ] **Step 1: Write failing request/response tests** +- [x] **Step 1: Write failing request/response tests** ```python def test_persistent_directory_request_enqueues_replayable_job(client: TestClient) -> None: @@ -808,11 +808,11 @@ def test_removing_rule_preserves_overlapping_membership(client: TestClient) -> N Cover invalid source combinations, exclusions outside source, default best-effort, strict atomicity, effective provenance, per-file overrides, patching rule exclusions, non-retryable jobs, and 50/200 pagination. -- [ ] **Step 2: Run tests and observe failure** +- [x] **Step 2: Run tests and observe failure** Run: `uv run pytest tests/api/test_memberships.py tests/api/test_jobs.py -v` -- [ ] **Step 3: Implement validated schemas and routes** +- [x] **Step 3: Implement validated schemas and routes** Use discriminated Pydantic models `FileSource(type="file", path)` and `DirectorySource(type="directory", path, mode, excluded_directories)`. Reject mode/exclusions on files @@ -823,7 +823,7 @@ effective membership plus `direct`, contributing rule IDs, and override state. I collection IDs and uses the whole document root when neither is present. Retry accepts failed or completed-with-errors jobs only. -- [ ] **Step 4: Verify and commit** +- [x] **Step 4: Verify and commit** Run: `uv run pytest tests/api/test_memberships.py tests/api/test_jobs.py -v && uv run ruff check . && uv run mypy src tests` diff --git a/src/recoll_modern/api/collections.py b/src/recoll_modern/api/collections.py index 5d7f23a..836a42b 100644 --- a/src/recoll_modern/api/collections.py +++ b/src/recoll_modern/api/collections.py @@ -7,7 +7,7 @@ from fastapi import APIRouter, Request, Response from fastapi.encoders import jsonable_encoder from recoll_modern.api.dependencies import require_user -from recoll_modern.api.schemas import CollectionCreateBody, CollectionPatchBody +from recoll_modern.api.schemas import CollectionCreateBody, CollectionPatchBody, validated_sources from recoll_modern.domain.errors import AppError from recoll_modern.domain.models import UNSET, CollectionCreate, CollectionPatch @@ -37,14 +37,18 @@ def create_collection( body: CollectionCreateBody, request: Request, response: Response ) -> dict[str, object]: require_user(request) + sources = ( + validated_sources(request.app.state.path_policy, body.sources) + if body.sources else [] + ) collection = request.app.state.collections.create( CollectionCreate(body.name, body.description) ) payload = jsonable_encoder(asdict(collection)) - if body.sources: + if sources: job = request.app.state.jobs.enqueue( "membership_add", - {"collection_id": collection.id, "sources": body.sources}, + {"version": 1, "collection_id": collection.id, "sources": sources}, body.fail_on_file_error, ) request.app.state.worker.notify() @@ -76,7 +80,7 @@ def delete_collection( raise AppError("confirmation_required", "Collection name confirmation is required", 409) paths = request.app.state.collections.delete(collection_id) job = request.app.state.jobs.enqueue( - "collection_cleanup", {"paths": paths}, False + "collection_cleanup", {"version": 1, "paths": paths}, False ) request.app.state.worker.notify() return {"job_id": job.id} diff --git a/src/recoll_modern/api/jobs.py b/src/recoll_modern/api/jobs.py new file mode 100644 index 0000000..2333257 --- /dev/null +++ b/src/recoll_modern/api/jobs.py @@ -0,0 +1,78 @@ +"""Durable indexing-job HTTP resources.""" + +from dataclasses import asdict +from typing import cast + +from fastapi import APIRouter, Request +from fastapi.encoders import jsonable_encoder + +from recoll_modern.api.dependencies import require_user +from recoll_modern.api.schemas import ReindexBody +from recoll_modern.domain.errors import AppError +from recoll_modern.domain.jobs import Job +from recoll_modern.indexing.paths import PathPolicyError + +router = APIRouter() + + +def _job_payload(job: Job) -> dict[str, object]: + return cast(dict[str, object], jsonable_encoder(asdict(job))) + + +@router.get("/jobs") +def list_jobs(request: Request, limit: int = 50, offset: int = 0) -> dict[str, object]: + require_user(request) + if not 1 <= limit <= 200 or offset < 0: + raise AppError("validation_failed", "Invalid pagination", 422) + jobs, total = request.app.state.jobs.list_jobs(limit, offset) + return {"items": [_job_payload(job) for job in jobs], "total": total} + + +@router.post("/jobs/reindex", status_code=202) +def reindex(body: ReindexBody, request: Request) -> dict[str, object]: + require_user(request) + policy = request.app.state.path_policy + paths: list[str] = [] + for path in body.paths or []: + try: + normalized = policy.normalize(path) + policy.resolve_file(normalized) + except PathPolicyError: + raise AppError("invalid_source", "Reindex path is invalid", 422) from None + paths.append(normalized) + for collection_id in body.collections or []: + request.app.state.collections.get(collection_id) + job = request.app.state.jobs.enqueue( + "reindex", + {"version": 1, "paths": sorted(set(paths)), + "collections": sorted(set(body.collections or []))}, + body.fail_on_file_error, + ) + request.app.state.worker.notify() + return {"job": _job_payload(job)} + + +@router.get("/jobs/{job_id}") +def get_job(job_id: str, request: Request) -> dict[str, object]: + require_user(request) + job = request.app.state.jobs.get(job_id) + if job is None: + raise AppError("job_not_found", "Job was not found", 404) + errors = request.app.state.jobs.file_errors(job_id) + return {"job": _job_payload(job), "file_errors": jsonable_encoder(errors)} + + +@router.post("/jobs/{job_id}/retry", status_code=202) +def retry_job(job_id: str, request: Request) -> dict[str, object]: + require_user(request) + job = request.app.state.jobs.get(job_id) + if job is None: + raise AppError("job_not_found", "Job was not found", 404) + try: + request.app.state.jobs.retry(job_id) + except ValueError: + raise AppError("job_not_retryable", "Job cannot be retried", 409) from None + request.app.state.worker.notify() + retried = request.app.state.jobs.get(job_id) + assert retried is not None + return {"job": _job_payload(retried)} diff --git a/src/recoll_modern/api/memberships.py b/src/recoll_modern/api/memberships.py new file mode 100644 index 0000000..5f89f50 --- /dev/null +++ b/src/recoll_modern/api/memberships.py @@ -0,0 +1,115 @@ +"""Collection membership, document provenance, and directory-rule routes.""" + +from dataclasses import asdict +from typing import cast + +from fastapi import APIRouter, Request +from fastapi.encoders import jsonable_encoder + +from recoll_modern.api.dependencies import require_user +from recoll_modern.api.schemas import ( + DirectorySource, + MembershipBody, + RulePatchBody, + validated_sources, +) +from recoll_modern.domain.errors import AppError +from recoll_modern.storage.membership import DirectoryRule + +router = APIRouter() + + +def _rule(request: Request, collection_id: str, rule_id: str) -> DirectoryRule: + rule = cast(DirectoryRule | None, request.app.state.memberships.get_rule(rule_id)) + if rule is None or rule.collection_id != collection_id: + raise AppError("directory_rule_not_found", "Directory rule was not found", 404) + return rule + + +@router.post("/collections/{collection_id}/memberships", status_code=202) +def add_memberships( + collection_id: str, body: MembershipBody, request: Request +) -> dict[str, object]: + require_user(request) + request.app.state.collections.get(collection_id) + if not body.sources: + raise AppError("invalid_source", "At least one source is required", 422) + sources = validated_sources(request.app.state.path_policy, body.sources) + job = request.app.state.jobs.enqueue( + "membership_add", + {"version": 1, "collection_id": collection_id, "sources": sources}, + body.fail_on_file_error, + ) + request.app.state.worker.notify() + return {"job": jsonable_encoder(asdict(job))} + + +@router.get("/collections/{collection_id}/documents") +def list_documents( + collection_id: str, request: Request, limit: int = 50, offset: int = 0 +) -> dict[str, object]: + require_user(request) + request.app.state.collections.get(collection_id) + if not 1 <= limit <= 200 or offset < 0: + raise AppError("validation_failed", "Invalid pagination", 422) + items, total = request.app.state.memberships.list_documents(collection_id, limit, offset) + return {"items": items, "total": total} + + +@router.delete("/collections/{collection_id}/documents/{document_hash}", status_code=202) +def remove_document( + collection_id: str, document_hash: str, request: Request +) -> dict[str, object]: + require_user(request) + request.app.state.collections.get(collection_id) + if request.app.state.documents.get(document_hash) is None: + raise AppError("document_not_found", "Document was not found", 404) + job = request.app.state.jobs.enqueue( + "membership_remove", + {"version": 1, "collection_id": collection_id, "document_hash": document_hash}, + False, + ) + request.app.state.worker.notify() + return {"job": jsonable_encoder(asdict(job))} + + +@router.get("/collections/{collection_id}/directory-rules") +def list_rules(collection_id: str, request: Request) -> dict[str, object]: + require_user(request) + request.app.state.collections.get(collection_id) + rules = [ + asdict(rule) for rule in request.app.state.memberships.list_rules() + if rule.collection_id == collection_id + ] + return {"items": rules, "total": len(rules)} + + +@router.patch("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202) +def patch_rule( + collection_id: str, rule_id: str, body: RulePatchBody, request: Request +) -> dict[str, object]: + require_user(request) + rule = _rule(request, collection_id, rule_id) + validated = validated_sources(request.app.state.path_policy, [DirectorySource( + type="directory", path=rule.relative_path, mode="persistent", + excluded_directories=body.excluded_directories, + )]) + job = request.app.state.jobs.enqueue( + "rule_patch", + {"version": 1, "rule_id": rule_id, + "excluded_directories": validated[0]["excluded_directories"]}, + False, + ) + request.app.state.worker.notify() + return {"job": jsonable_encoder(asdict(job))} + + +@router.delete("/collections/{collection_id}/directory-rules/{rule_id}", status_code=202) +def delete_rule(collection_id: str, rule_id: str, request: Request) -> dict[str, object]: + require_user(request) + _rule(request, collection_id, rule_id) + job = request.app.state.jobs.enqueue( + "rule_delete", {"version": 1, "rule_id": rule_id}, False + ) + request.app.state.worker.notify() + return {"job": jsonable_encoder(asdict(job))} diff --git a/src/recoll_modern/api/schemas.py b/src/recoll_modern/api/schemas.py index 2d70a2f..932323f 100644 --- a/src/recoll_modern/api/schemas.py +++ b/src/recoll_modern/api/schemas.py @@ -1,15 +1,83 @@ """Small HTTP request bodies for collection metadata.""" -from pydantic import BaseModel +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field + +from recoll_modern.domain.errors import AppError +from recoll_modern.indexing.paths import PathPolicy, PathPolicyError + + +class FileSource(BaseModel): + model_config = ConfigDict(extra="forbid") + type: Literal["file"] + path: str + + +class DirectorySource(BaseModel): + model_config = ConfigDict(extra="forbid") + type: Literal["directory"] + path: str + mode: Literal["snapshot", "persistent"] + excluded_directories: list[str] = Field(default_factory=list) + + +type SourceBody = Annotated[FileSource | DirectorySource, Field(discriminator="type")] + + +class MembershipBody(BaseModel): + sources: list[SourceBody] + fail_on_file_error: bool = False + + +class RulePatchBody(BaseModel): + excluded_directories: list[str] + + +class ReindexBody(BaseModel): + paths: list[str] | None = None + collections: list[str] | None = None + fail_on_file_error: bool = False class CollectionCreateBody(BaseModel): name: str description: str | None = None - sources: list[dict[str, object]] | None = None + sources: list[SourceBody] | None = None fail_on_file_error: bool = False class CollectionPatchBody(BaseModel): name: str | None = None description: str | None = None + + +def validated_sources( + policy: PathPolicy, sources: list[FileSource | DirectorySource] +) -> list[dict[str, object]]: + """Normalize sources and reject invalid or repeated root-relative selections.""" + result: list[dict[str, object]] = [] + seen: set[tuple[str, str]] = set() + for source in sources: + try: + path = policy.normalize(source.path) + if source.type == "file": + policy.resolve_file(path) + item: dict[str, object] = {"type": "file", "path": path} + else: + policy.resolve_directory(path) + excluded = [policy.normalize(value) for value in source.excluded_directories] + if any(not value.startswith(path + "/") for value in excluded): + raise PathPolicyError("invalid_source") + item = { + "type": "directory", "path": path, "mode": source.mode, + "excluded_directories": excluded, + } + except PathPolicyError: + raise AppError("invalid_source", "Membership source is invalid", 422) from None + key = (source.type, path) + if key in seen: + raise AppError("invalid_source", "Membership source is duplicated", 422) + seen.add(key) + result.append(item) + return result diff --git a/src/recoll_modern/app.py b/src/recoll_modern/app.py index 6c6070a..95de8aa 100644 --- a/src/recoll_modern/app.py +++ b/src/recoll_modern/app.py @@ -8,12 +8,15 @@ from datetime import timedelta from fastapi import FastAPI from recoll_modern.api import auth, collections, files, health +from recoll_modern.api import jobs as job_routes +from recoll_modern.api import memberships as membership_routes from recoll_modern.api.errors import install_error_handlers from recoll_modern.api.middleware import PrefixMiddleware from recoll_modern.config import Settings from recoll_modern.domain.auth import AuthService from recoll_modern.domain.collections import CollectionService from recoll_modern.domain.jobs import Job +from recoll_modern.domain.membership import MembershipService from recoll_modern.engine.protocol import Engine from recoll_modern.engine.recoll import RecollEngine from recoll_modern.indexing.handlers import MembershipJobHandler @@ -53,6 +56,12 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - jobs = JobRepository(database) map_rows = MembershipMapRows(memberships, policy.root) export_membership_map(map_rows, settings.membership_map_path) + membership_service = MembershipService(policy, documents, memberships) + + def reindex_changed(paths: tuple[str, ...]) -> None: + export_membership_map(map_rows, settings.membership_map_path) + if paths: + engine.force_reindex(tuple(str(policy.root / path) for path in sorted(set(paths)))) def cleanup(job: Job) -> None: export_membership_map(map_rows, settings.membership_map_path) @@ -71,6 +80,50 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - if changed: engine.force_reindex(tuple(str(policy.root / path) for path in sorted(changed))) + def remove_membership(job: Job) -> None: + collection_id = job.request.get("collection_id") + content_hash = job.request.get("document_hash") + if not isinstance(collection_id, str) or not isinstance(content_hash, str): + raise ValueError("invalid membership removal job") + changed = membership_service.plan_remove_document(collection_id, content_hash) + reindex_changed(changed) + + def patch_rule(job: Job) -> None: + rule_id = job.request.get("rule_id") + exclusions = job.request.get("excluded_directories") + if not isinstance(rule_id, str) or not isinstance(exclusions, list): + raise ValueError("invalid rule patch job") + if not all(isinstance(item, str) for item in exclusions): + raise ValueError("invalid rule patch job") + result = membership_service.plan_patch_rule(rule_id, tuple(exclusions)) + reindex_changed(result.changed_paths) + + def delete_rule(job: Job) -> None: + rule_id = job.request.get("rule_id") + if not isinstance(rule_id, str): + raise ValueError("invalid rule deletion job") + reindex_changed(membership_service.plan_delete_rule(rule_id)) + + def reindex(job: Job) -> None: + paths = job.request.get("paths", []) + collections_requested = job.request.get("collections", []) + if not isinstance(paths, list) or not isinstance(collections_requested, list): + raise ValueError("invalid reindex job") + for rule in memberships.list_rules(): + reconcile_rule(rule.id, policy, documents, memberships) + export_membership_map(map_rows, settings.membership_map_path) + selected = { + str(policy.root / path) for path in paths if isinstance(path, str) + } + requested_ids = {item for item in collections_requested if isinstance(item, str)} + for path, ids in memberships.membership_map_rows(policy.root): + if requested_ids.intersection(ids): + selected.add(path) + if not selected and not paths and not collections_requested: + selected.add(str(policy.root)) + if selected: + engine.force_reindex(tuple(sorted(selected))) + membership_handler = MembershipJobHandler( policy, documents, memberships, jobs, settings.membership_map_path, engine ) @@ -78,6 +131,10 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - "membership_add": membership_handler.handle, "collection_cleanup": cleanup, "reconcile_rules": reconcile_all, + "membership_remove": remove_membership, + "rule_patch": patch_rule, + "rule_delete": delete_rule, + "reindex": reindex, }) app.state.settings = settings app.state.database = database @@ -90,7 +147,7 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - app.state.jobs = jobs app.state.worker = worker if memberships.list_rules(): - jobs.enqueue("reconcile_rules", {}, False) + jobs.enqueue("reconcile_rules", {"version": 1}, False) worker.start() try: yield @@ -104,4 +161,6 @@ def create_app(settings: Settings | None = None, engine: Engine | None = None) - app.include_router(health.router) app.include_router(files.router) app.include_router(collections.router) + app.include_router(membership_routes.router) + app.include_router(job_routes.router) return app diff --git a/src/recoll_modern/storage/jobs.py b/src/recoll_modern/storage/jobs.py index 4dbb169..855190e 100644 --- a/src/recoll_modern/storage/jobs.py +++ b/src/recoll_modern/storage/jobs.py @@ -1,5 +1,7 @@ """Atomic SQLite transitions for indexing jobs.""" +from __future__ import annotations + import json import secrets import sqlite3 @@ -42,6 +44,15 @@ class JobRepository: row = connection.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() return _job_from_row(row) if row is not None else None + def list_jobs(self, limit: int, offset: int) -> tuple[list[Job], int]: + with self._database.connect() as connection: + total = int(connection.execute("SELECT COUNT(*) FROM jobs").fetchone()[0]) + rows = connection.execute( + "SELECT * FROM jobs ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?", + (limit, offset), + ).fetchall() + return [_job_from_row(row) for row in rows], total + def claim_next(self) -> Job | None: with self._database.transaction() as connection: row = connection.execute( @@ -130,7 +141,7 @@ class JobRepository: cursor = connection.execute( "UPDATE jobs SET state = 'queued', error_code = NULL, error_message = NULL, " "started_at = NULL, finished_at = NULL, progress = 0, total = 0 " - "WHERE id = ? AND state = 'failed'", + "WHERE id = ? AND state IN ('failed', 'completed_with_errors')", (job_id,), ) if cursor.rowcount != 1: diff --git a/src/recoll_modern/storage/membership.py b/src/recoll_modern/storage/membership.py index fa99044..ecc2e67 100644 --- a/src/recoll_modern/storage/membership.py +++ b/src/recoll_modern/storage/membership.py @@ -195,6 +195,58 @@ class MembershipRepository: if (collections := _effective(connection, str(row[1]))) ] + def list_documents( + self, collection_id: str, limit: int, offset: int + ) -> tuple[list[dict[str, object]], int]: + with self._database.connect() as connection: + hashes = [ + str(row[0]) + for row in connection.execute( + "SELECT content_hash FROM documents ORDER BY content_hash" + ) + if collection_id in _effective(connection, str(row[0])) + ] + items: list[dict[str, object]] = [] + for content_hash in hashes[offset:offset + limit]: + doc = connection.execute( + "SELECT byte_size, state FROM documents WHERE content_hash = ?", + (content_hash,), + ).fetchone() + canonical = connection.execute( + "SELECT relative_path FROM document_paths " + "WHERE document_hash = ? AND is_canonical = 1", + (content_hash,), + ).fetchone() + direct = connection.execute( + "SELECT 1 FROM direct_memberships " + "WHERE collection_id = ? AND document_hash = ?", + (collection_id, content_hash), + ).fetchone() is not None + rules = [ + str(row[0]) + for row in connection.execute( + "SELECT r.id FROM rule_memberships AS rm " + "JOIN directory_rules AS r ON r.id = rm.rule_id " + "WHERE r.collection_id = ? AND rm.document_hash = ? ORDER BY r.id", + (collection_id, content_hash), + ) + ] + override = connection.execute( + "SELECT 1 FROM membership_exclusions " + "WHERE collection_id = ? AND document_hash = ?", + (collection_id, content_hash), + ).fetchone() is not None + items.append({ + "document_id": content_hash, + "path": str(canonical[0]) if canonical is not None else None, + "size": int(doc[0]), + "state": str(doc[1]), + "direct": direct, + "contributing_rules": rules, + "override": override, + }) + return items, len(hashes) + def _effective(connection: sqlite3.Connection, content_hash: str) -> list[str]: return [ diff --git a/tests/api/test_jobs.py b/tests/api/test_jobs.py new file mode 100644 index 0000000..2a8bc0b --- /dev/null +++ b/tests/api/test_jobs.py @@ -0,0 +1,34 @@ +import time + +from fastapi.testclient import TestClient + + +def test_job_reindex_enqueues_versioned_request(authenticated: tuple[TestClient, str]) -> None: + client, csrf = authenticated + response = client.post("/jobs/reindex", json={}, headers={"X-CSRF-Token": csrf}) + assert response.status_code == 202 + job = response.json()["job"] + assert job["request"]["version"] == 1 + assert job["fail_on_file_error"] is False + listed = client.get("/jobs").json() + assert listed["total"] == 1 + + +def test_retry_refuses_completed_job(authenticated: tuple[TestClient, str]) -> None: + client, csrf = authenticated + response = client.post("/jobs/reindex", json={}, headers={"X-CSRF-Token": csrf}) + job_id = response.json()["job"]["id"] + for _ in range(100): + if client.get(f"/jobs/{job_id}").json()["job"]["state"] == "completed": + break + time.sleep(0.01) + response = client.post(f"/jobs/{job_id}/retry", headers={"X-CSRF-Token": csrf}) + assert response.status_code == 409 + assert response.json()["error"]["code"] == "job_not_retryable" + + +def test_job_list_validates_pagination(authenticated: tuple[TestClient, str]) -> None: + client, _ = authenticated + response = client.get("/jobs", params={"limit": 201}) + assert response.status_code == 422 + assert response.json()["error"]["code"] == "validation_failed" diff --git a/tests/api/test_memberships.py b/tests/api/test_memberships.py new file mode 100644 index 0000000..53398e0 --- /dev/null +++ b/tests/api/test_memberships.py @@ -0,0 +1,179 @@ +import time +from pathlib import Path +from typing import cast + +from fastapi.testclient import TestClient + + +def _collection(client: TestClient, csrf: str) -> str: + response = client.post( + "/collections", json={"name": "Books"}, headers={"X-CSRF-Token": csrf} + ) + return cast(str, response.json()["id"]) + + +def _wait(client: TestClient, job_id: str) -> dict[str, object]: + for _ in range(100): + job = client.get(f"/jobs/{job_id}").json()["job"] + if job["state"] not in ("queued", "running"): + return cast(dict[str, object], job) + time.sleep(0.01) + raise AssertionError("job did not finish") + + +def test_persistent_directory_request_enqueues_replayable_job( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + books = tmp_path / "docs" / "books" + books.mkdir() + (books / "a.txt").write_text("a") + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent", + "excluded_directories": []}]}, + ) + assert response.status_code == 202 + assert response.json()["job"]["state"] == "queued" + job_id = response.json()["job"]["id"] + assert _wait(client, job_id)["state"] == "completed" + rules = client.get(f"/collections/{collection_id}/directory-rules").json()["items"] + assert len(rules) == 1 + members = client.get(f"/collections/{collection_id}/documents").json()["items"] + assert len(members) == 1 + assert members[0]["contributing_rules"] == [rules[0]["id"]] + + +def test_invalid_file_source_combination_is_rejected( + authenticated: tuple[TestClient, str] +) -> None: + client, csrf = authenticated + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "file", "path": "a.txt", "mode": "persistent"}]}, + ) + assert response.status_code == 422 + + +def test_duplicate_normalized_sources_are_rejected( + authenticated: tuple[TestClient, str] +) -> None: + client, csrf = authenticated + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [ + {"type": "file", "path": "books/a.txt"}, + {"type": "file", "path": "books//a.txt"}, + ]}, + ) + assert response.status_code == 422 + + +def test_strict_job_keeps_membership_empty_on_file_error( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + (tmp_path / "docs" / "good.txt").write_text("good") + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [ + {"type": "file", "path": "good.txt"}, + {"type": "file", "path": "gone.txt"}, + ], "fail_on_file_error": True}, + ) + assert _wait(client, response.json()["job"]["id"])["state"] == "failed" + assert client.get(f"/collections/{collection_id}/documents").json()["items"] == [] + + +def test_removing_one_rule_preserves_overlapping_membership( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + root = tmp_path / "docs" + for name in ("books", "other"): + directory = root / name + directory.mkdir() + (directory / "a.txt").write_text("same") + collection_id = _collection(client, csrf) + for name in ("books", "other"): + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": name, "mode": "persistent"}]}, + ) + assert _wait(client, response.json()["job"]["id"])["state"] == "completed" + rules = client.get(f"/collections/{collection_id}/directory-rules").json()["items"] + response = client.delete( + f"/collections/{collection_id}/directory-rules/{rules[0]['id']}", + headers={"X-CSRF-Token": csrf}, + ) + assert _wait(client, response.json()["job"]["id"])["state"] == "completed" + members = client.get(f"/collections/{collection_id}/documents").json()["items"] + assert len(members) == 1 + assert members[0]["contributing_rules"] == [rules[1]["id"]] + + +def test_rule_exclusion_patch_removes_rule_membership( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + skipped = tmp_path / "docs" / "books" / "skip" + skipped.mkdir(parents=True) + (skipped / "a.txt").write_text("a") + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]}, + ) + assert _wait(client, response.json()["job"]["id"])["state"] == "completed" + rule_id = client.get(f"/collections/{collection_id}/directory-rules").json()["items"][0]["id"] + patch = client.patch( + f"/collections/{collection_id}/directory-rules/{rule_id}", + headers={"X-CSRF-Token": csrf}, + json={"excluded_directories": ["books/skip"]}, + ) + assert _wait(client, patch.json()["job"]["id"])["state"] == "completed" + assert client.get(f"/collections/{collection_id}/documents").json()["items"] == [] + + +def test_remove_document_creates_override_then_direct_add_restores_it( + authenticated: tuple[TestClient, str], tmp_path: Path +) -> None: + client, csrf = authenticated + books = tmp_path / "docs" / "books" + books.mkdir() + (books / "a.txt").write_text("a") + collection_id = _collection(client, csrf) + response = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "directory", "path": "books", "mode": "persistent"}]}, + ) + assert _wait(client, response.json()["job"]["id"])["state"] == "completed" + documents = client.get(f"/collections/{collection_id}/documents").json()["items"] + document_id = documents[0]["document_id"] + removed = client.delete( + f"/collections/{collection_id}/documents/{document_id}", + headers={"X-CSRF-Token": csrf}, + ) + assert _wait(client, removed.json()["job"]["id"])["state"] == "completed" + assert client.get(f"/collections/{collection_id}/documents").json()["items"] == [] + added = client.post( + f"/collections/{collection_id}/memberships", + headers={"X-CSRF-Token": csrf}, + json={"sources": [{"type": "file", "path": "books/a.txt"}]}, + ) + assert _wait(client, added.json()["job"]["id"])["state"] == "completed" + members = client.get(f"/collections/{collection_id}/documents").json()["items"] + assert len(members) == 1 + assert members[0]["direct"] is True + assert members[0]["override"] is False