admin-git/DealDocumentScreening
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
DealDocumentScreening/docs/ARCHITECTURE.md
febux 85db70887d Fix Vector config and OpenObserve auth for live E2E verification 
Ticket 10 verification surfaced runtime issues that static validation
missed; all verified against running Docker stack (Vector 0.43 + OO 0.92.2):

- Rewrite VRL for Vector 0.43: no coalesce/default funcs, merge!/replace!
  for guarded fallible calls, no two-value parse_json destructuring
- Quote env-interpolated sink credentials (empty env left a bare YAML null)
- Fix logs sink URI: OO needs /api/{org}/{stream}/_json (stream segment
  was missing, causing 404); logs now land in contract_check stream
- Replace OPENOBSERVE_AUTH_TOKEN with root email/password Basic auth:
  prometheus_remote_write ignores request.headers, so remote-write got 401
- Update .env.example, DEPLOY.md, ARCHITECTURE.md, ticket 05 accordingly

Verified live: logs with service + correlation_id searchable in
OpenObserve; contract_check_* metrics queryable via its Prometheus API;
no outbound port 4318 connections from app containers.
2026-09-06 19:41:04 +03:00

1793 lines
89 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 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 + OpenObserve/Vector**, 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; the lightweight `observer` profile uses Vector+OpenObserve.
OpenTelemetry push has been removed in favour of passive collection.
- **Payments (ЮKassa) live** behind a provider port (`core/billing/port.py`),
webhook-driven state machine, plans/subscriptions + credit top-ups, and a
`worker-billing` scheduler (renewals/expiry/reconciliation). Disabled by
default (`YOOKASSA_ENABLED=false` → 503, catalog stays readable).
- **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,
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 + worker-billing + worker-notify + bot | (web 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 + SSE (`GET /reports/{id}/events`, web-контракт `analysisResultSchema`); 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, Vector/OpenObserve | 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 | Removed (stage-0 standalone benchmark no longer needed) | Kept as standalone benchmark |
| Tests | pytest+respx unit + testcontainers integration | — |
| Python | **3.14** | py3.13 |
| 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)
├── 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/prometheus.yml (scrape api + workers :9100..:9105)
│ ├── vector-config.yaml (OpenObserve profile: Vector log/metric collection)
│ ├── loki-config.yaml (Grafana/Loki profile)
│ ├── tempo.yaml (planned — trace storage for Grafana stack)
│ └── 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 (no-op stubs; OTLP push removed)
│ │ ├── 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, PasskeyCredential, Document, Report, PrescreenResult, Job, ServiceToken, ApiKey, ApiKeyRequest, Invoice)
│ │ │ ├── session.py (async_sessionmaker, engine)
│ │ │ └── enums.py (DocStatus, JobStatus, QueueName, RoutingDecision, 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 LLM adapter + ocr adapter)
│ │ │ ├── __init__.py
│ │ │ ├── extractor.py (← from contract_check/extractor.py; worker_extract 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; /events — SSE)
│ │ │ ├── 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, update-mode/webhook settings)
│ ├── client.py (ApiClient: typed httpx wrapper; ApiError/NoCreditsError/...)
│ └── handlers.py (/start, doc upload→POST api, poll→send report)
└── 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` (outbound to the api), `aiohttp` (inbound
webhook-mode transport, aiogram's HTTP engine) + 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.)
---
## 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 & data access (Alembic migrations + repositories)
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.
### 7.1 Data-access layer
All production SQL lives behind the repository package:
```
src/contract_check/core/db/repositories/
├── api_keys.py (ApiKeyRepository)
├── credits.py (CreditsRepository — thin facade)
├── documents.py (DocumentRepository)
├── jobs.py (JobRepository)
├── passkeys.py (PasskeyRepository)
├── prescreen_results.py (PrescreenResultRepository)
├── reports.py (ReportRepository)
├── service_tokens.py (ServiceTokenRepository)
└── users.py (UserRepository)
```
Rules:
- A repository receives an `AsyncSession`; it **never** commits.
- Callers own transactions and commits.
- Raw `text()` is allowed only inside repositories or migrations.
- `api/deps.py` keeps lightweight dependency helpers but no direct SQL; it calls
`UserRepository`, `ApiKeyRepository`, etc.
- `core/credits.py` remains the public facade for credit operations; it now
delegates to `CreditsRepository`.
### 7.2 Migrations
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 ix_prescreen_results_document_id 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 (billing, migration 0011; money = kopecks int)
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')),
kind VARCHAR(12) CHECK (kind IN ('topup','subscription','renewal')),
credits_purchased INT, -- topup: pack size
subscription_id UUID, -- subscription/renewal origin
provider TEXT, -- 'yookassa'
external_id TEXT, -- ЮKassa payment id
confirmation_url TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
paid_at TIMESTAMPTZ
);
CREATE INDEX invoices_user_idx ON invoices (user_id, created_at DESC);
-- user_profiles (1:1 passive settings, migration 0011)
CREATE TABLE user_profiles (
user_id UUID PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
language VARCHAR(4), -- 'ru' | 'be' | 'en' (stored passively)
timezone VARCHAR(64), -- IANA name
notif_prefs JSONB NOT NULL DEFAULT '{"report_ready": true, "security": true, "marketing": false}',
dashboard_prefs JSONB NOT NULL DEFAULT '{"severity_filter": "all", "per_page": 10, "density": "comfortable"}',
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- plans (lookup, seeded)
CREATE TABLE plans (
code VARCHAR(16) PRIMARY KEY, -- free | lite | pro | max
name TEXT NOT NULL,
price_kopecks INT NOT NULL, -- 0 / 49000 / 149000 / 390000
monthly_quota INT NOT NULL, -- 0 / 5 / 20 / 60
is_active BOOLEAN NOT NULL DEFAULT TRUE,
sort INT NOT NULL
);
-- subscriptions (one active per user, partial unique index)
CREATE TABLE subscriptions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
user_id UUID NOT NULL REFERENCES users(id),
plan_code VARCHAR(16) NOT NULL REFERENCES plans(code),
status VARCHAR(12) NOT NULL
CHECK (status IN ('active','past_due','expired','cancelled')),
current_period_start TIMESTAMPTZ NOT NULL,
current_period_end TIMESTAMPTZ NOT NULL,
auto_renew BOOLEAN NOT NULL DEFAULT FALSE, -- opt-in, default OFF
origin_invoice_id UUID REFERENCES invoices(id),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX uq_subscriptions_active
ON subscriptions (user_id) WHERE status IN ('active','past_due');
-- quota_usage (per-period plan usage ledger; UNIQUE(document_id) = idempotency)
CREATE TABLE quota_usage (
id BIGSERIAL PRIMARY KEY,
user_id UUID NOT NULL,
subscription_id UUID NOT NULL REFERENCES subscriptions(id),
document_id UUID NOT NULL REFERENCES documents(id),
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (document_id)
);
-- credit_events (append-only credit ledger; balance_after self-verifying)
CREATE TABLE credit_events (
id BIGSERIAL PRIMARY KEY,
user_id UUID NOT NULL,
delta INT NOT NULL CHECK (delta <> 0),
kind VARCHAR(16) NOT NULL CHECK (kind IN
('signup','admin_grant','topup','reserve','refund_auto',
'clawback','quota_refund','hold_clear')),
document_id UUID,
invoice_id UUID,
balance_after INT NOT NULL CHECK (balance_after >= 0),
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- users: + billing_hold BOOLEAN NOT NULL DEFAULT FALSE (refund-clawback gate,
-- uploads answer 402 "billing hold" until an admin clears it)
```
**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`.
---
## 8a. Billing: plans, payments, refunds (ЮKassa, migration 0011)
Money is **integer kopecks everywhere**; RUB decimals exist only at the ЮKassa
HTTP boundary. Domain code lives in `core/billing/` (port + yookassa adapter +
quota + fulfillment + refunds), mirroring `core/llm/`.
### Payment provider port
`core/billing/port.py` defines `PaymentProvider`: `create_payment`,
`get_payment`, `refund` — all idempotence-keyed (key = invoice uuid). The
ЮKassa adapter is httpx-based and respx-testable. `YOOKASSA_ENABLED=false`
⇒ `build_payment_provider` raises `PaymentsDisabled` ⇒ mutating billing
routes answer 503 while `GET /billing/plans` stays readable (dev degrade).
### Invoice state machine (webhook-driven, idempotent)
```
pending ──payment.succeeded──► succeeded (topup: +credits ledger event;
│ subscription: grant active period)
├──payment.canceled───► cancelled
└──reconcile (pending >15m)──► re-fetch via API, apply same edges
succeeded ──refund──► refunded (clawback + billing_hold, §8b)
```
Webhook trust model (`api/routes/webhooks.py`): ЮKassa notification Basic auth
(shop id + secret, constant-time compare), then **re-fetch the payment via the
REST API** — the push payload's amount is never trusted. Applying is delegated
to `core/billing/fulfillment.apply_payment_status`, the same function the
worker-billing reconciliation sweep uses; replays are no-ops.
### Subscription lifecycle (worker-billing, 60 s tick)
- `auto_renew = TRUE` and `period_end − 3d < now` and no pending renewal
invoice → create renewal invoice + payment.
- `period_end < now`: renewal `succeeded` → roll period forward (`active`);
else → `past_due` (7-day grace) → `expired`.
- `auto_renew = FALSE` (default): expire at period end. Monthly quota resets
on period roll — **no rollover**.
### Upload quota ordering (D7)
```
billing_hold = TRUE? → 402 "billing hold" (hard stop, admin clears)
active subscription & quota rows this period < plan.monthly_quota
→ insert quota_usage row (source='quota')
else credits_left > 0 → atomic credit reserve (source='credits')
else → 402 "no credits available"
```
Implemented in `core/billing/quota.reserve_document_slot`; gated behind
`PLANS_ENABLED` (false = legacy credits-only).
### Refund policy (D10) — `core/billing/refunds.py`
Usage = credit `reserve` events + `quota_usage` rows created after the
invoice's `paid_at` (FIFO approximation).
- within `REFUND_WINDOW_DAYS` (14) **and** usage ≤
`REFUND_FULL_USAGE_THRESHOLD` (20%) of purchased ⇒ **full** refund;
- else ⇒ **proportional**: `max(0, amount − used_docs × PRICE_PER_DOC_KOPECKS)`;
- already refunded / not succeeded / outside window / zero result ⇒ 409.
Execution: provider `refund` → invoice `refunded` → **clawback** (remove
remaining purchased credits / cancel subscription + drop current-period
quota rows, `clawback` ledger events) → if the balance went negative or a
subscription was pulled back, set `users.billing_hold = TRUE` → uploads 402
until an admin clears it (audit trail: log lines, not ledger — `delta <> 0`
CHECK forbids zero-delta events).
### Admin panel
`/admin/invoices` (global list, status/kind filters) + user-card billing
blocks: invoices, credit-events tail, hold badge + "снять hold", manual
refund (same service as the API route) with a full-refund override for
support cases. ЮKassa credentials are env-only — never entered in the panel.
---
## 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`. 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` |
| `PLANS_ENABLED` | `false` | subscription quota logic; false = credits-only legacy |
| `YOOKASSA_ENABLED` | `false` | payments master switch; false → billing mutations 503 |
| `YOOKASSA_SHOP_ID` / `YOOKASSA_SECRET_KEY` | (empty) | ЮKassa merchant credentials |
| `YOOKASSA_RETURN_BASE_URL` | (empty) | `/pay/{invoice_id}` base for return links |
| `PRICE_PER_DOC_KOPECKS` | `19900` | pay-per-doc price (199 ₽) |
| `BILLING_RETURN_JWT_SECRET` | (empty → `JWT_SECRET`) | signs `/pay/{id}` tokens |
| `BILLING_RETURN_TOKEN_TTL_MINUTES` | `15` | pay-page token TTL |
| `REFUND_WINDOW_DAYS` | `14` | full-refund window (§8a) |
| `REFUND_FULL_USAGE_THRESHOLD` | `0.20` | usage ratio for full refund |
| `SENTRY_DSN` | (empty) | if set, init sentry |
| `OPENOBSERVE_ROOT_USER_EMAIL` | `root@example.com` | OpenObserve root user; Vector uses it for Basic auth |
| `OPENOBSERVE_ROOT_USER_PASSWORD` | `Complexpass#123` | OpenObserve root password; Vector uses it for Basic auth |
| `OTEL_EXPORTER_OTLP_ENDPOINT` | (removed) | no longer used; the application does not push OTLP |
| `OTEL_EXPORTER_OTLP_HEADERS` | (removed) | no longer used |
| `OTEL_SERVICE_NAME` | (removed) | service name is set by `configure_logging()` and compose labels |
| `GRAFANA_PORT` | `3000` | host port for the Grafana UI (profile `obs`) |
| `GRAFANA_ADMIN_USER` | `admin` | initial Grafana admin user |
| `GRAFANA_ADMIN_PASSWORD` | `admin` | initial Grafana admin password |
| `GRAFANA_ROOT_URL` | `http://localhost:3000` | external Grafana URL for callbacks |
### 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_BIND_HOST` | `127.0.0.1` | host interface the API /metrics port is bound to (compose only) |
| `API_METRICS_PORT` | `9100` | host port the API /metrics endpoint is published on (compose only; default loopback) |
| `B2B_DEFAULT_RATE_LIMIT_RPS` | `3` | per API key; mirrors Ollama Pro concurrency, overridable per `api_keys.rate_limit_rps` |
| `METRICS_BEARER_TOKEN` | (empty) | Bearer token protecting `/metrics`; empty leaves the endpoint open, so set a secret in production |
| `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 |
| `AUTH_RATE_LIMIT_IP_RPS` | `5` | per-IP token-bucket rate for public auth endpoints (register, login, forgot/reset password, magic-link) |
| `AUTH_RATE_LIMIT_EMAIL_RPS` | `2` | per-recipient token-bucket rate for mail-sending auth endpoints (forgot-password, magic-link request) |
| `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`) |
| `BOT_UPDATE_MODE` | `polling` — `webhook` runs the aiohttp server from `bot/webhook.py`: secret-path updates (403 on `X-Telegram-Bot-Api-Secret-Token` mismatch) + `GET /healthz`; registers/deregisters the Telegram webhook on start/stop |
| `BOT_WEBHOOK_PUBLIC_BASE_URL` | (empty; required in webhook mode) public https base URL of the TLS edge, no trailing slash |
| `BOT_WEBHOOK_SECRET_TOKEN` | (empty; required in webhook mode) `A-Za-z0-9_-`, max 256; keys both the header check and the HMAC-derived path |
| `BOT_WEBHOOK_LISTEN_HOST` / `BOT_WEBHOOK_LISTEN_PORT` | `0.0.0.0` / `8080` — webhook server bind (inside the container) |
### 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.14)
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.14-trixie-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.14-slim-trixie 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, libmagic1t64 | `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` | — |
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 + libmagic1t64). 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.14"
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"]
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"]
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" }]
prescreen = [{ include-group = "db" }, { include-group = "mq" },
{ include-group = "s3" }, { include-group = "obs" }]
bot = ["aiogram>=3.4"]
dev = [{ include-group = "api" }, { include-group = "extract" },
{ include-group = "prescreen" }, { include-group = "analyze" },
{ include-group = "bot" },
"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` profile = api + workers (no bot).
- `bot` profile = Telegram bot adapter; can run on the same host or a separate
server (`docs/DEPLOY.md` §14).
- `obs` (Grafana + Loki + Prometheus) and `observer` (OpenObserve + Vector)
are mutually exclusive observability stacks.
- `edge` = Nginx + certbot.
```
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
worker-billing: build srv/worker-billing/Dockerfile; depends_on pg/minio-init healthy; 9105
worker-notify: build srv/worker-notify/Dockerfile; depends_on rabbit healthy; 9103
# ── BOT (profile: bot) ──
bot: build srv/bot/Dockerfile; no depends_on so it can run on a
separate host; healthchecks API at runtime
# ── OBSERVABILITY — GRAFANA STACK (profile: obs) ──
loki: grafana/loki; filesystem-backed single-node log store
promtail: grafana/promtail; ships Docker container logs into Loki
grafana: grafana/grafana; provisioned Loki datasource + logs dashboard
prometheus: prom/prometheus; scrapes service /metrics endpoints
# ── OBSERVABILITY — OPENOBSERVE STACK (profile: observer) ──
openobserve: public.ecr.aws/zinclabs/openobserve:latest; UI + JSON/Prometheus-write backend on :5080
vector: timberio/vector:0.43.0-alpine; reads Docker logs + scrapes /metrics,
forwards logs/metrics to OpenObserve
# ── EDGE (profile: edge) ──
nginx: nginx:alpine; reverse proxy → api/openobserve/grafana; TLS via certbot
certbot: certbot/certbot; renew cron sidecar
volumes: { pgdata, pgwal, redisdata, rabbitmq, minio, loki-data, grafana-data, prometheus-data, openobserve-data }
```
`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.
- Distributed tracing is intentionally out of scope; `correlation_id` in
structlog already provides request-level correlation across services.
- The same `correlation_id` is parsed by the Loki datasource as a derived field,
so you can click through from any log line to every other log line with the
same id.
### 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 + Loki
- `obs` profile adds `loki`, `promtail`, `grafana` to compose.
- Promtail discovers all compose containers via the Docker socket and pushes
their stdout/stderr to Loki; labels include `service`, `project`, `profile`,
and `container`.
- Logs retain the JSON format emitted by structlog in `staging`/`prod` and the
pretty console format in `dev`; Loki stores the raw line.
- Provisioned datasource `Loki` and a starter `Contract Check — Logs`
dashboard at `http://localhost:${GRAFANA_PORT:-3000}/d/contract-check-logs`.
- Dashboard has a `service` filter and a free-text search box; error words are
highlighted as annotations. Use correlation_id values to trace one upload
across `api` → `worker-extract` → `worker-prescreen` → `worker-analyze`.
### Prometheus + Tempo (Grafana stack, planned enhancements)
- `prometheus` in the `obs` profile scrapes `:9100`/`:9101`/`:9102`/`:9104`/`:9105`/`:9103`
metrics exposed by the services.
- Tracing may be added later as a separate feature evaluated against the
passive-collection principle.
- Starter dashboards for queue depth, job latency, LLM tokens/min, refund rate,
and 429/fallback rate will ship with the metrics datasources.
### OpenObserve + Vector (`observer` profile)
- `vector` runs in the `observer` profile. It reads container logs from the
local Docker socket and scrapes Prometheus `/metrics` endpoints inside the
compose network (`api:8000/metrics`, workers on `:9101`/`:9102`/`:9104`/`:9105`/`:9103`).
- Vector enriches logs with `service` (container name) and preserves
`correlation_id` from JSON log lines.
- Logs are sent to
`http://openobserve:5080/openobserve/api/default/contract_check/_json` and
metrics to `http://openobserve:5080/openobserve/api/default/prometheus/api/v1/write`,
both authenticated with the OpenObserve root credentials
(`OPENOBSERVE_ROOT_USER_EMAIL` / `OPENOBSERVE_ROOT_USER_PASSWORD`).
- OpenObserve UI is exposed under `/openobserve/` via nginx.
- Worker metrics ports are no longer published on the Docker host; they are only
reachable inside the compose network for scraping.
- The API metrics port is bound to `127.0.0.1` by default to avoid accidental
public exposure.
### Telemetry module (`core/telemetry.py`)
- `setup_telemetry()` / `shutdown_telemetry()` are safe no-op stubs. The
application does not import or initialize the OpenTelemetry SDK.
- `configure_logging` routes structlog through a root `StreamHandler`, so logs
reach stdout/stderr and are collected passively by Vector or Promtail.
---
## 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` — `core.analysis.extractor` text extraction path.
- `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/`) — `telegram.py`, `password.py`, `passkeys.py`, `magic_links.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/reports/{document_id}/events` | user JWT (header или `?access_token=` — для нативного EventSource) | SSE-стрим статусов: безымянные кадры с payload `analysisResultSchema` (camelCase, `pending/processing/completed/failed`); терминальные `completed/failed` закрывают стрим; служебные события `timeout`/`error`; keep-alive комментарии между кадрами |
| 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…"). Web clients can instead subscribe to the SSE stream
`GET /api/v1/reports/{id}/events`: every frame carries the normalized
`analysisResultSchema` payload (`pending/processing/completed/failed`,
issues, derived riskScore/summary) — see `api/schemas/analysis_result.py`.
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).
Update delivery is config-driven (`BOT_UPDATE_MODE`, §11): long-polling by
default; in webhook mode `bot/webhook.py` runs a small aiohttp server —
secret-derived path (`/tg-webhook/<HMAC-SHA256(secret, key=token)[:32]>`),
403 on `X-Telegram-Bot-Api-Secret-Token` mismatch before the dispatcher,
`GET /healthz`, `set_webhook(drop_pending_updates)` on startup and
`delete_webhook` on graceful shutdown. Transport only; the api boundary is
unchanged (docs/DEPLOY.md §14.4).
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,pay,healthz,readyz,metrics}` → `api:8000`.
- TLS via certbot; cert files bind-mounted; renew cron sidecar.
- ЮKassa webhook target: `https://<domain>/api/v1/webhooks/yookassa`
(configure in the ЮKassa merchant dashboard; Basic auth = shop id + secret).
### 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.
---
## 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 10 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.*`).
- *Later removed:* stage-0 `prototype/` standalone CLI deleted as no longer needed.
- **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.
---
## 21. Open questions / deferred
- **Payments (ЮKassa)** — **done** (plans/subscriptions/top-ups/refunds,
`core/billing/`, worker-billing; see §8a and `.scratch/user-profile-billing/`).
- **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`
Reference in a new issue View git blame Copy permalink