# ARCHITECTURE (AS BUILT) — «Контракт-чек»

> This document describes the system **as it exists in the code today**, verified
> against `src/`, `migrations/`, `tests/`, and compose files. The design-time
> rationale, decision history and future steps remain in
> [`docs/ARCHITECTURE.md`](ARCHITECTURE.md) (design doc); this file is the
> as-built map. Domain language is defined in the root [`CONTEXT.md`](../CONTEXT.md).

---

## 1. What the system is

LLM-powered screening of contract (deal) documents under Russian / Belarusian
civil law. A user (Telegram bot, web JWT auth, or B2B API key) uploads a
document; an event-driven pipeline extracts text, prescreens it, optionally
runs deep LLM analysis, and produces a markdown report with findings, quotes
and clause references. Document consumption is metered: subscription Quota
first, then prepaid Credits, with ЮKassa payments (topups, subscriptions,
renewals, refunds with clawback).

**Stack:** Python 3.14, uv + hatchling, FastAPI, SQLAlchemy 2 (async) +
Alembic, aio-pika (RabbitMQ), MinIO (S3), Redis, aiogram 3, httpx,
Prometheus + structlog + Sentry, OpenObserve/Vector (or Grafana/Loki) for
observability, Nginx + certbot at the edge. Docker Compose deployment
(one Dockerfile per service in `srv/`), k8s-ready but not k8s-running.

---

## 2. Hexagonal shape and import rules

```
src/contract_check/
├── core/            # domain — owns ALL state and side effects
│   ├── db/          # models, session, repositories/
│   ├── mq/          # topology, publisher, consumer, messages, management
│   ├── s3/          # Storage port + MinIO adapter
│   ├── llm/         # LLMProvider port + ollama_cloud / yandex_gpt adapters
│   ├── billing/     # PaymentProvider port + yookassa, quota, fulfillment, refunds
│   ├── extraction/  # DocumentExtractor port + format adapters
│   ├── analysis/    # chunker, checklist, analyzer, report_schema
│   ├── notifications/ # SMTP transport + notify publisher
│   ├── security/    # argon2 passwords
│   └── config.py logging.py metrics.py telemetry.py sentry.py errors.py
│     credits.py tokens.py api_keys.py rate_limit.py auth*.py passkeys.py
│     redis_client.py review/
├── api/             # FastAPI image (routes/, admin/, billing/ pay pages, schemas/)
├── worker_extract/  # CPU image: pymupdf + mammoth + striprtf + chardet + tesseract
├── worker_prescreen/# hybrid metadata extraction + routing
├── worker_analyze/  # I/O image: LLM provider
├── worker_notify/   # email (aiosmtplib)
├── worker_billing/  # timer scheduler (no MQ)
└── bot/             # aiogram adapter — HTTP-only
```

**Boundary rules (statically enforced for the bot by
`tests/unit/test_bot_boundary.py`):**

- `core/*` may import anything.
- `api/*`, `worker_*/*` import only `core/*` (+ their own modules).
- `bot/*` imports only `httpx`/`aiohttp`/aiogram and its own modules — never
  `core.db`, `core.s3`, `core.llm`, `core.mq`, `core.credits`. It talks to the
  api over HTTP exclusively, and may run on a separate host
  (`docker-compose.bot.yml`).

**Control plane:** `api` is the only writer for user-initiated mutations
(upload → reserve Document Slot → MinIO → publish). Workers write their own
stage rows / reports; credits move only through the idempotent compensation
helpers (`core/billing/quota.compensate_document_slot`, `core/credits.refund_credit`).

---

## 3. Services and the pipeline

```
Telegram ──► bot (aiogram, HTTP-only) ──HTTP──► api (FastAPI) ──publish──► RabbitMQ
                                                  owns: PG, MinIO, Redis,       │
                                                  quota/credits, tokens          ▼
                                                                        extract.q ──► worker-extract
                                                                       (CPU: pymupdf/mammoth/tesseract)
                                                                                    │ publish
                                                                                    ▼
                                                                       prescreen.q ──► worker-prescreen
                                                                     (hybrid heuristic + optional LLM)
                                                                       │ deep_analysis        │ manual_review (terminal)
                                                                       ▼                      ▼
                                                                       analyze.q ──► worker-analyze ──► Postgres (Report, done)
notify.x ──► notify.q ──► worker-notify (SMTP: reset, magic link, …)
worker-billing (60 s Postgres-advisory-lock tick; no MQ)
```

