1530 lines
74 KiB
Markdown
1530 lines
74 KiB
Markdown
# ARCHITECTURE — «Контракт-чек» production refactor
|
||
|
||
> **Audience:** a fresh agent (future me) picking this repo up cold. This doc
|
||
> is the single source of truth for the refactor. It supersedes the stage-1
|
||
> architecture in `IMPLEMENTATION_PLAN.md` and the relevant tickets in
|
||
> `TICKETS.md` (T-E1-008/011/012/015 and anything mentioning `arq`, Selectel S3,
|
||
> or the `MODE=...` dispatcher). The stage-0 prototype and its tickets
|
||
> (T-E0-*) remain valid. `BUSINESS_IDEA.md` (product/business) is unchanged.
|
||
>
|
||
> **One-line summary:** take the stage-0 prototype (PDF/DOCX → text → Ollama
|
||
> Cloud → markdown report) and rebuild it as a production, event-driven,
|
||
> multi-container application: a FastAPI **api**, three RabbitMQ-consuming
|
||
> **workers** (extract/OCR, deterministic prescreen, and LLM-analyze, split for
|
||
> CPU vs I/O profiles), a Telegram **bot** adapter, backed by
|
||
> **Postgres + RabbitMQ + MinIO + Redis**, observable via
|
||
> **structlog + Sentry + Prometheus/Grafana + OpenTelemetry**, behind
|
||
> **Nginx + certbot**. Monorepo, six fine-tuned Docker images
|
||
> (`srv/` — api + three workers + bot + prototype), shared `contract_check.core`
|
||
> domain package. Compose now, k8s-ready later.
|
||
> HA-*ready* (quorum queues, WAL archiving Postgres), not HA-*running* on one
|
||
> VPS — see §10.
|
||
|
||
---
|
||
|
||
## 1. Why this refactor (the pivot)
|
||
|
||
The prototype proved the LLM path end-to-end (T-E0-005). The original plan
|
||
(`IMPLEMENTATION_PLAN.md`) carried that into production with three assumptions:
|
||
|
||
1. **arq + Redis** as the job queue.
|
||
2. **Selectel S3** (cloud, RF-located) as object storage.
|
||
3. **One shared Docker image** with a `MODE=api|worker|bot` runtime dispatcher.
|
||
|
||
The owner has explicitly pivoted away from all three:
|
||
|
||
1. Queue → **RabbitMQ**, and the worker becomes **event-driven** (pipeline
|
||
fan-out: extract stage, then analyze stage), not a single in-process job.
|
||
2. Storage → **MinIO** (local/self-hosted S3), not Selectel.
|
||
3. Images → **one Dockerfile per service**, deps fine-tuned per image; the
|
||
`MODE` dispatcher is killed (one image = one entrypoint).
|
||
|
||
Additional decisions locked during the architecture questionnaire:
|
||
|
||
- **Three worker pools**, not one: `worker-extract` (pymupdf + Tesseract, CPU),
|
||
`worker-prescreen` (hybrid heuristic → optional LLM-fallback metadata extraction, CPU but
|
||
tiny), and `worker-analyze` (LLM, I/O). The prescreen stage routes low-risk
|
||
contracts to manual/auto-approval and high-risk contracts to deep LLM analysis,
|
||
protecting the expensive LLM worker from trivial documents.
|
||
- **LLM behind a provider interface** (`core/llm/port.py`), Ollama Cloud as
|
||
the first realization — swappable later for self-hosted Ollama / GigaChat.
|
||
- **Refund policy is a runtime switch** (`REFUND_POLICY=all|infra_only`) with
|
||
a failure-class taxonomy; credit refund stays idempotent.
|
||
- **Observability is full**, not logs-only: structlog+correlation_id, Sentry,
|
||
Prometheus+Grafana, OpenTelemetry.
|
||
- **Payments (ЮKassa) deferred** to a follow-up; the `invoices` table ships
|
||
empty in the initial migration for forward compatibility.
|
||
- **Web SPA deferred**; api/worker/bot land first.
|
||
- **Landing is incremental**, verified green at each step.
|
||
|
||
Non-goals for this refactor (do NOT build these now): multi-tenant orgs,
|
||
ЮKassa integration, React SPA, K8s manifests, RAG/vector DB, template
|
||
generation, E-sign/Gosuslugi.
|
||
|
||
---
|
||
|
||
## 2. System overview
|
||
|
||
Hexagonal (ports & adapters). The **core** owns all state and side effects
|
||
(DB, S3, MQ, LLM, credits). **Adapters** (bot, future web/cli) are thin
|
||
HTTP clients to the api and touch nothing but the api.
|
||
|
||
```
|
||
┌───────────────────── HTTPS (Nginx + certbot) ─────────────────────┐
|
||
│ │
|
||
Telegram ───► bot (aiogram) ──HTTP──► api (FastAPI) ──publish──► RabbitMQ (direct)
|
||
(adapter, HTTP-only) (core) │
|
||
│ owns: Postgres, MinIO, ▼
|
||
│ Redis, credits, tokens extract.q ──► worker-extract
|
||
│ (core) (quorum) (core, CPU: pymupdf+tesseract)
|
||
▼ │ publish PrescreenRequested
|
||
Postgres ▼
|
||
MinIO ◄──── read/write ────► prescreen.q ──► worker-prescreen
|
||
Redis (rate limit/sess) (quorum) (hybrid heuristic router)
|
||
│ publish AnalyzeRequested
|
||
▼
|
||
analyze.q ──► worker-analyze
|
||
(quorum) (core, I/O: LLM provider)
|
||
│ save Report, status=done
|
||
▼
|
||
Postgres
|
||
```
|
||
|
||
Data plane:
|
||
|
||
- **Postgres** — source of truth (users, documents, reports, jobs, tokens,
|
||
invoices). Replication-ready (§10).
|
||
- **MinIO** — raw doc blobs + extracted text blobs, TTL-purged.
|
||
- **RabbitMQ** — durable job pipeline, quorum queues, TTL retry, DLQ.
|
||
- **Redis** — rate limiting (future), sessions (future). Idle for now but in
|
||
compose; do not use it as a job queue.
|
||
|
||
Control plane:
|
||
|
||
- **api** is the only writer to Postgres/MinIO for user-initiated mutations.
|
||
Workers write their own stage rows and reports but never create users or
|
||
move credits except via the idempotent `refund_credit` helper.
|
||
|
||
---
|
||
|
||
## 3. Locked decisions (cheat sheet)
|
||
|
||
| Area | Decision | Replaces |
|
||
|---|---|---|
|
||
| Repo | Monorepo, shared `contract_check.core`, 6 Dockerfiles in `srv/<svc>/` | one-image `MODE` dispatch |
|
||
| Scope now | api + worker-extract + worker-prescreen + worker-analyze + bot (+ prototype benchmark) | (web/payments later) |
|
||
| Queue | RabbitMQ, direct exchange, pipeline fan-out | arq + Redis |
|
||
| Retry | TTL retry queues, exponential backoff, final DLQ | — |
|
||
| Workers | Three pools (extract/OCR + prescreen + analyze/LLM) | single arq worker |
|
||
| Storage | MinIO, `users/{uid}/docs/{did}.{ext}` | Selectel S3 |
|
||
| Upload | Proxy through API (multipart) | — |
|
||
| Doc retention | TTL purge of raw docs after N days | — |
|
||
| Redis | Kept (rate limit/sessions future) | (no longer the queue) |
|
||
| Auth | Per-adapter `service_tokens`, revocable | single `SERVICE_TOKEN` |
|
||
| Report delivery | Polling now (fine-grained stage), SSE/webhook later | — |
|
||
| Sync `/analyze` | No | — |
|
||
| Doc status | Fine-grained `queued→extracting→prescreening→ocr→analyzing→done\|failed` | coarse status |
|
||
| Extra tables | jobs, service_tokens, invoices(stub) | 3-table plan |
|
||
| Report storage | JSONB + markdown column | — |
|
||
| Durability | HA-*ready* (quorum queues, WAL archive) | — |
|
||
| Observability | structlog+corr, Sentry, Prom/Grafana, OTel | docker logs |
|
||
| LLM | Provider port + Ollama Cloud adapter | direct client |
|
||
| Refund | Policy switch `all\|infra_only` + failure classes | refund-all |
|
||
| Edge | Nginx + certbot | — |
|
||
| Deploy | Compose now, k8s-ready later | — |
|
||
| Prototype | Kept as standalone benchmark | — |
|
||
| Tests | pytest+respx unit + testcontainers integration | — |
|
||
| Python | **3.13** (was 3.14) — wheel availability | py3.14 |
|
||
| Landing | Incremental, green per step | — |
|
||
|
||
---
|
||
|
||
## 4. Repository layout (target)
|
||
|
||
Existing modules migrate as annotated. New code is marked `(new)`.
|
||
|
||
```
|
||
DealDocumentScreening/
|
||
├── ARCHITECTURE.md (this file)
|
||
├── BUSINESS_IDEA.md (unchanged — product/business)
|
||
├── IMPLEMENTATION_PLAN.md (stage-0 valid; stage-1+ superseded here)
|
||
├── TICKETS.md (T-E0-* valid; T-E1-* superseded here)
|
||
├── README.md (rewrite: how to run the stack)
|
||
├── pyproject.toml (rewrite: core + per-service extras + dev group)
|
||
├── uv.lock (regenerated)
|
||
├── .env.example (rewrite: full env list, §11)
|
||
├── docker-compose.yml (rewrite: infra + services, profiles, §12)
|
||
├── srv/ (new — one Dockerfile per service, deps-tuned)
|
||
│ ├── api/Dockerfile (new)
|
||
│ ├── worker-extract/Dockerfile (new, tesseract layer)
|
||
│ ├── worker-analyze/Dockerfile (new, lean)
|
||
│ ├── bot/Dockerfile (new)
|
||
│ └── prototype/Dockerfile (new, preserves stage-0 CLI)
|
||
├── deploy/ (PLANNED — not yet built; lands with §20 steps 6 / E1-009..010)
|
||
│ ├── nginx/
|
||
│ │ ├── templates/contract-check.conf.template (reverse proxy + TLS)
|
||
│ │ └── certbot-init.sh (initial cert + nginx reload)
|
||
│ └── observability/
|
||
│ ├── prometheus.yml (planned — scrape api + workers :9100/:9101/:9102)
|
||
│ ├── otel-collector-config.yaml (planned)
|
||
│ ├── tempo.yaml (planned — trace storage)
|
||
│ └── grafana/provisioning/
|
||
│ ├── datasources/ (prometheus + tempo)
|
||
│ └── dashboards/ (starter: queue depth, job latency, LLM tokens)
|
||
├── migrations/ (new — alembic)
|
||
│ ├── env.py
|
||
│ ├── script.py.mako
|
||
│ └── versions/
|
||
│ ├── 0001_initial.py (6 tables, §7)
|
||
│ └── 0002_api_keys.py (api_keys, api_key_requests — B2B, §15)
|
||
├── src/contract_check/
|
||
│ ├── __init__.py
|
||
│ ├── core/ (new — shared domain, imported by every service image)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── config.py (pydantic-settings: base + per-service, §11)
|
||
│ │ ├── logging.py (structlog JSON + correlation_id contextvar)
|
||
│ │ ├── telemetry.py (OTel SDK init, FastAPI/asyncio instrumentation)
|
||
│ │ ├── sentry.py (sentry_sdk init helper)
|
||
│ │ ├── metrics.py (prometheus_client registry + counters/hists)
|
||
│ │ ├── api_keys.py (B2B key gen/hash/verify — sha256 + hmac.compare_digest)
|
||
│ │ ├── rate_limit.py (token-bucket per api_key_id; Memory + Redis backends)
|
||
│ │ ├── redis_client.py (async Redis client from redis_url)
|
||
│ │ ├── db/
|
||
│ │ │ ├── __init__.py
|
||
│ │ │ ├── models.py (SQLAlchemy 2 decl: User, Document, Report, Job, ServiceToken, Invoice, ApiKey, ApiKeyRequest)
|
||
│ │ │ ├── session.py (async_sessionmaker, engine)
|
||
│ │ │ └── enums.py (DocStatus, JobStatus, FailureClass — as plain str constants)
|
||
│ │ ├── mq/
|
||
│ │ │ ├── __init__.py
|
||
│ │ │ ├── topology.py (exchange/queue/rk constants + declare_all())
|
||
│ │ │ ├── publisher.py (aio-pika RobustChannel, publisher confirms)
|
||
│ │ │ ├── consumer.py (base Consumer class: connect/prefetch/handle/nack-retry)
|
||
│ │ │ └── messages.py (pydantic: DocumentUploaded, DocumentExtracted)
|
||
│ │ ├── s3/
|
||
│ │ │ ├── port.py (Storage Protocol: put/get/stat/delete/presign)
|
||
│ │ │ └── minio_storage.py (MinIO adapter, bucket init, key builders)
|
||
│ │ ├── llm/
|
||
│ │ │ ├── port.py (LLMProvider Protocol + AnalysisResult dataclass)
|
||
│ │ │ ├── ollama_cloud.py (ports prototype llm_client: repair-loop + 429 fallback)
|
||
│ │ │ └── factory.py (provider selection by LLM_PROVIDER env)
|
||
│ │ ├── extraction/ (new — DocumentExtractor port + adapters)
|
||
│ │ │ ├── port.py (ExtractedDocument, DocumentExtractor Protocol, errors)
|
||
│ │ │ ├── factory.py (mime/suffix → adapter; SUPPORTED_SUFFIXES gate)
|
||
│ │ │ ├── adapters/ (pdf_pymupdf, docx_mammoth, rtf_striprtf, txt_chardet, ocr_tesseract)
|
||
│ │ │ └── __init__.py (extract_document orchestrator with OCR fallback)
|
||
│ │ ├── analysis/ (existing; now consumed by prototype + ocr adapter)
|
||
│ │ │ ├── __init__.py
|
||
│ │ │ ├── extractor.py (← from contract_check/extractor.py; prototype use)
|
||
│ │ │ ├── chunker.py (← from contract_check/chunker.py; + chunk_markdown)
|
||
│ │ │ ├── checklist.py (← from contract_check/checklist.py)
|
||
│ │ │ ├── report_schema.py (← from contract_check/report_schema.py)
|
||
│ │ │ ├── ocr.py (pytesseract via pymupdf rasterize)
|
||
│ │ │ └── analyzer.py (orchestrate chunk→LLM→merge→dedupe→sort→markdown)
|
||
│ │ ├── credits.py (reserve_credit atomic UPDATE; refund_credit idempotent + policy)
|
||
│ │ └── tokens.py (service-token hash/verify; FastAPI dependency)
|
||
│ ├── api/ (new — FastAPI image)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── __main__.py (uvicorn entrypoint)
|
||
│ │ ├── app.py (create_app: middleware, routes, lifespan)
|
||
│ │ ├── deps.py (db session, s3, publisher, service-token dep, api-key dep, rate-limit)
|
||
│ │ ├── services.py (shared upload→reserve→MinIO→publish logic; used by documents + b2b)
|
||
│ │ ├── routes/
|
||
│ │ │ ├── health.py (/healthz, /readyz)
|
||
│ │ │ ├── documents.py (POST /api/v1/documents — upload→MinIO→publish, reserve)
|
||
│ │ │ ├── reports.py (GET /api/v1/reports/{id} — 202+stage or 200+md)
|
||
│ │ │ ├── me.py (GET /api/v1/me — credits balance)
|
||
│ │ │ ├── metrics.py (/metrics — prometheus)
|
||
│ │ │ └── b2b.py (X-API-Key: POST /analyze, GET /b2b/reports, /b2b/usage, /b2b/keys CRUD)
|
||
│ │ └── middleware.py (correlation_id inject, request metrics, Sentry)
|
||
│ ├── worker_extract/ (new — CPU image: pymupdf + tesseract)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── __main__.py (entrypoint: start consumer + metrics server)
|
||
│ │ ├── handler.py (handle DocumentUploaded: MinIO dl → extraction factory → OCR fallback → upload .txt (markdown) → publish PrescreenRequested)
|
||
│ │ └── consumer.py (wire handler into core.mq.consumer base)
|
||
│ ├── worker_prescreen/ (new — lean CPU image: hybrid heuristic + optional LLM fallback)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── __main__.py (entrypoint: metrics server on :9104)
|
||
│ │ ├── config.py (PrescreenSettings, prefetch tunable)
|
||
│ │ ├── handler.py (handle PrescreenRequested: download text → hybrid extract → route → persist prescreen_results → publish AnalyzeRequested/auto-approve/manual-review)
|
||
│ │ ├── consumer.py (wire handler into core.mq.consumer base)
|
||
│ │ ├── extractor.py (PrescreenContractMeta + MetadataExtractor protocol + compat shim)
|
||
│ │ ├── extractor_heuristic.py (Stage 1: deterministic keyword/positional extractor, regex-free)
|
||
│ │ ├── extractor_llm.py (Stage 2: dict → validated PrescreenContractMeta)
|
||
│ │ ├── extractor_hybrid.py (orchestrator: heuristic → optional LLM fallback)
|
||
│ │ └── router.py (decision logic: auto_approve / manual_review / deep_analysis)
|
||
│ ├── worker_analyze/ (new — I/O image: LLM provider)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── __main__.py
|
||
│ │ ├── handler.py (handle AnalyzeRequested: text dl→chunk→LLM (with prescreen context)→validate→save report→done)
|
||
│ │ └── consumer.py
|
||
│ ├── bot/ (new — aiogram adapter, HTTP-only)
|
||
│ │ ├── __init__.py
|
||
│ │ ├── __main__.py
|
||
│ │ ├── config.py (BotSettings: BOT_TOKEN, API_URL, BOT_SERVICE_TOKEN, poll tuning)
|
||
│ │ ├── client.py (ApiClient: typed httpx wrapper; ApiError/NoCreditsError/...)
|
||
│ │ └── handlers.py (/start, doc upload→POST api, poll→send report)
|
||
│ └── prototype/ (moved from contract_check/prototype.py — kept as benchmark)
|
||
│ ├── __init__.py
|
||
│ └── __main__.py (stage-0 CLI, imports updated to core.analysis.*)
|
||
└── tests/
|
||
├── conftest.py (fixtures: testcontainers pg/rabbit/minio)
|
||
├── unit/
|
||
│ ├── test_chunker.py
|
||
│ ├── test_extraction_adapters.py
|
||
│ ├── test_extraction_factory.py
|
||
│ ├── test_extractor.py
|
||
│ ├── test_prescreen_extractor.py
|
||
│ ├── test_prescreen_router.py
|
||
│ ├── test_checklist_report.py
|
||
│ ├── test_llm_ollama_cloud.py (respx: 200 / 429-fallback / repair)
|
||
│ ├── test_credits.py
|
||
│ ├── test_messages.py
|
||
│ ├── test_rate_limit.py
|
||
│ ├── test_bot_client.py
|
||
│ └── test_bot_boundary.py (static AST: bot must not import core.db/s3/llm/mq/credits)
|
||
└── integration/
|
||
├── conftest.py (seed service-token + api-key; testcontainers wiring)
|
||
├── test_upload_pipeline.py (POST /documents → message on extract.q or prescreen.q)
|
||
├── test_extract_worker.py (DocumentUploaded → PrescreenRequested published)
|
||
├── test_analyze_worker.py (AnalyzeRequested → Report saved, status done)
|
||
├── test_credits_db.py (reserve/refund against real Postgres)
|
||
└── test_b2b_api.py (X-API-Key: 202 → poll report; 401/429 branches)
|
||
```
|
||
|
||
**Import rule (hexagonal boundary, enforced in review):**
|
||
|
||
- `core/*` may import anything.
|
||
- `api/*`, `worker_extract/*`, `worker_prescreen/*`, `worker_analyze/*` import only `core/*`.
|
||
- `bot/*` imports only `httpx` + its own modules. **It must NOT import
|
||
`core.db`, `core.s3`, `core.llm`, `core.mq`, `core.credits`.** It speaks to
|
||
the api over HTTP only. (Add a ruff/flake8 import-forbidden rule or a unit
|
||
test that asserts this.)
|
||
- `prototype/*` imports `core.analysis.*` but is otherwise standalone (no DB/MQ).
|
||
|
||
---
|
||
|
||
## 5. RabbitMQ topology (full spec)
|
||
|
||
No plugins required. Pure direct exchanges + TTL/DLX for retry. All
|
||
declarations in `core/mq/topology.py` and idempotent (declare on every
|
||
service start).
|
||
|
||
### Exchanges (direct)
|
||
|
||
| Name | Type | Purpose |
|
||
|---|---|---|
|
||
| `contracts.x` | direct | Main exchange. Routing keys: `extract`, `prescreen`, `analyze`. |
|
||
| `contracts.retry.x` | direct | DLX target of main queues; routing keys: `retry.extract`, `retry.prescreen`, `retry.analyze`. |
|
||
|
||
### Queues
|
||
|
||
| Name | Type | Args | Bound (exchange / rk) | Consumers |
|
||
|---|---|---|---|---|
|
||
| `extract.q` | **quorum** | `x-dead-letter-exchange=contracts.retry.x`, `x-dead-letter-routing-key=retry.extract` | contracts.x / `extract` | worker-extract |
|
||
| `prescreen.q` | **quorum** | `x-dead-letter-exchange=contracts.retry.x`, `x-dead-letter-routing-key=retry.prescreen` | contracts.x / `prescreen` | worker-prescreen |
|
||
| `analyze.q` | **quorum** | `x-dead-letter-exchange=contracts.retry.x`, `x-dead-letter-routing-key=retry.analyze` | contracts.x / `analyze` | worker-analyze |
|
||
| `extract.retry.q` | classic | `x-dead-letter-exchange=contracts.x`, `x-dead-letter-routing-key=extract`; per-message `expiration` | contracts.retry.x / `retry.extract` | none (delay slot) |
|
||
| `prescreen.retry.q` | classic | `x-dead-letter-exchange=contracts.x`, `x-dead-letter-routing-key=prescreen`; per-message `expiration` | contracts.retry.x / `retry.prescreen` | none (delay slot) |
|
||
| `analyze.retry.q` | classic | `x-dead-letter-exchange=contracts.x`, `x-dead-letter-routing-key=analyze`; per-message `expiration` | contracts.retry.x / `retry.analyze` | none (delay slot) |
|
||
| `extract.dlq` | **quorum** | — | — | none (manual requeue) |
|
||
| `prescreen.dlq` | **quorum** | — | — | none (manual requeue) |
|
||
| `analyze.dlq` | **quorum** | — | — | none (manual requeue) |
|
||
|
||
### Flow & retry mechanics
|
||
|
||
1. **api** publishes `DocumentUploaded` to `contracts.x` rk `extract`
|
||
(persistent, `correlation_id` UUID in both message id and
|
||
`headers["x-correlation-id"]`).
|
||
2. **worker-extract** consumes from `extract.q` with `prefetch=1..N` (config,
|
||
default 1 — extraction is CPU-bound). On success: upload extracted **Markdown**
|
||
to MinIO (`text/markdown`), publish `PrescreenRequested` to `contracts.x` rk
|
||
`prescreen` (`is_structured`/`has_tables` metadata included), ack. On failure:
|
||
see retry below.
|
||
3. **worker-prescreen** consumes from `prescreen.q` with `prefetch=1`. It runs a
|
||
**hybrid two-stage extractor**: Stage 1 is a deterministic regex-free heuristic
|
||
over keyword dictionaries and positional windows; Stage 2 is an LLM fallback
|
||
(`extract_prescreen` on `core/llm/port.py`) triggered only when heuristic
|
||
confidence is below `PRESCREEN_LLM_FALLBACK_THRESHOLD` and the kill-switch
|
||
`PRESCREEN_LLM_FALLBACK_ENABLED` is true. It computes a weighted
|
||
`confidence_score`, routes to `auto_approve` (lightweight report, disabled by
|
||
default), `manual_review` (terminal DB state for future admin SPA), or
|
||
`deep_analysis`. For `deep_analysis` it publishes `AnalyzeRequested` to
|
||
`contracts.x` rk `analyze` carrying `prescreen_meta`. On failure: retry.
|
||
The v1 regex implementation is kept behind `PRESCREEN_KEEP_REGEX=true` for one
|
||
release as a rollback safety net.
|
||
4. **worker-analyze** consumes from `analyze.q` with `prefetch=3` (mirrors
|
||
Ollama Pro concurrency). On success: save report, `status=done`, ack.
|
||
On failure: retry.
|
||
5. **Retry (nack):** the base consumer does **not** use `basic.nack(requeue=True)`
|
||
(instant re-redelivery, no backoff). Instead it publishes a copy of the
|
||
message to the matching `*.retry.q` with:
|
||
- `headers["x-attempt"] = attempt + 1`
|
||
- `expiration = str(int(BASE_DELAY_MS * 2 ** attempt))` (e.g.
|
||
`BASE_DELAY_MS=2000` → 2s, 4s, 8s, 16s, 32s)
|
||
- then `basic_ack` the original.
|
||
When the retry queue's per-message TTL expires, its DLX bounces the
|
||
message back to `contracts.x` with rk `extract`/`prescreen`/`analyze` → re-enters the
|
||
main quorum queue. Clean, plugin-free, exponential.
|
||
6. **Poison (max attempts):** when `headers["x-attempt"] >= MAX_ATTEMPTS`
|
||
(default 5), the consumer publishes to `extract.dlq` / `prescreen.dlq` /
|
||
`analyze.dlq` instead of retry, acks the original, sets `jobs.dlq=true`,
|
||
`jobs.last_failure_class`, transitions `documents.status=failed`, and
|
||
calls `refund_credit(document_id, failure_class)` per policy (§8).
|
||
7. **Manual requeue:** a CLI script `scripts/mq_requeue.py` (or mgmt UI)
|
||
moves a DLQ message back to the main exchange with reset attempt.
|
||
|
||
### Prefetch & concurrency
|
||
|
||
- worker-extract: RabbitMQ prefetch caps concurrent extraction jobs (CPU
|
||
bound; default 1, tune to CPU count).
|
||
- worker-prescreen: prefetch=1; Stage 1 heuristic is cheap but routing decisions must be
|
||
ordered per document.
|
||
- worker-analyze: prefetch caps concurrent LLM jobs (default 3, mirrors
|
||
Ollama Pro). **Additionally** keep the in-process `asyncio.Semaphore`
|
||
from the prototype `llm_client` — prefetch caps *jobs*, the semaphore
|
||
caps *parallel chunk requests within a job*. Both are needed for long
|
||
multi-chunk contracts.
|
||
|
||
### Idempotency
|
||
|
||
Every handler starts by re-reading `documents.status` (and `jobs` row) by
|
||
`correlation_id`/`document_id`. If already terminal (`done`/`failed`) or the
|
||
job is mid-flight by another consumer, ack and exit — do not re-run the LLM,
|
||
do not double-refund. This preserves the existing idempotency invariant from
|
||
T-E1-008 across the queue pivot.
|
||
|
||
### Message schemas (`core/mq/messages.py`, pydantic v2)
|
||
|
||
```python
|
||
class DocumentUploaded(BaseModel):
|
||
correlation_id: UUID
|
||
document_id: UUID
|
||
user_id: UUID
|
||
s3_key: str # users/{uid}/docs/{did}.{ext}
|
||
filename: str
|
||
mime: str
|
||
attempt: int = 0
|
||
|
||
|
||
class DocumentExtracted(BaseModel):
|
||
correlation_id: UUID
|
||
document_id: UUID
|
||
user_id: UUID
|
||
extracted_s3_key: str # users/{uid}/docs/{did}.txt
|
||
char_count: int
|
||
ocr_used: bool
|
||
is_structured: bool = False
|
||
has_tables: bool = False
|
||
attempt: int = 0
|
||
|
||
|
||
class PrescreenRequested(BaseModel):
|
||
correlation_id: UUID
|
||
document_id: UUID
|
||
user_id: UUID
|
||
text_s3_key: str # users/{uid}/docs/{did}.txt
|
||
filename: str
|
||
char_count: int = 0
|
||
ocr_used: bool = False
|
||
is_structured: bool = False
|
||
has_tables: bool = False
|
||
attempt: int = 0
|
||
|
||
|
||
class PrescreenCompleted(BaseModel):
|
||
correlation_id: UUID
|
||
document_id: UUID
|
||
user_id: UUID
|
||
text_s3_key: str
|
||
filename: str
|
||
prescreened_at: str
|
||
contract_type: str | None = None
|
||
party_a: str | None = None
|
||
party_b: str | None = None
|
||
total_amount: float | None = None
|
||
currency: str | None = None
|
||
start_date: str | None = None
|
||
end_date: str | None = None
|
||
has_penalty_clause: bool | None = None
|
||
has_termination_clause: bool | None = None
|
||
has_arbitration: bool | None = None
|
||
confidence_score: float | None = None
|
||
routing_decision: str # auto_approve | manual_review | deep_analysis
|
||
auto_summary: str | None = None
|
||
auto_findings: list[dict] = []
|
||
extractor_version: str = "heuristic-v2" # heuristic-v2 | hybrid-llm-v1
|
||
attempt: int = 0
|
||
|
||
|
||
class AnalyzeRequested(BaseModel):
|
||
correlation_id: UUID
|
||
document_id: UUID
|
||
user_id: UUID
|
||
extracted_s3_key: str
|
||
filename: str
|
||
char_count: int = 0
|
||
ocr_used: bool = False
|
||
is_structured: bool = False
|
||
has_tables: bool = False
|
||
prescreen_meta: PrescreenCompleted | None = None
|
||
attempt: int = 0
|
||
```
|
||
|
||
RabbitMQ `headers`: `x-correlation-id`, `x-attempt`, `x-origin` (api |
|
||
worker-extract | worker-prescreen). `content_type=application/json`,
|
||
`delivery_mode=2` (persistent). Validate on consume with the pydantic model;
|
||
on validation error → `.dlq` immediately with `failure_class=infra`.
|
||
|
||
---
|
||
|
||
## 6. MinIO (object storage)
|
||
|
||
### Bucket & keys
|
||
|
||
- One bucket per env: `contract-check-docs` (configurable, `S3_BUCKET`).
|
||
- Layout: `users/{user_id}/docs/{document_id}.{ext}` (original upload) and
|
||
`users/{user_id}/docs/{document_id}.txt` (extracted text, written by
|
||
worker-extract, read by worker-analyze).
|
||
- Key builders in `core/s3/minio_storage.py`: `original_key(uid, did, ext)`,
|
||
`extracted_key(uid, did)`. Never construct keys ad-hoc.
|
||
|
||
### Upload path (proxy through API)
|
||
|
||
`POST /api/v1/documents` (multipart): the api validates mime/size, reserves
|
||
the credit (§8), writes the blob to MinIO via the Storage port, creates the
|
||
`documents` row (`status=queued`), inserts a `jobs` row, publishes
|
||
`DocumentUploaded`, returns `202 + {document_id, correlation_id}`. Adapters
|
||
never see S3 credentials. Presigned URLs are a future optimization only.
|
||
|
||
### Initialization
|
||
|
||
A one-shot `minio-init` service in compose runs `mc alias set ... && mc mb
|
||
... && mc anonymous set none ...` on boot. Idempotent. Creates the bucket
|
||
before api/worker start. Alternatively `core/s3/minio_storage.py` does
|
||
`ensure_bucket()` lazily on first use — keep both; lazy is the safety net.
|
||
|
||
### Retention (152-ФЗ friendliness)
|
||
|
||
- MinIO **ILM lifecycle rule**: expire objects under `users/*/docs/*` after
|
||
`DOC_RETENTION_DAYS` (default 7, configurable). Configured via `mc ilm
|
||
add/queue` in `minio-init`, or via the MinIO console. The extracted `.txt`
|
||
may share the rule or a longer one (`TEXT_RETENTION_DAYS`, default 30).
|
||
- The **report** (findings + markdown) lives in **Postgres**, not MinIO, so
|
||
it survives the raw-doc purge. This is the 152-ФЗ lever: raw contract
|
||
text leaves the system on a schedule, only the structured findings stay.
|
||
|
||
### Encryption / versioning
|
||
|
||
- SSE-S3 (server-side encryption with MinIO-managed keys) **on** for
|
||
production env, off for dev — toggle via `S3_SERVER_SIDE_ENCRYPTION`.
|
||
- Versioning **off** (documents are write-once; versioning adds cost and
|
||
complicates TTL expiry).
|
||
|
||
---
|
||
|
||
## 7. Postgres schema (Alembic migrations)
|
||
|
||
Six core tables plus additive migrations. `status`/`queue`/`adapter`/`role`
|
||
columns are `TEXT + CHECK` (not Postgres enums) so migrations are additive —
|
||
matches the convention noted in `IMPLEMENTATION_PLAN.md` §1.2.
|
||
|
||
Migrations (hand-written, async `env.py`):
|
||
- `0001` initial schema: users/documents/reports/jobs/service_tokens/invoices
|
||
- `0002` `api_keys` + `api_key_requests` (B2B)
|
||
- `0003` webUI auth columns on `users` (`email`, `password_hash`, reset tokens, `is_active`)
|
||
- `0004` `users.role` for the admin panel (`'user'|'admin'`, default `'user'`)
|
||
- `0005` Telegram bot hardening columns on `users`
|
||
- `0006` `prescreen_results` table; expand `documents.status` and `jobs.queue` enums
|
||
|
||
```sql
|
||
-- users
|
||
CREATE TABLE users (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
telegram_id BIGINT UNIQUE,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
credits_left INTEGER NOT NULL DEFAULT 0,
|
||
email TEXT UNIQUE, -- added by 0003
|
||
password_hash TEXT, -- added by 0003
|
||
password_reset_token_hash TEXT, -- added by 0003
|
||
password_reset_expires_at TIMESTAMPTZ, -- added by 0003
|
||
is_active BOOLEAN NOT NULL DEFAULT TRUE, -- added by 0003
|
||
role TEXT NOT NULL DEFAULT 'user' -- added by 0004
|
||
CHECK (role IN ('user','admin')),
|
||
CONSTRAINT users_credits_nonneg CHECK (credits_left >= 0),
|
||
CONSTRAINT users_identity_present CHECK (telegram_id IS NOT NULL OR email IS NOT NULL)
|
||
);
|
||
CREATE INDEX users_role_idx ON users (role); -- added by 0004
|
||
|
||
-- documents
|
||
CREATE TABLE documents (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||
s3_key TEXT NOT NULL,
|
||
extracted_s3_key TEXT,
|
||
filename TEXT NOT NULL,
|
||
mime TEXT NOT NULL,
|
||
bytes BIGINT NOT NULL DEFAULT 0,
|
||
status TEXT NOT NULL DEFAULT 'queued'
|
||
CHECK (status IN ('queued','extracting','prescreening','ocr','analyzing','done','failed')),
|
||
stage TEXT, -- sub-stage / human label
|
||
refunded BOOLEAN NOT NULL DEFAULT FALSE,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
CREATE INDEX documents_user_created_idx ON documents (user_id, created_at DESC);
|
||
CREATE INDEX documents_status_idx ON documents (status);
|
||
|
||
-- reports
|
||
CREATE TABLE reports (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
document_id UUID NOT NULL UNIQUE REFERENCES documents(id) ON DELETE CASCADE,
|
||
content_json JSONB NOT NULL,
|
||
markdown TEXT NOT NULL,
|
||
model_used TEXT,
|
||
prompt_tokens INTEGER NOT NULL DEFAULT 0,
|
||
eval_tokens INTEGER NOT NULL DEFAULT 0,
|
||
latency_ms INTEGER NOT NULL DEFAULT 0,
|
||
prescreen_result_id UUID REFERENCES prescreen_results(id) ON DELETE SET NULL,
|
||
prescreen_meta JSONB,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
|
||
-- prescreen_results (added by 0006)
|
||
CREATE TABLE prescreen_results (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
|
||
correlation_id UUID NOT NULL,
|
||
contract_type TEXT,
|
||
party_a TEXT,
|
||
party_b TEXT,
|
||
total_amount NUMERIC(18, 2),
|
||
currency TEXT,
|
||
start_date DATE,
|
||
end_date DATE,
|
||
has_penalty_clause BOOLEAN,
|
||
has_termination_clause BOOLEAN,
|
||
has_arbitration BOOLEAN,
|
||
confidence_score NUMERIC(4, 3) CHECK (confidence_score >= 0 AND confidence_score <= 1),
|
||
routing_decision TEXT NOT NULL DEFAULT 'manual_review'
|
||
CHECK (routing_decision IN ('auto_approve','manual_review','deep_analysis')),
|
||
prescreened_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
processing_ms INTEGER,
|
||
extractor_version TEXT NOT NULL DEFAULT 'heuristic-v2'
|
||
CHECK (extractor_version IN ('regex-v1','heuristic-v2','hybrid-llm-v1')),
|
||
auto_summary TEXT,
|
||
auto_findings JSONB NOT NULL DEFAULT '[]',
|
||
error_message TEXT,
|
||
retry_count INTEGER NOT NULL DEFAULT 0 CHECK (retry_count >= 0)
|
||
);
|
||
CREATE INDEX prescreen_results_document_idx ON prescreen_results (document_id);
|
||
CREATE INDEX prescreen_results_routing_idx ON prescreen_results (routing_decision);
|
||
CREATE INDEX prescreen_results_confidence_idx ON prescreen_results (confidence_score);
|
||
|
||
-- jobs (RabbitMQ correlation; one document has up to 3 jobs: extract + prescreen + analyze)
|
||
CREATE TABLE jobs (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
document_id UUID NOT NULL REFERENCES documents(id) ON DELETE CASCADE,
|
||
correlation_id UUID NOT NULL,
|
||
queue TEXT NOT NULL CHECK (queue IN ('extract','prescreen','analyze')),
|
||
attempts INTEGER NOT NULL DEFAULT 0,
|
||
max_attempts INTEGER NOT NULL DEFAULT 5,
|
||
last_failure_class TEXT,
|
||
last_error TEXT,
|
||
dlq BOOLEAN NOT NULL DEFAULT FALSE,
|
||
status TEXT NOT NULL DEFAULT 'pending'
|
||
CHECK (status IN ('pending','running','retrying','dlq','done')),
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||
);
|
||
CREATE INDEX jobs_correlation_idx ON jobs (correlation_id);
|
||
CREATE INDEX jobs_document_idx ON jobs (document_id);
|
||
|
||
-- service_tokens (per-adapter auth)
|
||
CREATE TABLE service_tokens (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
name TEXT NOT NULL UNIQUE, -- "bot-prod", "web-prod"
|
||
token_hash TEXT NOT NULL, -- sha256 hex of the bearer secret
|
||
adapter TEXT NOT NULL CHECK (adapter IN ('bot','web','cli')),
|
||
revoked BOOLEAN NOT NULL DEFAULT FALSE,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
last_used_at TIMESTAMPTZ
|
||
);
|
||
|
||
-- invoices (STUB — no ЮKassa logic this refactor; forward-compatible schema)
|
||
CREATE TABLE invoices (
|
||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||
user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
||
amount INTEGER NOT NULL, -- kopecks
|
||
status TEXT NOT NULL DEFAULT 'draft'
|
||
CHECK (status IN ('draft','pending','succeeded','cancelled','refunded')),
|
||
provider TEXT, -- 'yookassa'
|
||
external_id TEXT,
|
||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||
paid_at TIMESTAMPTZ
|
||
);
|
||
CREATE INDEX invoices_user_idx ON invoices (user_id, created_at DESC);
|
||
```
|
||
|
||
**Alembic rules:** `env.py` uses the async engine; migrations are
|
||
hand-written (autogenerate used only to draft, never committed as-is);
|
||
`up` and `down` both clean; CI runs `alembic upgrade head` then
|
||
`alembic downgrade base` on a throwaway DB.
|
||
|
||
---
|
||
|
||
## 8. Credits & refund policy (billing invariants)
|
||
|
||
The reserve-on-enqueue invariant from T-E1-003 **survives the queue pivot
|
||
unchanged**. RabbitMQ only changes *who runs the job*, not *when the credit
|
||
moves*.
|
||
|
||
### Reserve (api, on enqueue, synchronous)
|
||
|
||
```python
|
||
async def reserve_credit(session, user_id) -> bool:
|
||
row = await session.execute(
|
||
text("""
|
||
UPDATE users SET credits_left = credits_left - 1
|
||
WHERE id = :u AND credits_left > 0
|
||
RETURNING credits_left
|
||
"""),
|
||
{"u": user_id},
|
||
)
|
||
return row.first() is not None
|
||
```
|
||
|
||
Atomic, never negative. If `False`, api returns `402 Payment Required`
|
||
synchronously — no MinIO write, no message published.
|
||
|
||
### Refund (worker, on terminal failure, idempotent)
|
||
|
||
Guarded by `documents.refunded` so a duplicate message / requeue cannot
|
||
double-refund:
|
||
|
||
```python
|
||
NON_REFUNDABLE_INFRA_ONLY = {"extraction_failed"} # user-garbage input
|
||
|
||
|
||
async def refund_credit(session, document_id, failure_class, policy) -> bool:
|
||
if policy == "infra_only" and failure_class in NON_REFUNDABLE_INFRA_ONLY:
|
||
return False # user pays for undetectable garbage
|
||
result = await session.execute(
|
||
text("""
|
||
UPDATE users
|
||
SET credits_left = credits_left + 1
|
||
WHERE id = (SELECT user_id FROM documents
|
||
WHERE id = :d AND refunded = FALSE)
|
||
RETURNING id
|
||
"""),
|
||
{"d": document_id},
|
||
)
|
||
if result.first() is None:
|
||
return False # already refunded (idempotent no-op)
|
||
await session.execute(
|
||
text("UPDATE documents SET refunded = TRUE WHERE id = :d"),
|
||
{"d": document_id},
|
||
)
|
||
return True
|
||
```
|
||
|
||
### Failure classes (`core/db/enums.py`)
|
||
|
||
| Class | Meaning | Refundable under `infra_only`? |
|
||
|---|---|---|
|
||
| `extraction_failed` | pymupdf couldn't extract (>100 chars expected, got less) and OCR also failed | **No** (likely user garbage) |
|
||
| `ocr_failed` | Tesseract raised | Yes |
|
||
| `llm_quota` | 429 / quota on both primary and fallback | Yes |
|
||
| `llm_invalid_output` | model returned unrepairable JSON | Yes |
|
||
| `llm_timeout` | Ollama timed out past retries | Yes |
|
||
| `infra` | DB/S3/MQ connectivity, unhandled exception | Yes |
|
||
| `unknown` | anything not classified | Yes |
|
||
|
||
Set on the `jobs.last_failure_class` column. `REFUND_POLICY` env selects
|
||
`all` (default) or `infra_only`.
|
||
|
||
---
|
||
|
||
## 9. LLM provider port + adapters
|
||
|
||
The prototype's `llm_client.py` (bearer httpx, `format: json-schema`,
|
||
5xx backoff, 429→fallback, repair-loop) is excellent. It becomes the first
|
||
adapter behind a port so we can later swap in self-hosted Ollama,
|
||
GigaChat, or YandexGPT without touching `analyzer.py`.
|
||
|
||
Registered providers are selected by `LLM_PROVIDER`:
|
||
- `ollama_cloud` (default)
|
||
- `yandex_gpt`
|
||
|
||
### Port (`core/llm/port.py`)
|
||
|
||
```python
|
||
from typing import Protocol, Sequence
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class AnalysisResult:
|
||
findings: list # list[Finding] (from core.analysis.report_schema)
|
||
model_used: str
|
||
fell_back: bool
|
||
repaired: bool
|
||
prompt_tokens: int
|
||
eval_tokens: int
|
||
latency_sec: float
|
||
|
||
|
||
class LLMProvider(Protocol):
|
||
async def analyze(
|
||
self, text: str, *, checklist: str, extra_context: str = ""
|
||
) -> AnalysisResult: ...
|
||
async def aclose(self) -> None: ...
|
||
```
|
||
|
||
### Adapter (`core/llm/ollama_cloud.py`)
|
||
|
||
Move the existing `OllamaCloudClient` + `ChatResult` + `_QuotaError` +
|
||
`_run_with_fallback` + `_run_repair_loop` + `_post_chat` verbatim, then add
|
||
an `analyze(text, *, checklist)` method that:
|
||
|
||
1. chunks the text via `core.analysis.chunker.chunk_markdown` (heading-aware when
|
||
the extractor produced structured Markdown; falls back to plain chunking),
|
||
2. appends `extra_context` (prescreen metadata for `AnalyzeRequested`) to each user prompt,
|
||
3. builds the system/user prompts (from `prototype.SYSTEM_PROMPT` +
|
||
`build_user_prompt`, lifted into `core/analysis/analyzer.py`),
|
||
4. fans out chunk requests under the existing `asyncio.Semaphore`,
|
||
5. merges/dedupes/sorts findings (lift `dedupe_findings` + `sort_findings`
|
||
into `core/analysis/analyzer.py`),
|
||
6. returns `AnalysisResult`.
|
||
|
||
Map adapter-internal failures to `FailureClass`:
|
||
|
||
- `_QuotaError` escapes → `llm_quota`
|
||
- `ValidationError` after repair → `llm_invalid_output`
|
||
- `httpx.TimeoutException` → `llm_timeout`
|
||
- anything else → `infra`
|
||
|
||
### Factory (`core/llm/factory.py`)
|
||
|
||
```python
|
||
def build_llm_provider(settings) -> LLMProvider:
|
||
match settings.llm_provider:
|
||
case "ollama_cloud":
|
||
return OllamaCloudProvider(settings)
|
||
case "yandex_gpt":
|
||
return YandexGPTProvider(settings)
|
||
case _:
|
||
raise ValueError(f"unknown LLM_PROVIDER={settings.llm_provider!r}")
|
||
```
|
||
|
||
`LLM_PROVIDER` env (default `ollama_cloud`). Additional providers register here.
|
||
|
||
---
|
||
|
||
## 10. Durability & HA posture (single VPS, HA-ready)
|
||
|
||
**Confirmed posture:** HA-*ready* on one VPS, not HA-*running*. True
|
||
mirrored queues and Postgres replication require multiple hosts; on one VPS
|
||
we get **durability** (survives reboot/crash) and **recoverability**
|
||
(point-in-time), and the path to true HA is a topology change, not a code
|
||
change.
|
||
|
||
### RabbitMQ — quorum queues
|
||
|
||
- `extract.q`, `prescreen.q`, `analyze.q`, `extract.dlq`, `prescreen.dlq`,
|
||
`analyze.dlq` are `x-queue-type: quorum`. Quorum queues persist every message
|
||
to disk and use Raft. On a single node they are durable; the moment a 3-node
|
||
RabbitMQ cluster is added (compose: `rabbitmq@rmq1/2/3`), the same queues
|
||
replicate with no application code change.
|
||
- Retry queues are `classic` (transient delay slots; safe to lose on
|
||
catastrophic failure — the originating message is already acked and
|
||
tracked in `jobs`).
|
||
- Enable **publisher confirms** on the api publisher (await confirm per
|
||
publish; if nacked/timeout, fail the request before telling the user
|
||
202). A paid job's message must never silently vanish.
|
||
|
||
### Postgres — replication-ready
|
||
|
||
- `wal_level=replica`, `archive_mode=on`, `archive_command` to a mounted
|
||
volume (or `pg_backrest` later). Gives PITR.
|
||
- A physical replication slot + hot standby is a documented add-host step
|
||
(add `postgres-replica` service, `primary_conninfo`). Patroni/repmgr when
|
||
automated failover is wanted.
|
||
- Daily `pg_dump` cron in compose (sidecar or host cron) — baseline backup.
|
||
|
||
### Volumes & restart
|
||
|
||
- Named volumes for `pgdata`, `rabbitmq`, `minio`, `redis`, `prometheus`,
|
||
`grafana`, `tempo`. Bind-mount only `./deploy` configs and cert dirs.
|
||
- `restart: unless-stopped` on every long-running service.
|
||
- `depends_on: condition: service_healthy` everywhere with real healthchecks
|
||
(pg `pg_isready`, rabbit `rabbitmq-diagnostics ping`, minio `mc ready`,
|
||
redis `redis-cli ping`).
|
||
|
||
### Scale-out checklist (when leaving one VPS)
|
||
|
||
1. Cluster RabbitMQ to 3 nodes across hosts (quorum queues auto-replicate).
|
||
2. Add Postgres replica + Patroni (automated failover).
|
||
3. Run ≥2 api, ≥2 worker-extract, ≥2 worker-analyze behind the same
|
||
Rabbit/PG/MinIO (stateless services scale horizontally for free).
|
||
4. Move MinIO to distributed mode (≥4 nodes, erasure coding).
|
||
5. Reconsider Nginx → managed LB.
|
||
|
||
None of 1–5 requires touching `core/` application code — only compose/infra.
|
||
|
||
---
|
||
|
||
## 11. Configuration (`core/config.py`, pydantic-settings)
|
||
|
||
12-factor: all config via env. Typed via `pydantic-settings`. Base `Settings`
|
||
+ per-service subclasses. No hardcoded model names, prompts, delays.
|
||
|
||
### Base (shared by all services)
|
||
|
||
| Env | Default | Notes |
|
||
|---|---|---|
|
||
| `ENV` | `dev` | dev/staging/prod — toggles TLS, sentry sample rate |
|
||
| `LOG_LEVEL` | `INFO` | structlog level |
|
||
| `LOG_FORMAT` | `json` | `json` (prod/staging) or `console` (dev) — explicit override of env-based default |
|
||
| `APP_VERSION` | `unknown` | added to every log line; set at build/deploy time |
|
||
| `DATABASE_URL` | (required) | `postgresql+asyncpg://...` |
|
||
| `REDIS_URL` | `redis://redis:6379/0` | rate limit / sessions (future) |
|
||
| `RABBITMQ_URL` | (required) | `amqp://guest:guest@rabbitmq:5672//` |
|
||
| `MQ_PREFETCH_EXTRACT` | `1` | CPU-bound |
|
||
| `MQ_PREFETCH_ANALYZE` | `3` | mirrors Ollama Pro concurrency |
|
||
| `MQ_PREFETCH_PRESCREEN` | `1` | deterministic heuristic is tiny but keep ordered |
|
||
| `MQ_MAX_ATTEMPTS` | `5` | before DLQ |
|
||
| `MQ_RETRY_BASE_MS` | `2000` | exponential base |
|
||
| `S3_ENDPOINT_URL` | (required) | MinIO URL |
|
||
| `S3_ACCESS_KEY` / `S3_SECRET_KEY` | (required) | |
|
||
| `S3_BUCKET` | `contract-check-docs` | |
|
||
| `S3_REGION` | `us-east-1` | MinIO default |
|
||
| `S3_SERVER_SIDE_ENCRYPTION` | `false` | true in prod |
|
||
| `DOC_RETENTION_DAYS` | `7` | MinIO ILM expiry |
|
||
| `TEXT_RETENTION_DAYS` | `30` | extracted `.txt` expiry |
|
||
| `REFUND_POLICY` | `all` | `all` \| `infra_only` |
|
||
| `SENTRY_DSN` | (empty) | if set, init sentry |
|
||
| `OTEL_EXPORTER_OTLP_ENDPOINT` | (empty) | OTel collector |
|
||
| `OTEL_SERVICE_NAME` | per-service | overridden in each service settings |
|
||
|
||
### LLM
|
||
|
||
| Env | Default | Notes |
|
||
|---|---|---|
|
||
| `LLM_PROVIDER` | `ollama_cloud` | factory selection: `ollama_cloud`, `yandex_gpt` |
|
||
| `OLLAMA_HOST` | (required if provider=ollama_cloud) | `https://ollama.com` for Ollama Cloud; `http://localhost:11434` for self-hosted |
|
||
| `OLLAMA_API_KEY` | (required) | bearer |
|
||
| `OLLAMA_MODEL` | `qwen2.5:14b` | primary; must be a model available at the configured host |
|
||
| `OLLAMA_FALLBACK_MODEL` | `qwen2.5:7b` | 429 fallback; must also exist on the host |
|
||
| `OLLAMA_TEMPERATURE` | `0.2` | |
|
||
| `OLLAMA_NUM_PREDICT` | `3072` | |
|
||
| `OLLAMA_TIMEOUT` | `120` | seconds |
|
||
| `OLLAMA_MAX_CONCURRENCY` | `3` | in-process semaphore (per analyze worker) |
|
||
| `CHUNK_SIZE_CHARS` | `10000` | chunker |
|
||
| `YANDEXGPT_API_KEY` | (required if provider=yandex_gpt) | API key for Yandex Cloud Foundation Models |
|
||
| `YANDEXGPT_FOLDER_ID` | (required if provider=yandex_gpt) | Yandex Cloud folder ID (`gpt://<folder_id>/<model>`) |
|
||
| `YANDEXGPT_MODEL` | `yandexgpt-lite` | primary YandexGPT model URI name |
|
||
| `YANDEXGPT_FALLBACK_MODEL` | (empty) | optional fallback model name |
|
||
| `YANDEXGPT_BASE_URL` | `https://llm.api.cloud.yandex.net` | completion API base URL |
|
||
| `YANDEXGPT_COMPLETION_PATH` | `/foundationModels/v1/completion` | completion API path |
|
||
| `YANDEXGPT_TEMPERATURE` | `0.2` | |
|
||
| `YANDEXGPT_MAX_TOKENS` | `3072` | max completion tokens |
|
||
| `YANDEXGPT_TIMEOUT` | `120` | seconds |
|
||
| `YANDEXGPT_MAX_CONCURRENCY` | `3` | in-process semaphore (per analyze worker) |
|
||
|
||
### Prescreen
|
||
|
||
| Env | Default | Notes |
|
||
|---|---|---|
|
||
| `PRESCREEN_ENABLED` | `true` | Run the prescreen stage between extraction and analysis. |
|
||
| `PRESCREEN_AUTO_APPROVE` | `false` | Allow auto-approval of low-risk contracts (disabled until accuracy proven). |
|
||
| `PRESCREEN_CONFIDENCE_THRESHOLD` | `0.75` | Minimum weighted field-coverage score to route away from `manual_review`. |
|
||
| `PRESCREEN_HIGH_VALUE_THRESHOLD` | `100000` | Amount (in contract currency) above which contract is always sent to deep analysis. |
|
||
| `PRESCREEN_LLM_FALLBACK_ENABLED` | `false` | Kill-switch for the Stage-2 LLM fallback in the hybrid extractor. Flip after burn-in. |
|
||
| `PRESCREEN_LLM_FALLBACK_THRESHOLD` | `0.75` | Heuristic confidence below this triggers the LLM fallback (keep ≤ confidence threshold). |
|
||
| `PRESCREEN_LLM_MAX_CHARS` | `20000` | Cap on characters sent to the LLM fallback (cost control). |
|
||
| `PRESCREEN_KEEP_REGEX` | `false` | Temporary: run the legacy regex-v1 extractor via `extract_contract_meta` shim. |
|
||
|
||
### API
|
||
|
||
| Env | Default | Notes |
|
||
|---|---|---|
|
||
| `API_HOST` | `0.0.0.0` | |
|
||
| `API_PORT` | `8000` | |
|
||
| `API_METRICS_PORT` | `9100` | |
|
||
| `B2B_DEFAULT_RATE_LIMIT_RPS` | `3` | per API key; mirrors Ollama Pro concurrency, overridable per `api_keys.rate_limit_rps` |
|
||
| `CORS_ORIGINS` | (empty, future web) | |
|
||
|
||
### Auth (JWT + Telegram identity verification + webUI + admin panel)
|
||
|
||
| Env | Default | Notes |
|
||
|---|---|---|
|
||
| `TELEGRAM_BOT_TOKEN` | (empty) | used by API to verify Login Widget / Mini App signatures |
|
||
| `JWT_SECRET` | (required) | HS256 secret for signing user JWTs; generate with `openssl rand -hex 32` |
|
||
| `JWT_ALGORITHM` | `HS256` | |
|
||
| `JWT_ACCESS_TTL_MINUTES` | `1440` (24h) | access-token lifetime; tune per env |
|
||
| `JWT_REFRESH_TTL_DAYS` | `30` | refresh-token lifetime for webUI sessions |
|
||
| `WEB_AUTH_ENABLED` | `true` | toggle `/api/v1/auth/{register,login,logout,forgot-password,reset-password}` |
|
||
| `WEB_APP_BASE_URL` | `http://localhost:5173` | base URL of the future web SPA; used for password-reset links |
|
||
| `PASSWORD_RESET_TTL_MINUTES` | `60` | reset-link validity |
|
||
| `PASSWORD_MIN_LENGTH` | `8` | enforced at register, reset, and in the `/admin` create form |
|
||
| `WEB_ADMIN_ENABLED` | `true` | toggle the `/admin/*` server-rendered management UI |
|
||
| `ADMIN_REQUIRED_ROLE` | `admin` | `users.role` value required to enter `/admin` (must match the DB CHECK constraint) |
|
||
|
||
### Bot
|
||
|
||
| Env | Default |
|
||
|---|---|
|
||
| `BOT_TOKEN` | (required) |
|
||
| `API_URL` | `http://api:8000` |
|
||
| `BOT_SERVICE_TOKEN` | (required — bearer for adapter auth to `/api/v1/auth/telegram/bot`, looked up against `service_tokens`) |
|
||
|
||
### Prototype (standalone benchmark image)
|
||
|
||
Same Ollama env as above; no DB/MQ/S3 env needed.
|
||
|
||
---
|
||
|
||
## 12. Docker — images & compose
|
||
|
||
### Per-service Dockerfile pattern (uv, multi-stage, py3.13)
|
||
|
||
Dockerfiles live in `srv/<service>/Dockerfile` (one per service). Common shape
|
||
(shown for api):
|
||
|
||
```dockerfile
|
||
# syntax=docker/dockerfile:1
|
||
FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim AS builder
|
||
ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_PYTHON_DOWNLOADS=never \
|
||
UV_PROJECT_ENVIRONMENT=/app/.venv
|
||
WORKDIR /app
|
||
COPY pyproject.toml uv.lock ./
|
||
COPY README.md ./
|
||
COPY src ./src
|
||
RUN --mount=type=cache,target=/root/.cache/uv \
|
||
uv sync --frozen --no-default-groups --group api
|
||
|
||
FROM python:3.13-slim AS runtime
|
||
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PATH=/app/.venv/bin:$PATH
|
||
WORKDIR /app
|
||
COPY --from=builder /app/.venv /app/.venv
|
||
# Tesseract ONLY in srv/worker-extract/Dockerfile:
|
||
# RUN apt-get update && apt-get install -y --no-install-recommends \
|
||
# tesseract-ocr tesseract-ocr-rus tesseract-ocr-eng \
|
||
# && rm -rf /var/lib/apt/lists/*
|
||
EXPOSE 8000
|
||
CMD ["python", "-m", "contract_check.api"]
|
||
```
|
||
|
||
Per-service differences (`uv` uses PEP 735 dependency-groups — see `pyproject.toml`):
|
||
|
||
| Dockerfile | Extra installed | Runtime apt | CMD | Expose |
|
||
|---|---|---|---|---|
|
||
| `srv/api/Dockerfile` | `--group api` | none | `python -m contract_check.api` | 8000, 9100 |
|
||
| `srv/worker-extract/Dockerfile` | `--group extract` | tesseract-ocr, -rus, -eng, libmagic1 | `python -m contract_check.worker_extract` | 9101 |
|
||
| `srv/worker-prescreen/Dockerfile` | `--group prescreen` | none | `python -m contract_check.worker_prescreen` | 9104 |
|
||
| `srv/worker-analyze/Dockerfile` | `--group analyze` | none | `python -m contract_check.worker_analyze` | 9102 |
|
||
| `srv/bot/Dockerfile` | `--group bot` | none | `python -m contract_check.bot` | — |
|
||
| `srv/prototype/Dockerfile` | `--group prototype` | none | `python -m contract_check.prototype` | — |
|
||
|
||
The bot image is the leanest (no DB driver, no S3 client, no pymupdf). The
|
||
analyze image has httpx but no tesseract/pymupdf. The extract image is the
|
||
heaviest (tesseract + language packs + libmagic1). This is the "fine-tuned deps
|
||
per service" payoff.
|
||
|
||
### pyproject.toml dependency-groups (actual — PEP 735)
|
||
|
||
The project uses `[dependency-groups]` (not `[project.optional-dependencies]`).
|
||
Core deps (installed in every image) stay minimal; service-specific libs live
|
||
in groups so each Docker image runs `uv sync --no-default-groups --group <svc>`.
|
||
Shared `db`/`mq`/`s3`/`obs` groups are `include-group`-ed by the service groups
|
||
that need them. Sketch (see `pyproject.toml` for the authoritative list):
|
||
|
||
```toml
|
||
[project]
|
||
name = "contract-check"
|
||
requires-python = ">=3.13"
|
||
dependencies = [
|
||
"pydantic>=2.7", "pydantic-settings>=2.3", "structlog>=24.1",
|
||
"python-dotenv>=1.0", "httpx[http2]>=0.27",
|
||
]
|
||
|
||
[dependency-groups]
|
||
db = ["sqlalchemy>=2.0", "asyncpg>=0.29", "alembic>=1.13"]
|
||
mq = ["aio-pika>=9.4"]
|
||
s3 = ["minio>=7.2"]
|
||
obs = ["prometheus-client>=0.20", "sentry-sdk>=2",
|
||
"opentelemetry-sdk>=1.24", "opentelemetry-exporter-otlp>=1.24"]
|
||
|
||
api = [{ include-group = "db" }, { include-group = "mq" },
|
||
{ include-group = "s3" }, { include-group = "obs" },
|
||
"fastapi>=0.110", "uvicorn[standard]>=0.29", "python-multipart>=0.0.9",
|
||
"redis>=5.0",
|
||
"opentelemetry-instrumentation-fastapi>=0.45b0",
|
||
"opentelemetry-instrumentation-asgi>=0.45b0"]
|
||
extract = [{ include-group = "db" }, { include-group = "mq" },
|
||
{ include-group = "s3" }, { include-group = "obs" },
|
||
"pymupdf>=1.24", "pytesseract>=0.3.10", "pillow>=10",
|
||
"mammoth>=1.8", "striprtf>=0.0.26", "chardet>=5.2",
|
||
"python-magic>=0.4.27"]
|
||
analyze = [{ include-group = "db" }, { include-group = "mq" },
|
||
{ include-group = "s3" }, { include-group = "obs" },
|
||
"opentelemetry-instrumentation-httpx>=0.45b0"]
|
||
prescreen = [{ include-group = "db" }, { include-group = "mq" },
|
||
{ include-group = "s3" }, { include-group = "obs" }]
|
||
bot = ["aiogram>=3.4"]
|
||
prototype = ["pymupdf>=1.24", "python-docx>=1.1"]
|
||
dev = [{ include-group = "api" }, { include-group = "extract" },
|
||
{ include-group = "prescreen" }, { include-group = "analyze" },
|
||
{ include-group = "bot" }, { include-group = "prototype" },
|
||
"pytest>=8", "pytest-asyncio>=0.23", "respx>=0.21", "ruff>=0.5",
|
||
"mypy>=1.10", "anyio>=4", "aiosqlite>=0.20",
|
||
"testcontainers[rabbitmq,postgres,minio]>=4", "asgi-lifespan>=2.1.0"]
|
||
```
|
||
|
||
### docker-compose.yml structure
|
||
|
||
One file, profiles. Default (`docker compose up`) = infra only. Services
|
||
behind `--profile services`. Observability behind `--profile obs`. Edge
|
||
behind `--profile edge`.
|
||
|
||
```
|
||
services:
|
||
# ── INFRA (default profile) ──
|
||
postgres: postgres:18-alpine, wal_level=replica, archive, healthcheck, volumes (pgdata, pgwal); host port 15432
|
||
redis: redis:8-alpine, aof, healthcheck, volume; host port 17379
|
||
rabbitmq: rabbitmq:4-management-alpine, healthcheck, volume; AMQP 5672, mgmt UI 15672
|
||
minio: minio/minio, healthcheck, volume, console on :9001; S3 API :9000
|
||
minio-init: one-shot: mc alias + mb + ilm rule; depends_on minio healthy
|
||
|
||
# ── SERVICES (profile: services) ──
|
||
api: build srv/api/Dockerfile; depends_on pg/rabbit/minio-init healthy;
|
||
healthcheck /healthz; ports 8000,9100
|
||
worker-extract: build srv/worker-extract/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9101
|
||
worker-prescreen: build srv/worker-prescreen/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9104
|
||
worker-analyze: build srv/worker-analyze/Dockerfile; depends_on pg/rabbit/minio-init healthy; 9102
|
||
bot: build srv/bot/Dockerfile; depends_on api healthy (NOT pg/rabbit)
|
||
|
||
# ── OBSERVABILITY (profile: obs) ── PLANNED (T-E1-010)
|
||
prometheus: prom/prometheus; scrape api:9100, extract:9101, analyze:9102
|
||
grafana: grafana/grafana; provisioned datasources + starter dashboards
|
||
otel-collector: otel/opentelemetry-collector-contrib; receives OTLP
|
||
tempo: grafana/tempo; trace storage
|
||
|
||
# ── EDGE (profile: edge) ── PLANNED (T-E1-009)
|
||
nginx: nginx:alpine; reverse proxy → api; TLS via certbot
|
||
certbot: certbot/certbot; renew cron sidecar
|
||
|
||
volumes: { pgdata, pgwal, redisdata, rabbitmq, minio }
|
||
```
|
||
|
||
`bot` depends on `api` healthy (not on infra) — it speaks HTTP to the api,
|
||
enforcing the adapter boundary even in dependency ordering.
|
||
|
||
---
|
||
|
||
## 13. Observability
|
||
|
||
### Structured logging (`core/logging.py`)
|
||
|
||
- `structlog` with JSON renderer in prod, console renderer in dev.
|
||
- A `correlation_id` `contextvars.ContextVar` is the spine of every log line.
|
||
- **api** middleware reads `X-Correlation-ID` header or mints a UUID4, sets
|
||
the contextvar, and the api publisher stamps `headers["x-correlation-id"]`
|
||
on every published message.
|
||
- **workers** read `headers["x-correlation-id"]` on consume and set the same
|
||
contextvar before handling. So a single upload's logs trace
|
||
api → rabbit → worker-extract → worker-analyze → DB under one ID.
|
||
- OTel baggage/span context propagates the same ID for distributed traces.
|
||
|
||
### Sentry (`core/sentry.py`)
|
||
|
||
- `sentry_sdk.init(dsn=SENTRY_DSN, environment=ENV, traces_sample_rate=...)`
|
||
in each service's entrypoint.
|
||
- FastAPI integration in api, asyncio integration in workers.
|
||
- Sample rate: 1.0 in dev, 0.1 in prod (env-driven).
|
||
|
||
### Prometheus (`core/metrics.py` + `/metrics`)
|
||
|
||
- api: `http_requests_total`, `http_request_duration_seconds`,
|
||
`documents_uploaded_total`, `credits_reserved_total`, `mq_publish_total`.
|
||
- worker-extract: `extract_duration_seconds{format=...}`,
|
||
`extraction_total{format,structured}`, expose on `:9101`.
|
||
- worker-prescreen: `prescreen_duration_seconds{decision=...}`,
|
||
`prescreen_runs_total{decision,contract_type}`, `prescreen_confidence`,
|
||
`prescreen_fallback_runs_total{outcome=used|failed|skipped|disabled}`,
|
||
`prescreen_extraction_stage_seconds{stage=heuristic|llm}`,
|
||
expose on `:9104`.
|
||
- worker-analyze: `analyze_duration_seconds`,
|
||
`llm_tokens_total{kind=prompt|eval}`, `llm_fell_back_total`,
|
||
expose on `:9102`.
|
||
- Workers run `prometheus_client.start_http_server(port)` in a background
|
||
thread alongside the async consumer.
|
||
|
||
### Grafana
|
||
|
||
- Provisioned Prometheus + Tempo datasources.
|
||
- Starter dashboard: queue depth (`rabbitmq_queue_messages`), job duration
|
||
histogram, LLM tokens/min, refund rate, 429/fallback rate.
|
||
|
||
### OpenTelemetry (`core/telemetry.py`)
|
||
|
||
- OTLP exporter to `otel-collector` → Tempo.
|
||
- Auto-instrument FastAPI (api), httpx (all outbound, incl. Ollama calls).
|
||
- Spans carry the same `correlation_id` as logs. Trace from HTTP request →
|
||
RabbitMQ publish → consume → LLM call is one trace tree.
|
||
|
||
---
|
||
|
||
## 14. Testing strategy
|
||
|
||
### Unit (`tests/unit/`, fast, no I/O)
|
||
|
||
- `test_chunker.py` — `chunk_text` + `chunk_markdown`.
|
||
- `test_extraction_adapters.py`, `test_extraction_factory.py` — adapters and factory.
|
||
- `test_extractor.py` — legacy prototype extractor path remains valid.
|
||
- `test_llm_ollama_cloud.py` — **respx** mocks: 200 happy, 429→fallback,
|
||
invalid-JSON→repair→success, invalid-JSON→repair→fail, timeout.
|
||
- `test_credits_policy.py` — `reserve_credit` never negative; `refund_credit`
|
||
idempotent (double-call refunds once); `infra_only` skips
|
||
`extraction_failed`; concurrency: 5 parallel reserves against 1 credit →
|
||
exactly 1 succeeds.
|
||
- `test_messages.py` — pydantic round-trip + header validation.
|
||
- **adapter-boundary test**: `bot/` must not import any `core.db`/`core.s3`/
|
||
`core.llm`/`core.mq`/`core.credits` symbol (static AST check).
|
||
|
||
### Integration (`tests/integration/`, testcontainers, slow)
|
||
|
||
- `conftest.py` fixtures start **real** Postgres + RabbitMQ + MinIO via
|
||
`testcontainers` (one container each per session), apply migrations, wire
|
||
to a throwaway MinIO bucket and RabbitMQ vhost.
|
||
- `test_upload_pipeline.py` — `POST /api/v1/documents` (TestClient) →
|
||
credit reserved, blob in MinIO, `DocumentUploaded` lands on `extract.q`
|
||
(assert via `aio-pika` consumer).
|
||
- `test_extract_worker.py` — publish `DocumentUploaded`, run
|
||
worker-extract handler in-process → `PrescreenRequested` published,
|
||
extracted **Markdown** in MinIO, `documents.status=advanced to prescreening`.
|
||
- `test_analyze_worker.py` — publish `AnalyzeRequested` with respx-mocked
|
||
Ollama, run worker-analyze handler → report row in Postgres with
|
||
`status=done`, markdown contains disclaimer.
|
||
- `test_retry_and_dlq.py` — force Ollama 500 repeatedly → message cycles
|
||
retry queues with growing TTL, lands on `analyze.dlq` after
|
||
`MAX_ATTEMPTS`, `documents.status=failed`, credit refunded per policy.
|
||
|
||
Marks: `@pytest.mark.integration` excluded from the default fast run; CI
|
||
runs both. `pytest -q` (unit) < 5s; integration as a separate stage.
|
||
|
||
### DoD per step
|
||
|
||
Every landing step ends with all three green:
|
||
|
||
```bash
|
||
ruff check . && mypy src && pytest -q # unit, fast
|
||
pytest -m integration -q # integration, CI only
|
||
```
|
||
|
||
---
|
||
|
||
## 15. API surface (FastAPI)
|
||
|
||
JSON API routes live under `/api/v1`. The server-rendered **admin panel**
|
||
is mounted at `/admin/*` and is the only HTML surface in the api; it reuses
|
||
the user JWT for authentication (stored in an HttpOnly cookie).
|
||
|
||
Three JSON auth modes:
|
||
|
||
- **User JWT** (`Authorization: Bearer <jwt>`) — common token for bot users,
|
||
Telegram Login Widget users, and Telegram Mini App users. Issued by
|
||
`/api/v1/auth/telegram/*` after verifying the Telegram identity proof
|
||
(`core/auth.py`). User endpoints rely on `api/deps.py:require_current_user`.
|
||
- **Service token** (`Authorization: Bearer <token>`) — adapter-level auth,
|
||
validated against `service_tokens` (`core/tokens.py`). Used only by the
|
||
bot adapter to call `/api/v1/auth/telegram/bot` and exchange a verified
|
||
`telegram_id` for a user JWT.
|
||
- **B2B API key** (`X-API-Key`) — validated against `api_keys.key_hash`
|
||
(`core/api_keys.py` + `api/deps.py:require_api_key`) with per-key token-bucket
|
||
rate-limit (`core/rate_limit.py`). Used by external B2B clients.
|
||
|
||
Health/metrics exempt from auth.
|
||
|
||
### Auth endpoints (`api/routes/auth.py`)
|
||
|
||
| Method | Path | Auth | Behavior |
|
||
|---|---|---|---|
|
||
| POST | `/api/v1/auth/telegram/bot` | service token | bot exchanges verified `telegram_id` for a user JWT |
|
||
| POST | `/api/v1/auth/telegram/web` | — | verify Telegram Login Widget payload → issue user JWT |
|
||
| POST | `/api/v1/auth/telegram/miniapp` | — | verify Mini App `initData` HMAC → issue user JWT |
|
||
| POST | `/api/v1/auth/register` | — | email/password → new user + JWT pair (requires `WEB_AUTH_ENABLED`) |
|
||
| POST | `/api/v1/auth/login` | — | email/password → JWT pair (requires `WEB_AUTH_ENABLED`) |
|
||
| POST | `/api/v1/auth/logout` | refresh token | revoke refresh (requires `WEB_AUTH_ENABLED`) |
|
||
| POST | `/api/v1/auth/forgot-password` | — | enqueue reset email (requires `WEB_AUTH_ENABLED`) |
|
||
| POST | `/api/v1/auth/reset-password` | reset token | rotate password (requires `WEB_AUTH_ENABLED`) |
|
||
| GET | `/api/v1/auth/me` | user JWT | introspect JWT claims |
|
||
|
||
### User endpoints (user JWT)
|
||
|
||
| Method | Path | Auth | Behavior |
|
||
|---|---|---|---|
|
||
| GET | `/healthz` | none | 200 liveness (process alive) |
|
||
| GET | `/readyz` | none | 200 readiness (DB+Rabbit+MinIO reachable) |
|
||
| GET | `/metrics` | none | Prometheus exposition |
|
||
| POST | `/api/v1/documents` | user JWT | multipart → reserve credit → MinIO put → row `queued` → publish `DocumentUploaded` → `202 {document_id, correlation_id}`. `402` if no credit. `400` bad mime/size. |
|
||
| GET | `/api/v1/documents/{id}` | user JWT | status + stage + filename (for polling UI) |
|
||
| GET | `/api/v1/reports/{document_id}` | user JWT | `202 {status, stage}` while not done; `200 {markdown, findings, ...}` when done |
|
||
| GET | `/api/v1/me` | user JWT | `{telegram_id, credits_left}` |
|
||
|
||
### B2B endpoints (`X-API-Key`, `api/routes/b2b.py`)
|
||
|
||
| Method | Path | Auth | Behavior |
|
||
|---|---|---|---|
|
||
| POST | `/api/v1/analyze` | api key | multipart → reserve owner's credit → publish → `202 {document_id, correlation_id}`; rate-limited (`429`), `401` bad/revoked key |
|
||
| GET | `/api/v1/b2b/reports/{document_id}` | api key | scoped to key owner; `202 {status, stage}` or `200 {markdown, findings, ...}` |
|
||
| GET | `/api/v1/b2b/usage` | api key | `{monthly_quota, monthly_used, requests_this_month, resets_at}` for the authenticating key |
|
||
| POST | `/api/v1/b2b/keys` | user JWT | create key; raw `api_key` returned **once** (only hash stored) |
|
||
| GET | `/api/v1/b2b/keys` | user JWT | list owner's keys |
|
||
| POST | `/api/v1/b2b/keys/{id}/revoke` | user JWT | revoke (instant auth disable) |
|
||
| GET | `/api/v1/b2b/keys/{id}/usage` | user JWT | per-month request counts |
|
||
|
||
### Admin panel (`api/admin/`)
|
||
|
||
Mounted at `/admin/*`, gated by `WEB_ADMIN_ENABLED` and by the
|
||
`ADMIN_REQUIRED_ROLE` check on `users.role`. Stack: FastAPI + Jinja2 + HTMX
|
||
(client-side via CDN). Reuses the same access JWT, but stored in an
|
||
HttpOnly cookie (`cc_admin_token`) so browsers can navigate it.
|
||
|
||
| Method | Path | Behavior |
|
||
|---|---|---|
|
||
| GET/POST | `/admin/login` | email/password → cookie; non-`admin` roles rejected |
|
||
| POST | `/admin/logout` | clear cookie |
|
||
| GET | `/admin` / `/admin/` | redirect to `/admin/users` if logged in, else `/admin/login` |
|
||
| GET | `/admin/users` | paginated, searchable user list |
|
||
| GET | `/admin/users/new` | create-user form |
|
||
| POST | `/admin/users` | create a user (email, password, role, credits, optional telegram_id) |
|
||
| GET | `/admin/users/{user_id}` | user detail + document stats |
|
||
| POST | `/admin/users/{user_id}` | update role / active flag / credits |
|
||
| POST | `/admin/users/{user_id}/toggle-active` | ban/unban (HTMX partial swap) |
|
||
| GET | `/admin/users/{user_id}/credits` | bump credits by `delta` (HTMX partial swap) |
|
||
|
||
To grant admin access: set `users.role = 'admin'` (or the value of
|
||
`ADMIN_REQUIRED_ROLE`) for the target account. The Makefile has helpers:
|
||
`make admin-promote EMAIL=...` and `make admin-list`.
|
||
|
||
No synchronous `/analyze` (locked). Adapters poll `/reports/{id}`; the
|
||
fine-grained `stage` field powers a progress signal in the bot ("Extracting
|
||
text…", "Analyzing…"). SSE/webhook added later.
|
||
|
||
---
|
||
|
||
## 16. Worker internals
|
||
|
||
### worker-extract (`worker_extract/handler.py`)
|
||
|
||
```
|
||
on DocumentUploaded(msg):
|
||
if documents.status by msg.document_id is terminal: ack; return
|
||
set status=extracting (or ocr), jobs.status=running
|
||
download s3_key from MinIO → temp file
|
||
try: text = extractor.extract_text(tmp)
|
||
ocr_used = False
|
||
except ExtractionError (<100 chars, likely scan):
|
||
set status=ocr
|
||
text = ocr.ocr_pdf(tmp); ocr_used = True
|
||
if still <100: raise with failure_class=extraction_failed
|
||
upload extracted_key=text to MinIO
|
||
publish DocumentExtracted(correlation_id, document_id, extracted_s3_key,
|
||
char_count=len(text), ocr_used, attempt)
|
||
ack
|
||
on failure: consumer-base retry/dlq logic (§5), refund per policy (§8)
|
||
```
|
||
|
||
### worker-analyze (`worker_analyze/handler.py`)
|
||
|
||
```
|
||
on DocumentExtracted(msg):
|
||
if documents.status terminal: ack; return
|
||
set status=analyzing, jobs.status=running
|
||
text = s3.get(extracted_s3_key)
|
||
result = llm_provider.analyze(text, checklist=checklist_for_prompt())
|
||
validate result.findings via ReportPayload (repair already in provider)
|
||
markdown = analyzer.render_markdown(result.findings, ...)
|
||
save report (content_json, markdown, tokens, latency, model_used)
|
||
set documents.status=done
|
||
ack
|
||
on failure: classify failure_class, retry/dlq, refund per policy
|
||
```
|
||
|
||
LLM provider failures bubble as `FailureClass`; the consumer-base catches
|
||
and routes to retry/DLQ/refund.
|
||
|
||
### Concurrency reminders
|
||
|
||
- worker-extract: RabbitMQ prefetch caps jobs (CPU). Tesseract/pymupdf are
|
||
sync → run in `asyncio.to_thread`/threadpool within the handler.
|
||
- worker-analyze: RabbitMQ prefetch (3) caps jobs; the provider's internal
|
||
`asyncio.Semaphore(OLLAMA_MAX_CONCURRENCY=3)` caps parallel chunk LLM
|
||
calls within a single multi-chunk contract. Both layers matter.
|
||
|
||
---
|
||
|
||
## 17. Bot adapter (`bot/`)
|
||
|
||
aiogram 3. Pure HTTP client to the api. **Forbidden imports**: `core.db`,
|
||
`core.s3`, `core.llm`, `core.mq`, `core.credits`, `sqlalchemy`, `minio`,
|
||
`aio_pika`, `pymupdf`. Enforced by a unit test (§14).
|
||
|
||
Flow:
|
||
|
||
- `/start` → `GET /api/v1/me` → greeting + credit balance.
|
||
- User sends PDF/DOCX → forward multipart to `POST /api/v1/documents` with
|
||
`Authorization: Bearer $BOT_SERVICE_TOKEN`. React to `402` (no credit,
|
||
polite refusal), `400` (bad format), `202` (acknowledged).
|
||
- Poll `GET /api/v1/reports/{id}` with backoff; surface `stage` as a status
|
||
message ("Extracting…", "Analyzing…").
|
||
- On done: send markdown report; if >4096 chars, send as `.md` attachment.
|
||
Every report carries the disclaimer (bot appends if api omitted).
|
||
|
||
No business logic, no direct state. The hexagonal rule.
|
||
|
||
---
|
||
|
||
## 18. Edge & deployment
|
||
|
||
### Nginx + certbot (`deploy/nginx/`)
|
||
|
||
- Nginx reverse-proxies `/{api,v1,healthz,readyz,metrics}` → `api:8000`.
|
||
- TLS via certbot; cert files bind-mounted; renew cron sidecar.
|
||
- Webhook target for future ЮKassa: `https://<domain>/api/v1/webhooks/yookassa`.
|
||
|
||
### Single-VPS deploy
|
||
|
||
1. VPS (Hetzner/Selectel), Docker + compose installed.
|
||
2. Clone, `cp .env.example .env`, fill secrets (DB, Rabbit, MinIO, Ollama,
|
||
Sentry, tokens).
|
||
3. Seed a `service_tokens` row for the bot (hash of `BOT_SERVICE_TOKEN`)
|
||
via a one-shot `python -m contract_check.api seed-token bot bot-prod`.
|
||
4. `docker compose --profile services --profile obs --profile edge up -d`.
|
||
5. `pg_dump` cron for backups.
|
||
|
||
### k8s-ready posture
|
||
|
||
- Every service exposes a healthcheck (`/healthz`) and is configurable
|
||
purely via env (12-factor). No code change needed to move to k8s; only
|
||
manifests/Helm charts (future).
|
||
|
||
---
|
||
|
||
## 19. Conventions & rules of the road
|
||
|
||
- **Lint/types/tests green per step**: `ruff check . && mypy src && pytest`.
|
||
- **No comments** in code (house style — existing prototype has none).
|
||
- **Hexagonal boundary**: adapters never import core state/infra. Enforced
|
||
by a unit test.
|
||
- **Reserve-on-enqueue** invariant is sacred: credit moves on `POST
|
||
/documents`, never in the worker.
|
||
- **Refund idempotency** is sacred: `documents.refunded` guards it.
|
||
- **Config via env**, never hardcoded model/prompt/delay.
|
||
- **Every report carries the disclaimer** "не заменяет юриста" — render in
|
||
`analyzer.render_markdown`, assert in tests.
|
||
- **Migrations hand-written**, `up` and `down` both clean.
|
||
- **One image = one entrypoint**; `MODE` dispatcher is dead.
|
||
- **Publisher confirms** on; a published-but-unconfirmed message fails the
|
||
HTTP request.
|
||
- **Correlation ID** flows HTTP → message header → logs → traces.
|
||
|
||
---
|
||
|
||
## 20. Landing sequence (incremental, green per step)
|
||
|
||
Each step is a verifiable unit. Do not start step N+1 until N is green
|
||
(`ruff && mypy && pytest` + relevant integration).
|
||
|
||
### Step 1 — Infra + core skeleton + initial migration
|
||
|
||
- Compose: postgres (replication-ready), redis, rabbitmq (mgmt UI exposed),
|
||
minio, minio-init. All with healthchecks + volumes. Default profile =
|
||
infra only.
|
||
- `src/contract_check/core/`: config, logging, telemetry, sentry, metrics,
|
||
db/models (all 6 tables), db/session, mq/topology, mq/messages, s3/port,
|
||
s3/minio_storage, llm/port, analysis/* (migrate extractor/chunker/checklist/
|
||
report_schema from prototype), credits, tokens.
|
||
- Alembic: init + initial migration (6 tables).
|
||
- Migrate `prototype.py` → `prototype/__main__.py` (update imports to
|
||
`core.analysis.*`).
|
||
- **DoD**: `docker compose up` infra healthy; `alembic upgrade head` +
|
||
`downgrade base` clean; `ruff && mypy && pytest` green (unit tests ported
|
||
for chunker/extractor/credits/messages).
|
||
|
||
### Step 2 — api image (done)
|
||
|
||
- `srv/api/Dockerfile`, `src/contract_check/api/*`: app factory, middleware
|
||
(correlation_id, request metrics, Sentry), routes (health, ready, metrics,
|
||
documents upload→MinIO→publish with confirms, reports poll with stage, me,
|
||
**b2b.py** — X-API-Key endpoints), deps (db session, s3, publisher,
|
||
service-token, api-key, rate-limit), services.py (shared upload logic).
|
||
- Reserve-on-enqueue; `402` path.
|
||
- Seed-token one-shot CLI.
|
||
- **DoD**: `POST /api/v1/documents` reserves credit, writes MinIO, publishes
|
||
`DocumentUploaded` (verified on extract.q); unit + integration
|
||
(`test_upload_pipeline`, `test_b2b_api`) green.
|
||
|
||
### Step 3 — worker-extract image (done)
|
||
|
||
- `srv/worker-extract/Dockerfile` (tesseract layer), `worker_extract/*`:
|
||
handler + consumer wiring + metrics server on :9101.
|
||
- `core/analysis/ocr.py`.
|
||
- Retry + DLQ via base consumer; failure classification; status updates.
|
||
- **DoD**: `test_extract_worker` + `test_retry_and_dlq` green; real PDF in
|
||
MinIO → extracted text published on analyze.q.
|
||
|
||
### Step 4 — worker-analyze image (done)
|
||
|
||
- `srv/worker-analyze/Dockerfile`, `worker_analyze/*`: handler + consumer +
|
||
metrics on :9102.
|
||
- `core/llm/ollama_cloud.py` (ported) + factory; analyzer orchestration;
|
||
report save with JSONB+markdown; refund-on-fail per policy.
|
||
- **DoD**: `test_analyze_worker` green (respx-mocked Ollama); end-to-end
|
||
upload→report in integration; disclaimer present.
|
||
|
||
### Step 5 — bot image (done)
|
||
|
||
- `srv/bot/Dockerfile`, `bot/*`: aiogram handlers, HTTP-only.
|
||
- Boundary test (no forbidden imports).
|
||
- **DoD**: local polling: send PDF → get report; boundary test green.
|
||
|
||
### Step 6 (later) — payments, web, dashboards
|
||
|
||
- ЮKassa + `invoices` logic + recurring (separate effort).
|
||
- React SPA + Telegram Login + SSE/webhook delivery.
|
||
- Grafana dashboards polished; Tempo/Jaeger trace UI.
|
||
|
||
---
|
||
|
||
## 21. Open questions / deferred
|
||
|
||
- **Payments (ЮKassa)** — next focused iteration after core lands.
|
||
- **Web SPA + Telegram Login auth** — stage 2.
|
||
- **B2B API keys + rate limit** — stage 3 (done; Redis + `core/api_keys.py` +
|
||
`core/rate_limit.py` + `api/routes/b2b.py`).
|
||
- **Multi-host HA** — when load justifies; path documented in §10.
|
||
- **Self-hosted Ollama / GigaChat** — behind the provider port when 152-ФЗ
|
||
forces it or Ollama Cloud overage bites.
|
||
- **OCR backend** (Yandex Vision) — behind an `OCRBackend` port later.
|
||
- **DLQ admin UI** — a small management route/script for requeue; mgmt UI
|
||
suffices initially.
|
||
|
||
---
|
||
|
||
## 22. Quick reference — file-to-rule index
|
||
|
||
- "What's the queue topology?" → §5, `core/mq/topology.py`
|
||
- "How does a credit move?" → §8, `core/credits.py`
|
||
- "What statuses can a document have?" → §7, `core/db/enums.py`
|
||
- "How is a job retried?" → §5 (retry mechanics), `core/mq/consumer.py`
|
||
- "How do I add a new LLM provider?" → §9, `core/llm/factory.py`
|
||
- "What can the bot import?" → §4 (boundary), §17, `tests/unit/test_bot_boundary.py`
|
||
- "Where do env vars live?" → §11, `core/config.py`, `.env.example`
|
||
- "How is this deployed?" → §18
|
||
- "How do we scale to HA?" → §10
|
||
- "What's the next thing to build?" → §20, step 1
|
||
- **"How does the B2B API work?"** → §15, `api/routes/b2b.py`, `core/api_keys.py`
|
||
- **"How is rate limiting enforced?"** → §15, `core/rate_limit.py`
|
||
- **"Where are the Dockerfiles?"** → §12, `srv/<service>/Dockerfile`
|