### api (`api/`, `srv/api/`)

- `create_app` (`api/app.py:115`): logging/Sentry, MinIO `ensure_bucket`,
  MQ publisher (publisher confirms), Redis rate-limiter with in-memory
  fallback, admin seed. Middleware: CORS, correlation-id propagation,
  `http_request_duration`, access logs, redacted debug payloads, generic 500.
- Routes under `/api/v1`: `auth/` (Telegram bot/web/miniapp, email+password
  JWT pair, passkeys, magic links, support), `documents` (upload → 202),
  `reports` (polling + SSE `/events`), `me` (profile, overview, documents,
  telegram bind, password), `billing` (plans, checkout, invoices,
  subscription, autorenew, refund), `b2b` (API-key analyze/status/usage/
  profile/keys CRUD), `webhooks/yookassa`. Plus `/healthz`, `/readyz`,
  `/metrics` (bearer-gated), server-rendered `/pay/{invoice_id}` (short-lived
  JWT pay token), and `/admin/*` (Jinja2 + HTMX panel: users, credits,
  invoices/refunds, hold clearing, gift subscriptions, manual-review queue,
  MQ queues/DLQ requeue).
- **Upload flow** (`api/services.py:56` `upload_and_enqueue`): validate
  suffix/format → reject on `billing_hold` (402) → stream body with size cap
  (413) → MinIO put (DB failure triggers best-effort S3 cleanup) → insert
  `documents` (queued) + `jobs` (extract/pending) → reserve Document Slot
  (Quota-then-Credit when `PLANS_ENABLED`, else legacy `reserve_credit`; 402
  on `NoCredits`) → publish `DocumentUploaded` (publisher confirms; on MQ
  failure: release slot / refund credit, status `publish_failed`) →
  `202 {document_id, correlation_id, credits_left}`.

### worker-extract (`worker_extract/`, metrics :9101, prefetch 1)

Consumes `extract.q` / `DocumentUploaded`. Idempotency check against terminal
statuses → status `extracting` → download blob → `extract_document`
(`core/extraction/factory.py`: python-magic sniff → uploader MIME → suffix;
adapters: pdf/pymupdf, docx/mammoth, rtf/striprtf, txt+csv/chardet,
images/tesseract OCR; unsupported formats are terminal + refunded) → upload
extracted markdown to MinIO → publish `PrescreenRequested` (or
`DocumentExtracted` straight to analyze.q when `PRESCREEN_ENABLED=false`).
Failure classes: `extraction_failed` (non-refundable under `infra_only`),
`ocr_failed`, `infra`. Terminal failure → DLQ + slot compensation.

### worker-prescreen (`worker_prescreen/`, metrics :9104, prefetch 1)

Consumes `prescreen.q` / `PrescreenRequested`. Handler
(`worker_prescreen/handler.py:116`): transition `prescreening` → download
text → **hybrid extractor** (`extractor_hybrid.py`): Stage 1 deterministic
heuristic (`heuristic-v2`, regex-free, keyword/positional); Stage 2 LLM
fallback only if enabled AND confidence < `PRESCREEN_LLM_FALLBACK_THRESHOLD`
(LLM failure never fatal; LLM fills heuristic `None`s, booleans OR-merged,
confidence rescored) → **router** (`router.py:30`): missing
type/parties/low-confidence → `manual_review` (terminal status, admin queue);
amount ≥ `PRESCREEN_HIGH_VALUE_THRESHOLD` or penalty/arbitration clause →
`deep_analysis` (publish `AnalyzeRequested` with `prescreen_meta`);
`auto_approve` disabled by default (remapped to manual_review; when enabled
writes a lightweight Report, `status=done`). Persists `prescreen_results`.

### worker-analyze (`worker_analyze/`, metrics :9102, prefetch 3)

Consumes `analyze.q` / `AnalyzeRequested`. Status `analyzing/llm` → download
text → `provider.analyze(text, checklist)` (chunking, fan-out under
`asyncio.Semaphore(LLM_MAX_CONCURRENCY)`, json-schema + repair loop, 429 →
fallback model — all inside the LLM adapter) → `render_markdown` (findings,
quotes, clause refs, «не заменяет юриста» disclaimer) → `reports` upsert,
`status=done`. Failure classes: `llm_quota`, `llm_timeout`,
`llm_invalid_output`, `infra`; DLQ → slot compensation.

### worker-notify (`worker_notify/`, metrics :9103, prefetch 5)

Consumes `notify.q` / `NotificationMessage` (kinds: password_reset, welcome,
email_verification, magic_link). SMTP via aiosmtplib; dev logger when
`SMTP_HOST` empty.

### worker-billing (`worker_billing/`, metrics :9105, no MQ)

60-second tick guarded by a **Postgres advisory lock** (single replica):
renewal invoices for auto-renew subscriptions expiring < 3 days; period roll /
`past_due` (grace) / `expired` transitions; reconciliation of pending
invoices older than 15 minutes via the same `apply_payment_status` used by
the webhook. No-op safe when ЮKassa disabled.

### bot (`bot/`, `srv/bot/`)

aiogram 3, polling or webhook mode (aiohttp, secret-token auth). Commands:
`/start /help /profile /balance /plans /reports /status`; checkout callbacks
(`plan:*`, `topup:*`); document upload → API → poll report with stage labels;
per-user rate limit (Redis, memory fallback); service-token login cached per
`telegram_id`; `ensure_disclaimer` guarantees the legal disclaimer.

---

## 4. RabbitMQ topology (declared idempotently in `core/mq/topology.py`)

| Object | Type / args | Purpose |
|---|---|---|
| `contracts.x` | direct | main exchange; RK `extract` / `prescreen` / `analyze` |
| `extract.q` / `prescreen.q` / `analyze.q` | **quorum**, DLX → `contracts.retry.x` | main work queues |
| `contracts.retry.x` + `*.retry.q` | direct + classic (per-message TTL, `lazy` policy via mgmt API) | delay slots; TTL expiry dead-letters the message back to `contracts.x` |
| `extract.dlq` / `prescreen.dlq` / `analyze.dlq` | quorum | poison; manual requeue via admin panel / `core/mq/management.py` |
| `notify.x` / `notify.q` / `notify.retry.q` / `notify.dlq` | mirror of the above | notification pipeline |

**Retry mechanics** (`core/mq/consumer.py`): generic `Consumer[MsgT]` with
`handle`/`classify`/`on_failure`/`on_dlq` hooks. On failure the consumer
re-publishes to the retry exchange with `expiration = MQ_RETRY_BASE_MS ·
2^(attempt−1)` (default 2 s base → 2/4/8/16/32 s) and acks the original.
`x-attempt >= MQ_MAX_ATTEMPTS` (5) → DLQ with `x-failure-class` headers,
`jobs.dlq=true`, `documents.status=failed`, slot compensation. Pydantic
validation failure → straight to DLQ (`infra` poison). `core.errors.TerminalError`
bypasses retries. If the failure hooks themselves raise, the message is
retried (hook-failure fallback). Publisher uses **publisher confirms** —
unconfirmed publish fails the request, so a paid slot never silently vanishes.

Message schemas (`core/mq/messages.py`, pydantic v2, base `PipelineMessage`
with `next_attempt()`): `DocumentUploaded`, `DocumentExtracted`,
`PrescreenRequested`, `AnalyzeRequested` (carries `prescreen_meta`),
`PrescreenCompleted`, `NotificationMessage`. Headers: `x-correlation-id`,
`x-attempt`, `x-origin`; `delivery_mode=2`.

**Idempotency:** every handler re-reads `documents.status` first; terminal or
in-flight → ack and exit. No double-LLM, no double-refund.

---

## 5. Data model (Postgres, 16 tables, `core/db/models.py`; migrations 0001–0011)

| Table | Role |
|---|---|
| `users` | identity (telegram_id and/or email), argon2 `password_hash`, reset/magic-link token hashes, `role` (user/admin), `credits_left` (CHECK ≥ 0), `billing_hold` |
| `passkey_credentials` | WebAuthn credentials (credential_id, public_key, sign_count) |
| `documents` | one upload: s3 keys, status `queued→extracting→prescreening→ocr→analyzing→done\|failed\|manual_review`, `refunded` |
| `reports` | 1:1 with document: `content_json` JSONB + `markdown`, model, tokens, latency, prescreen link |
| `prescreen_results` | extracted metadata (type, parties, amount, dates, clause flags), `confidence_score`, `routing_decision`, `extractor_version` |
| `jobs` | per-stage correlation (UNIQUE document+queue), attempts, `last_failure_class`, `dlq` |
| `service_tokens` | per-adapter bearer auth (bot/web/cli), revocable |
| `api_keys` / `api_key_requests` | B2B keys (SHA-256 hash, per-key rps + monthly quota) and per-call usage ledger |
| `invoices` | money in **integer kopecks**; kinds `topup`/`subscription`/`renewal`; statuses draft→pending→succeeded/cancelled/refunded |
| `plans` / `subscriptions` / `quota_usage` | seeded plan catalog; one active-or-past_due subscription per user (partial unique); quota ledger with UNIQUE(document_id) idempotency |
| `credit_events` | append-only credit ledger (`delta ≠ 0`, `balance_after ≥ 0` self-verifying) |
| `user_profiles` | 1:1 passive settings (language, TZ, notif/dashboard prefs) |

**Data access:** all SQL lives in `core/db/repositories/` (14 repositories);
a repository receives an `AsyncSession` and never commits — callers own
transactions. Raw `text()` only inside repositories/migrations. Status/enum
columns are `TEXT + CHECK` (not PG enums) so migrations stay additive. Alembic
is async; CI exercises `upgrade head` → `downgrade base`.

---

## 6. Credits, Quota, refunds (billing invariants, `core/billing/`)

- **Reserve-on-enqueue:** the API synchronously reserves a Document Slot
  before publishing (`core/billing/quota.py:22`
  `reserve_document_slot`): `SELECT … FOR UPDATE` on the active subscription →
  quota rows this period < `plan.monthly_quota` → insert `quota_usage`
  (`ON CONFLICT DO NOTHING`) → source `quota`; else atomic credit reserve →
  `credits`; else 402. `PLANS_ENABLED=false` → legacy credits-only path.
- **Compensation (exactly-once):** on terminal processing failure,
  `compensate_document_slot` (`quota.py:100`) releases the quota row or
  refund the credit (guarded by `documents.refunded`), honouring
  `REFUND_POLICY=all|infra_only` with the failure-class taxonomy
  (`extraction_failed` non-refundable under `infra_only`).
- **Payments:** `PaymentProvider` port (`port.py`) + `YookassaProvider`
  (httpx, Basic auth, Idempotence-Key = invoice UUID, kopecks↔RUB only at
  the HTTP boundary). `YOOKASSA_ENABLED=false` → `PaymentsDisabled` →
  billing mutations 503, catalog readable.
- **Webhook trust** (`api/routes/webhooks.py`): Basic auth constant-time
  check, then the payment is **re-fetched via REST** — the push payload's
  amount is never trusted; fulfillment is idempotent
  (`core/billing/fulfillment.apply_payment_status`), also used by the
  reconciliation sweep.
- **Refunds (D10)** (`core/billing/refunds.py`): within `REFUND_WINDOW_DAYS`
  (14) and usage ≤ 20 % of purchased → full; else proportional
  `max(0, amount − used × PRICE_PER_DOC)`; otherwise 409. Execution:
  provider refund → invoice `refunded` → **clawback** (unspent credits /
  subscription + period quota rows) → `users.billing_hold = TRUE` when the
  balance would go negative; uploads 402 until an admin clears the hold.

---

## 7. LLM provider port (`core/llm/`)

`LLMProvider` Protocol (`port.py:31`): `analyze(text, *, checklist)` and
`extract_prescreen(text)`. Registered providers via `LLM_PROVIDER`:
`ollama_cloud` (default; `format: json-schema` chat, model
`qwen2.5:14b`, fallback `qwen2.5:7b`) and `yandex_gpt`
(`yandexgpt-lite`, `Api-Key`, JSON_OBJECT). Shared behaviour in each adapter:
chunk fan-out under a semaphore, repair loop (one REPAIR re-send on invalid
JSON, then `LLMError`), 429 → fallback model → `LLMQuotaError`, exponential
HTTP backoff for timeouts/5xx, terminal `LLMConfigError` for connect/404.
Checklist: 10 frozen items (`core/analysis/checklist.py`); `Finding` schema
(`report_schema.py`) doubles as the Ollama json-schema and tolerates alias
field names.

---

## 8. AuthN/AuthZ surface

| Caller | Mechanism |
|---|---|
| bot adapter | `Authorization: Bearer <service_token>` (only `/api/v1/auth/telegram/bot`) → user JWT exchange |
| Telegram web / Mini App | Login Widget hash / `initData` verification → JWT |
| web user | email+password → JWT pair (access 24 h + refresh 30 d, Redis-backed revocation); passkeys (WebAuthn, Redis challenges) and magic links → single access JWT |
| B2B | `X-API-Key` (SHA-256, token-bucket rate limit, monthly quota) |
| admin panel | HttpOnly cookie + `users.role = admin` |
| ЮKassa webhook | Basic (shopId:secret) |
| health/metrics | none / bearer |

Redis is used only for rate limiting, refresh-token store, passkey
challenges and bot limiter — never as a job queue.

---

## 9. Storage (MinIO, `core/s3/`)

Single bucket `contract-check-docs`; keys `users/{uid}/docs/{did}.{ext}`
(original) and `users/{uid}/docs/{did}.txt` (extracted markdown), built only
via key builders. Uploads proxy through the API (multipart, 25 MiB cap,
streamed with limit). `minio-init` one-shot creates the bucket and the ILM
expiry (`DOC_RETENTION_DAYS` 7 / `TEXT_RETENTION_DAYS` 30) — 152-ФЗ lever:
raw text leaves on schedule; reports live in Postgres and survive the purge.

---

## 10. Observability & deploy

- **Logs/metrics/traces:** structlog JSON + correlation-id contextvar
  (propagated through HTTP headers and MQ headers), Prometheus per-service
  (`/metrics` on api :9100; workers :9101–:9105 scraped in-network by
  Vector), Sentry (DSN-gated). Passive collection: `observer` profile =
  OpenObserve + Vector; `obs` profile = Prometheus + Loki + Promtail +
  Grafana. OTLP push was removed.
- **Compose** (`docker-compose.yml`): default = infra only (postgres:18 with
  `wal_level=replica` + WAL archive, redis:8, rabbitmq:4, minio + minio-init);
  profiles `services` (api + 5 workers), `bot` (standalone-able),
  `edge` (nginx + certbot + exporter), `observer`/`obs`. Host ports offset:
  PG 15432, Redis 17379, Rabbit 5672/15672, MinIO 9000/9001, api 8000,
  OpenObserve 5080, edge 80/443.
- **Images:** one Dockerfile per service in `srv/` (two-stage uv build,
  `uv sync --no-default-groups --group <grp>` → python:3.14-slim); dependency
  groups (PEP 735) per service keep images minimal; bot is leanest.
- **CI** (`.github/workflows/ci.yml`): ruff check+format, `ty` typecheck,
  unit tests with `--cov-fail-under=50`, integration tests against an
  isolated compose test stack (`docker-compose.test.yml`, offset ports).

---

## 11. Testing

- **Unit** (`tests/unit/`, ~40 files / 326 tests): no infra; respx for LLM
  HTTP; includes `test_bot_boundary.py` (AST check of the hexagonal bot rule).
- **Integration** (`tests/integration/`, 21 files / 134 tests, `-m
  integration`): full docker-compose test stack, alembic upgrade, seeded
  service token, ASGI httpx + LifespanManager; covers upload pipeline, each
  worker, credits/quota DB, auth flows, passkeys/magic links, B2B, admin
  panel, billing checkout/webhook/pay page, subscriptions, refunds.
- `tests/conftest.py` deduplicates `src.contract_check.*` vs
  `contract_check.*` module loading (avoids duplicate Prometheus timeseries).

---

## 12. Cross-cutting conventions

1. Money is integer kopecks everywhere; decimals exist only at the ЮKassa
   HTTP boundary.
2. Enum-ish columns are `TEXT + CHECK`; migrations additive and hand-written.
3. Repositories never commit; callers own transactions.
4. Every queue handler is idempotent via `documents.status` re-read;
   refunds/slot releases are exactly-once via guards.
5. Publisher confirms on every publish that follows a paid reservation.
6. External providers (LLM, payments, storage) sit behind ports in `core/`;
   adapters are swappable and respx-testable.
7. Feature degrade switches: `PLANS_ENABLED`, `YOOKASSA_ENABLED`,
   `PRESCREEN_ENABLED`, `PRESCREEN_LLM_FALLBACK_ENABLED`,
   `SMTP_HOST` (empty = dev logger), `REFUND_POLICY`.
8. Domain language (Plan/Subscription/Quota/Credits/Topup/Invoice/Renewal/
   Saved Payment Method/Dunning/Signup Bonus/ЕРИП/Billing Hold/Clawback/
   Document Slot) is normative — see root `CONTEXT.md`; ADRs live in
   `docs/adr/`.