add graphify, update readme, claude md and gitignore

Reviewed-on: #16

.claude/agents/code-reviewer.md

@@ -0,0 +1,7 @@
+---
+name: code-reviewer
+description: Expert code review specialist. Proactively reviews code for quality, security, and maintainability. Use immediately after writing or modifying code.
+tools: Read, Grep, Glob, Bash
+model: sonnet
+---
+You are a senior code reviewer. Read the recent git diff or modified files, then report only what matters: correctness bugs, security vulnerabilities, and maintainability traps. Lead with the highest-severity finding.

.claude/settings.local.json

@@ -0,0 +1,9 @@
+{
+  "$schema": "https://json.schemastore.org/claude-code-settings.json",
+  "sandbox": {
+    "filesystem": {
+      "allowRead": ["."],
+      "denyRead": ["**/.env", "./DB", "./RTMQ", "./.vscode", "./.pytest_cache", "**/__pycache__"]
+    }
+  }
+}

.gitignore

@@ -17,11 +17,16 @@ __pycache__/
 .DS_Store
 Thumbs.db
 
-#env
+# env files
 *.env
-#db
+
+# DB
 *.db
 DB/
+
+# rabbitmq
+RTMQ/
+
 # logs
 logs/
 
@@ -30,3 +35,4 @@ input/
 output/
 allure-results/
 .coverage
+graphify-out/

CLAUDE.md

@@ -0,0 +1,252 @@
+# The DisExcel Project — Context for Claude Code
+
+Backend: FastAPI + SQLAlchemy (async) + Pydantic v2 + PostgreSQL. JWT auth
+(access + refresh tokens), RBAC permissions, Redis cache, RabbitMQ background
+workers. Python >=3.13,<4.0, Poetry for dependency management.
+
+## Auth & Permissions
+
+- Users have `direct_permissions` (list of `Permissions`) and `group`
+  (list of `PermissionsGroups`, each with its own `permissions`) —
+  many-to-many both ways.
+- Effective permissions = `direct_permissions ∪ (union of all groups' permissions)`.
+- `require_permissions(*permissions)` in `src/web/protected_routes` is a
+  FastAPI dependency factory — wraps `CurrentUserService.get_current_user`.
+  Call with no args (`require_permissions()`) for "just authenticated, no
+  specific permission needed".
+- Access tokens carry a `jti` claim. Logout writes `revoked_access_token:{jti}`
+  to Redis with TTL = remaining token lifetime — `get_current_user` checks
+  this key before anything else.
+- `secure` flag on refresh_token cookie is driven by `env_settings.PROD_MODE`
+  (bool) — `False` locally/tests so cookies work over plain HTTP, `True` in
+  prod.
+
+## Redis (`src/cache/`)
+
+- `RedisClient(redis.Redis)` — module-level shared singleton, subclasses
+  `redis.Redis` directly (inherits all commands, no manual wrapping needed).
+- Three uses: permissions is-cache was considered and rejected (no real DB
+  savings — `get_user_by_id` already eager-loads everything via `selectin`
+  in one call); rate limiting on login (`RateLimit.rate_limit(ip)` —
+  `INCR` + `EXPIRE` on first attempt, blocks >5/60s); access-token revoke
+  blacklist (see above).
+- Rate limit is only triggered inside `except HTTPException` on `/protected/token`
+  — i.e. only on failed logins, not successful ones (otherwise legitimate
+  repeated logins would trip it).
+
+## RabbitMQ (`src/messaging/`)
+
+- `RabbitMQClient` — shared class, lazy `connect()` (can't be async `__init__`),
+  holds one `connection` + one `channel`, `get_channel()` ensures setup.
+  `connect()` retries `connect_robust()` up to 5 times with exponential
+  backoff (`asyncio.sleep(2**attempt)`) before raising — needed because
+  when running the app locally (outside Docker Compose) right after
+  `docker compose up rabbitmq`, the broker's AMQP listener isn't
+  immediately ready and resets the connection mid-handshake
+  (`ConnectionResetError`). Compose's own `depends_on: condition:
+  service_healthy` only helps when the app itself also runs as a Compose
+  service — it does nothing when the app is started on the host.
+- **Topology is centralized, not declared ad-hoc.** `src/models/rabbitmq_models/email.py`
+  defines the whole topology as pydantic data (`Topology`/`ExchangeSpec`/`QueueSpec`,
+  exported as `email_topology`) — topic exchange `"email"`, one durable
+  quorum queue per message type (`queue_welcome_email`/`email.welcome`,
+  `queue_reset_email`/`email.reset`), each with a matching dead-letter queue.
+  `src/messaging/topology_setup.py::apply_topology(channel, topology)` is
+  the only place that actually calls `declare_exchange`/`declare_queue`/`bind`
+  against RabbitMQ. It's called once at startup — in `main.py`'s `lifespan`
+  (web) and in `daemon_run.py`'s `main()` (daemon entrypoint) — **before**
+  any producer/consumer touches the exchange/queues. `EmailProducer`/
+  `WelcomeEmailConsumer`/`ResetEmailConsumer` no longer declare or bind
+  anything themselves — they only `get_exchange("email")` /
+  `get_queue("queue_...")`, assuming `apply_topology` already ran.
+  Changing a queue's arguments in `email_topology` requires manually
+  deleting the existing queue in RabbitMQ first — `declare_queue` on an
+  already-existing queue with different arguments fails with
+  `PRECONDITION_FAILED`, it doesn't update in place.
+- **Dead-lettering**: each original queue's `x-dead-letter-exchange` points
+  at `email.dlx`, a **`direct`** exchange (not `fanout` — fanout would
+  broadcast every dead-lettered message to *all* DLQs, mixing up
+  welcome/reset failures). Each DLQ is bound to `email.dlx` with the same
+  routing key as the original queue (`email.welcome`/`email.reset`);
+  RabbitMQ preserves a message's original routing key when dead-lettering,
+  so no explicit `x-dead-letter-routing-key` override is needed — the
+  `direct` exchange alone routes each dead letter back to the correct DLQ.
+- `message.process(ignore_processed=True)` + manual `await message.nack(...)`
+  in `process_message`, not the plain auto-ack/nack that `message.process()`
+  does by default — needed because business-logic exceptions are caught
+  *inside* the block (to log them and decide retry vs. dead-letter) rather
+  than left to propagate, so `process()` itself must be told not to also
+  try to ack/nack on exit. Classification: transient errors (SMTP/network:
+  `SMTPServerDisconnected`, `SMTPConnectError`, `TimeoutError`,
+  `ConnectionRefusedError`) → `nack(requeue=True)`; everything else
+  (bad message body, `SMTPAuthenticationError`, bugs) → `nack(requeue=False)`
+  → DLQ. Note `SMTPAuthenticationError` is a known rough edge here — it's
+  a systemic config problem (bad credentials), not a per-message one, but
+  is currently DLQ'd per-message like any other permanent error rather
+  than halting the daemon with an alert.
+- Email templates: Jinja2 (`src/service/email/jinja_env.py` — one shared
+  `Environment`/`FileSystemLoader` singleton, templates in
+  `src/service/email/templates/`), inline CSS (email clients don't support
+  `<style>` reliably), `EmailMessage` with `set_content()` (plain-text
+  fallback) + `add_alternative(html, subtype="html")`.
+- Reset-password flow mirrors welcome: `ResetEmailSender`
+  (`src/service/email/email_reset.py`) renders `templates/reset.html`
+  (`{{ temp_password }}`, no longer hardcoded) the same way
+  `DaemonEmailSender` does `welcome.html`. `ResetEmailConsumer.process_message`
+  reads both `email` and `temp_password` from the message body and uses the
+  same transient/permanent classification as `WelcomeEmailConsumer`.
+  `EmailProducer.send_reset_email(email, temp_password)` takes the password
+  as a second argument now. **Still missing**: nothing in the app actually
+  calls `send_reset_email` yet — there's no password-reset route that
+  generates a `temp_password` and publishes it. Don't assume the
+  reset-password feature is reachable end-to-end until that route exists.
+
+## Logging (`src/logging/`)
+
+- All log output (HTTP endpoints, SQL, daemons) funnels through one
+  module-level `asyncio.Queue` (`log_queue` in `src/logging/logger.py`) and
+  a single `LogWriter.log_writer()` background task that drains it and
+  appends to `./logs/{type}_log_{month}_{year}.txt` — this avoids the
+  original bug of firing an unsupervised `asyncio.create_task` per log
+  line (unordered writes, lost logs if the process died before the task
+  ran). **`log_writer()` must be started explicitly wherever the process
+  entrypoint is** — it's `asyncio.create_task(writer.log_writer())` in
+  `main.py`'s `lifespan` for the web process, and the same call in
+  `daemon_run.py`'s `main()` for the daemon process. `log_queue` is
+  process-local (plain in-memory `asyncio.Queue`, not shared across
+  processes), so **every new entrypoint needs its own writer task** or its
+  logs silently queue forever and are never written (unbounded memory
+  growth, not a crash).
+- `logging.Handler` subclasses (`LoggerDB` for `sqlalchemy.engine`,
+  `LoggerDaemon` for the `"daemon"` logger) just push `(type, formatted_msg)`
+  onto `log_queue` from `emit()` — do **not** give them a custom `__init__`
+  that doesn't call `super().__init__()`; skipping it means `self.level`/
+  `self.filters`/etc. never get set and any log call raises `AttributeError:
+  'LoggerX' object has no attribute 'level'`.
+- Two `ContextVar`s tag log lines with a correlation id without threading
+  it through every function call: `request_id_ctx` (set once per HTTP
+  request in `LoggingMiddleware.dispatch`) and `message_id_ctx` (meant to
+  be set once per RabbitMQ message in `process_message`). Only visible
+  within the same async call chain — a `ContextVar` set in an HTTP request
+  is `"-"` (the default) inside daemon code, and vice versa; they don't
+  cross the process boundary either.
+- **`src/logging/logger.py` vs `src/logging/http_logger.py` split matters
+  for Docker.** `logger.py` has zero `fastapi`/`starlette` imports —
+  intentional, because the `daemon` Poetry group (and therefore the daemon
+  Docker image) never installs `fastapi`. `http_logger.py` holds
+  `LoggingMiddleware`/`ProcessingTimeMiddleware` (the only things that
+  actually need FastAPI/Starlette) and is imported only from `main.py`.
+  `src/logging/__init__.py` imports only from `logger.py`. **Never import
+  `src.logging.http_logger` from anything that runs in the daemon
+  process** (`consumers.py`, `email_welcome.py`, `daemon_run.py`, etc.) —
+  it would drag in `fastapi`, which raises `ModuleNotFoundError` in the
+  daemon container.
+- Plain `logger.exception(...)` must only be called from inside an
+  `except` block — it pulls the active exception via `sys.exc_info()` to
+  attach a traceback. Called outside `except` (e.g. for a routine "message
+  received" log line), it still runs but appends a literal `NoneType: None`
+  instead of a traceback, since there's no active exception to format.
+
+## Daemons / workers (`src/daemons/`)
+
+- `BaseDaemon` ABC (`name` + async `run()`), one subclass per consumer
+  (`WelcomeEmailDaemon`, `ResetEmailDaemon`, more to come — e.g. reports).
+- `DAEMONS` registry dict maps string name → daemon class.
+- `daemon_run.py` (project root) is the single entrypoint: `python
+  daemon_run.py <name>` runs one daemon, `python daemon_run.py --all` reads
+  `configs/daemons.json` (`{"daemons": [...]}`) and runs all enabled ones
+  concurrently via `asyncio.gather`. `main()` also applies the RabbitMQ
+  topology (`apply_topology`) and starts the log writer task before
+  dispatching to either mode, inside a `try`/`finally` so both the
+  single-daemon and `--all` paths cancel the writer task on the way out.
+- Each daemon runs as its own Docker service/container (`command: ["python",
+  "daemon_run.py", "<name>"]`), same pattern as the `migration` service.
+- `daemons.json` is read with a plain Pydantic `BaseModel` + manual
+  `json.load`, NOT `pydantic-settings` `json_file` — that requires wiring
+  `settings_customise_sources` manually in this pydantic-settings version
+  and isn't worth the complexity here.
+
+## Docker / Poetry groups
+
+- `pyproject.toml` uses PEP 621 `[project.dependencies]` for shared deps
+  (sqlalchemy, redis, aio-pika, pydantic, bcrypt, jose, aiofiles, asyncpg,
+  psycopg2-binary, alembic, greenlet).
+- `web` group: fastapi, uvicorn, gunicorn, python-multipart — only needed
+  by the API server.
+- `daemon` group: worker-only deps — `aiosmtpd`, `jinja2` (email sending
+  and templating). Must **not** gain a `fastapi`/`starlette` dependency;
+  see the `src/logging/logger.py` vs `http_logger.py` split above for why
+  that boundary is enforced deliberately.
+- `dev` group: pytest stack, allure, httpie, requests-async.
+- `claude` group: `graphifyy` (the `/graphify` Claude Code skill) — dev
+  tooling only, never referenced by any Dockerfile stage, install with
+  `poetry install --with claude` when needed locally.
+- Dockerfile has parallel builder→final stage pairs: `builder`→`prod`
+  (installs `main,web`) and `worker-builder`→`worker` (installs
+  `main,daemon`). Same base pattern: venv builder stage copies
+  `/opt/venv` into a clean final stage, poetry itself is uninstalled
+  after install to keep the final image lean.
+- Alembic runs against a **separate sync engine** (`asyncpg` swapped out,
+  psycopg2 used instead) — async SQLAlchemy engine can't drive Alembic
+  directly without the `run_sync` bridge, and a dedicated sync engine is
+  simpler than that bridge.
+- `DB_HOST` differs between contexts: `psql` (Docker service name) for
+  containers talking to each other, `localhost` for anything run on the
+  host (e.g. local `alembic revision --autogenerate`). Compose services
+  override `DB_HOST` via `environment:`; the `.env` file's own default is
+  for host-side runs.
+
+## Codebase knowledge graph (graphify)
+
+- `/graphify` (Claude Code skill, `graphifyy` package in the `claude`
+  Poetry group) builds a navigable knowledge graph of this repo into
+  `graphify-out/` (`graph.html`, `GRAPH_REPORT.md`, `graph.json`) —
+  gitignored, regenerable, never commit it.
+- Its own interpreter/venv detection defaults to `uv tool`/pipx-style
+  installs; **this project has neither** — `graphifyy` is installed
+  straight into the project's own `.venv` via Poetry, so point graphify
+  at `.venv/bin/python3` directly rather than letting it search for `uv`.
+- The sandbox's blanket `**/.env`/`**/.env.*` read-deny blocks graphify
+  from reading `configs/.env.example` too (not just the real `.env`) —
+  it shows up as `skipped_sensitive` in detection. This is a sandbox-level
+  block, not graphify's own sensitive-file heuristic; a real `.env` would
+  be excluded by graphify anyway, but `.env.example` (no real secrets)
+  would otherwise be safe to include if the sandbox allowed reading it.
+- A `GRAPH HEALTH WARNING` (dangling-endpoint edges, collapsed edges) on
+  a fresh build most likely means the AST extractor's node-ID format
+  changed since a previous partial run — re-run with `graphify extract
+  --force` if it persists across rebuilds; a one-off warning on first
+  build isn't necessarily a problem.
+
+## Testing (`tests/unit`, `tests/integrated`, `tests/e2e`)
+
+- **Recurring root cause of "different event loop" / `MissingGreenlet`-style
+  errors**: prod code uses module-level singletons (`engine`, `redis_client`)
+  created once at import time and reused for the app's whole lifetime — this
+  is correct for prod (one event loop, whole uptime) but breaks under
+  pytest-asyncio's default `function`-scoped event loop (a new loop per
+  test, but the singleton's connections stay bound to the *first* loop).
+  Fix: test fixtures create a **fresh** `engine`/`RedisClient` per test and
+  monkeypatch or inject them in place of the global singleton, then dispose
+  on teardown — not a global `session`-scoped event loop (that would mask
+  real isolation bugs).
+- e2e `MySession` must subclass `httpx.AsyncClient` (not `requests_async.
+  AsyncSession` — that library silently drops cookies between requests,
+  which broke refresh-token-cookie-dependent tests like logout).
+- `test_user_fixture` is `indirect=True` parametrized with
+  `(direct_permissions, group)` tuples.
+- `tests/unit/test_consumers.py` covers `RabbitMQClient`/`WelcomeEmailConsumer`/
+  `ResetEmailConsumer` entirely with mocks — no real broker involved.
+  Pattern: `monkeypatch.setattr(rabbitmq_client_module.aio_pika, "connect_robust", ...)`
+  patches the module attribute that `connect()` looks up at call time (not
+  the `rabbitmq_client` singleton's method — that's a bound method, it has
+  no attribute of its own to patch). `aio_pika.connect_robust`/`asyncio.sleep`
+  must both be mocked when testing the retry loop, or the test really
+  sleeps `2**attempt` seconds between attempts. `message.process(...)` is an
+  async context manager, not a plain awaitable — mocking it needs a
+  `MagicMock` with `__aenter__`/`__aexit__` set to `AsyncMock`s (see
+  `make_fake_message()` in that file), not just `AsyncMock()`. When
+  asserting on what a mocked async method returned, compare against
+  `mock.return_value` (or a variable captured before assigning it), never
+  against the mock itself — `some_mock is some_mock.return_value` is never
+  true, they're two different objects.

configs/.env.example

@@ -13,4 +13,17 @@ REDIS_PASSWORD="change_me"
 REDIS_PORT=change_me #int
 REDIS_HOST="change_me"
 
+
+RABBITMQ_PASSWORD="change_me"
+RABBITMQ_LOGIN="change_me"
+RABBITMQ_HOST="change_me"
+RABBITMQ_PORT=change_me   #int
+RABBITMQ_PORT_UI=chane_me #int
+
+EMAIL_PORT=change_me #int
+SMTP_SERVER="change_me"
+EMAIL_LOGIN="change_me"
+EMAIL_PASSWORD="change_me"
+
+
 PROD_MODE=bool

configs/daemons.json

@@ -0,0 +1,7 @@
+{
+    "daemons": 
+    [
+        "welcome_email", 
+        "reset_email"
+    ]
+}

daemon_run.py

@@ -0,0 +1,67 @@
+import asyncio
+import os
+import sys
+
+from src.daemons.registry import DAEMONS
+from src.logging.logger import LogWriter
+from src.messaging.rabbitmq_client import rabbitmq_client
+from src.messaging.topology_setup import apply_topology
+from src.models.configs_read.daemons_json import daemons_config
+from src.models.rabbitmq_models.email import email_topology
+
+writer = LogWriter()
+
+async def run_one(daemon_name: str) -> None:
+    daemon_cls = DAEMONS.get(daemon_name)
+    if daemon_cls is None:
+        print(f"Unknown daemon: {daemon_name}. Available: {list(DAEMONS.keys())}")
+        sys.exit(1)
+    daemon = daemon_cls()
+    await daemon.run()
+
+
+async def run_enabled_from_config() -> None:
+    
+    daemons = []
+    
+    for name in daemons_config.daemons:
+        name = name.strip()
+        if name not in DAEMONS:
+            print(f"Warning: unknown daemon '{name}' in config, skipping")
+            continue
+        daemons.append(DAEMONS[name]())
+
+    if not daemons:
+        print("No enabled daemons found in config")
+        sys.exit(1)
+        
+    await asyncio.gather(*(d.run() for d in daemons))
+
+
+async def main() -> None:
+    channel = await rabbitmq_client.get_channel()
+    await apply_topology(channel, email_topology)
+    writer_task = asyncio.create_task(writer.log_writer())
+    try:
+        if len(sys.argv) < 2:
+            print("Usage: python run_daemon.py <daemon_name> | --all")
+            sys.exit(1)
+
+        arg = sys.argv[1]
+        if arg == "--all":
+            await run_enabled_from_config()
+        else:
+            await run_one(arg)
+    finally:
+        writer_task.cancel()
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print('Interrupted')
+        try:
+            sys.exit(0)
+        except SystemExit:
+            os._exit(0)

docker/compose-dev.yaml

@@ -1,6 +1,7 @@
 name: disexcel
 
 services:
+
   backend-dev:
     profiles: ["dev"]
     image: "${DOCKER_REGISTRY:-local}/excel-dev:${IMAGE_TAG:-local}"
@@ -13,6 +14,8 @@ services:
     ipc: private #Inter-Process Communication (IPC) namespace for high-performance applications
     environment:
       - DB_HOST=psql          #rewrite DB_HOST var to communicate inside the docker network
+      - RABBITMQ_HOST=rabbitmq  
+      - REDIS_HOST=redis
     volumes:
       - type: bind
         source: ../src
@@ -32,6 +35,9 @@ services:
       - type: bind
         source: ../logs
         target: /home/excel-project/logs
+      - type: bind
+        source: ../daemon_run.py
+        target: /home/excel-project/daemon_run.py
     networks:
       - backend
     depends_on:
@@ -39,6 +45,8 @@ services:
         condition: service_healthy
       redis:
         condition: service_healthy
+      rabbitmq:
+        condition: service_healthy
     ports:
       - "80:8000"
     entrypoint: ["./entrypoint.sh", "--dev"]
@@ -55,6 +63,8 @@ services:
       ipc: private #Inter-Process Communication (IPC) namespace for high-performance applications
       environment:
         - DB_HOST=psql
+        - RABBITMQ_HOST=rabbitmq
+        - REDIS_HOST=redis
       volumes:
         - type: bind
           source: ../configs
@@ -76,11 +86,39 @@ services:
           condition: service_healthy
         redis:
           condition: service_healthy
+        rabbitmq:
+          condition: service_healthy
       ports:
         - "80:8000"
 
+  daemons:
+    profiles: ["prod", "dev"]
+    image: "${DOCKER_REGISTRY:-local}/excel-daemons:${IMAGE_TAG:-local}"
+    container_name: daemons
+    environment:
+      - RABBITMQ_HOST=rabbitmq  
+    build:
+      dockerfile: ./docker/dockerfile
+      context: ../
+      target: daemon
+    init: true
+    ipc: private
+    volumes:
+      - type: bind
+        source: ../configs
+        target: /home/excel-project/configs
+      - type: bind
+        source: ../logs
+        target: /home/excel-project/logs
+    depends_on:
+      rabbitmq:
+        condition: service_healthy
+    networks:
+      - backend
+    restart: unless-stopped
+
   psql:
-    profiles: ["prod", "dev", "db"]
+    profiles: ["prod", "dev", "local"]
     image: postgres:16-alpine
     container_name: psql
     init: true
@@ -106,7 +144,7 @@ services:
 
   redis:
     image: redis:latest
-    profiles: ["prod", "dev", "redis"]
+    profiles: ["prod", "dev", "local"]
     container_name: redis
     init: true
     ipc: private
@@ -119,7 +157,33 @@ services:
       timeout: 3s
       retries: 5
     restart: unless-stopped
+    networks:
+      - backend
   
+  rabbitmq:
+    image: rabbitmq:3-management-alpine
+    profiles: ["prod", "dev", "local"]
+    container_name: rabbitmq_broker
+    restart: always
+    init: true
+    ipc: private
+    environment:
+      - RABBITMQ_DEFAULT_USER=${RABBITMQ_LOGIN}
+      - RABBITMQ_DEFAULT_PASS=${RABBITMQ_PASSWORD}
+    ports:
+      - "${RABBITMQ_PORT}:5672"
+      - "${RABBITMQ_PORT_UI}:15672"
+    volumes:
+      - type: bind
+        source: ../RTMQ
+        target: /var/lib/rabbitmq
+    healthcheck:
+      test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+    networks:
+      - backend
 
 networks:
   backend:

docker/dockerfile

@@ -34,7 +34,7 @@ COPY pyproject.toml poetry.lock ./
 
 RUN pip install --no-cache-dir poetry \
     && poetry config virtualenvs.create false \
-    && poetry install --no-root --no-interaction --only main \
+    && poetry install --no-root --no-interaction --only main,web \
     && pip uninstall -y poetry poetry-core poetry-plugin-export 
 
 # --- Stage 2: Python Backend prod ---
@@ -63,3 +63,49 @@ RUN groupadd --gid 1000 appuser \
 USER appuser
 
 ENTRYPOINT ["./entrypoint.sh", "--prod"]
+
+# --- Stage 1: Python daemon builder ---
+
+FROM python:3.14-slim AS worker-builder
+
+LABEL org.opencontainers.image.title="The-DisExcel-project-dev"
+LABEL org.opencontainers.image.source="https://git.homyk.space/MH.Dmitrii/The_DisExcel_project"
+
+WORKDIR /home/excel-project
+
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+
+RUN python -m venv "$VIRTUAL_ENV"
+
+COPY pyproject.toml poetry.lock ./
+
+RUN pip install --no-cache-dir poetry \
+    && poetry config virtualenvs.create false \
+    && poetry install --no-root --no-interaction --only main,daemon \
+    && pip uninstall -y poetry poetry-core poetry-plugin-export 
+
+# --- Stage 2: Python daemons ---
+
+FROM python:3.14-slim AS daemon
+
+LABEL org.opencontainers.image.title="The-DisExcel-project-prod"
+LABEL org.opencontainers.image.source="https://git.homyk.space/MH.Dmitrii/The_DisExcel_project"
+
+WORKDIR /home/excel-project
+
+ENV VIRTUAL_ENV=/opt/venv
+ENV PATH="$VIRTUAL_ENV/bin:$PATH"
+
+COPY --from=worker-builder /opt/venv /opt/venv
+
+COPY pyproject.toml poetry.lock alembic.ini daemon_run.py  ./
+COPY src/ ./src/
+
+RUN groupadd --gid 1000 appuser \
+    && useradd --uid 1000 --gid appuser --shell /bin/bash --create-home appuser \
+    && chown -R appuser:appuser /home/excel-project
+
+USER appuser
+
+ENTRYPOINT ["python","-u", "daemon_run.py", "--all"]

main.py

@@ -1,25 +1,34 @@
+import asyncio
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-# import uvicorn
 from fastapi import FastAPI
 
 from src.cache.redis_client import redis_client
 from src.database.users.crud import Seed
-from src.logging.logger import LoggingMiddleware, ProcessingTimeMiddleware
+from src.logging.http_logger import LoggingMiddleware, ProcessingTimeMiddleware
+from src.logging.logger import LogWriter
+from src.messaging.rabbitmq_client import rabbitmq_client
+from src.messaging.topology_setup import apply_topology
+from src.models.rabbitmq_models.email import email_topology
 from src.web.protected_routes.auth_routes import router as protected_router
 from src.web.protected_routes.protected_user_action_routes import (
     router as protected_user_action_routes,
 )
 
+writer=LogWriter()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     create_dirs()
+    channel = await rabbitmq_client.get_channel()
+    await apply_topology(channel, email_topology)
     await create_first_user()
+    writer_task=asyncio.create_task(writer.log_writer())
     yield
+    writer_task.cancel()
     await redis_client.close()
-    
+    await rabbitmq_client.close()
 
 app=FastAPI(root_path="/", lifespan=lifespan)
 app.add_middleware(LoggingMiddleware)

makefile

@@ -17,11 +17,11 @@ help:
 ##
 .PHONY: run
 run: ## Run dev local application
-	docker compose -f docker/compose-dev.yaml --env-file configs/.env --profile db --profile redis up -d && ${VENV} uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+	docker compose -f docker/compose-dev.yaml --env-file configs/.env --profile local up -d && ${VENV} uvicorn main:app --host 0.0.0.0 --port 8000 --reload
 
 .PHONY: down
 down: ## Down dev local db
-	docker compose -f docker/compose-dev.yaml --env-file configs/.env --profile db --profile redis down
+	docker compose -f docker/compose-dev.yaml --env-file configs/.env --profile local down
 
 .PHONY: run-dev
 run-dev: ## Run dev application

pyproject.toml

@@ -7,27 +7,29 @@ authors = [
 ]
 license = "MH.Dmitrii's project"
 readme = "README.md"
-requires-python = ">=3.13"
+requires-python = ">=3.13,<4.0"
 
 dependencies = [
     "alembic (>=1.18.5,<2.0.0)",
-    "uvicorn (>=0.51.0,<0.52.0)",
-    "gunicorn (>=26.0.0,<27.0.0)",
-    "fastapi (>=0.139.1,<0.140.0)",
     "pydantic[email] (>=2.13.4,<3.0.0)",
     "pydantic-settings (>=2.14.2,<3.0.0)",
     "sqlalchemy[asyncio] (>=2.0.52,<3.0.0)",
-    "pandas (>=3.0.3,<4.0.0)",
     "bcrypt (>=5.0.0,<6.0.0)",
     "python-jose (>=3.5.0,<4.0.0)",
-    "python-multipart (>=0.0.32,<0.0.33)",
     "greenlet (>=3.5.4,<4.0.0)",
     "aiofiles (>=25.1.0,<26.0.0)",
     "asyncpg (>=0.31.0,<0.32.0)",
     "psycopg2-binary (>=2.9.12,<3.0.0)",
     "redis (>=8.1.0,<9.0.0)",
+    "aio-pika (>=10.0.1,<11.0.0)"
 ]
 
+[tool.poetry.group.web.dependencies]
+uvicorn = ">=0.51.0,<0.52.0"
+gunicorn = ">=26.0.0,<27.0.0"
+fastapi = ">=0.139.1,<0.140.0"
+python-multipart = ">=0.0.32,<0.0.33"
+
 [tool.poetry.group.dev.dependencies]
 pytest = ">=9.1.1,<10.0.0"
 pytest-cov = ">=7.1.0,<8.0.0"
@@ -38,6 +40,13 @@ httpie = ">=3.2.4,<4.0.0"
 pytest-asyncio = ">=1.4.0,<2.0.0"
 requests-async = ">=0.2.4,<0.3.0"
 
+[tool.poetry.group.daemon.dependencies]
+aiosmtpd = ">=1.4.6,<2.0.0"
+jinja2 = ">=3.1.6,<4.0.0"
+
+[tool.poetry.group.claude.dependencies]
+graphifyy = ">=0.9.63,<0.10.0"
+
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]
 build-backend = "poetry.core.masonry.api"
@@ -51,7 +60,12 @@ omit = [
     "__init__.py",
     "*/docker/*",
     "*/rate_limit.py",
-    "*/logger.py"
+    "*/logger.py",
+    "*/daemons/*",
+    "*/topology_setup.py",
+    "*/logging/*",
+    "*/email_reset.py",
+    "*/email_welcome.py"
 ]
 
 [tool.coverage.report]

readme.md

@@ -10,24 +10,50 @@ A FastAPI project combining Excel and digital data ("The Great Excel project tha
 - PostgreSQL (asyncpg, psycopg2)
 - Pydantic 2 / Pydantic Settings
 - Poetry — dependency management
+- Redis — caching, rate limiting, token revocation
+- RabbitMQ (aio-pika) — background email workers
+- Jinja2 — HTML email templates
 - Docker, Ansible — deployment
 
 ## Architecture
 
 ```
 src/
+├── cache/       # Redis client, rate limiting
+├── daemons/     # background worker entrypoints (BaseDaemon, registry)
 ├── database/    # DB CRUD operations
 ├── errors/      # HTTP errors
-├── logging/     # logging middleware
+├── logging/     # queue-based logging infra + HTTP middleware
+├── messaging/   # RabbitMQ client, producers, consumers, topology
 ├── migrations/  # Alembic migrations
-├── models/      # Pydantic and SQLAlchemy models, configs
+├── models/      # Pydantic and SQLAlchemy models, configs, RabbitMQ topology
 ├── reports/     # reports
-├── service/     # business logic (auth, users_crud)
+├── service/     # business logic (auth, users_crud, email sending)
 └── web/         # routes (protected_routes)
 ```
 
 Layers are connected top to bottom: `web → service → database → models`.
 
+## Background workers (RabbitMQ)
+
+Email sending (welcome / password-reset) runs as separate daemon processes,
+decoupled from the web API via a RabbitMQ topic exchange:
+
+```
+main.py / daemon_run.py → apply_topology() → RabbitMQ ("email" exchange)
+                                                   ├── queue_welcome_email → WelcomeEmailConsumer
+                                                   └── queue_reset_email  → ResetEmailConsumer
+```
+
+Each queue has a matching dead-letter queue for messages that fail
+permanently (bad data, non-retryable errors) instead of retrying forever.
+Run a worker locally with:
+
+```bash
+python daemon_run.py welcome_email   # single daemon
+python daemon_run.py --all           # all daemons enabled in configs/daemons.json
+```
+
 ## Authentication
 
 - JWT access + refresh tokens
@@ -46,12 +72,72 @@ tests/
 
 Uses pytest, pytest-asyncio, pytest-cov, pytest-mock, allure-pytest.
 
+### Allure report
+
+`make allure` needs the Allure **command-line tool** (Java-based, not a pip
+package) — `allure-pytest` only writes raw result files, the CLI turns them
+into an HTML report.
+
+1. Download Allure **2.44.0** from
+   [github.com/allure-framework/allure2/releases](https://github.com/allure-framework/allure2/releases).
+2. Extract it into `.venv/allure-2.44.0/` so that `.venv/allure-2.44.0/bin/allure`
+   exists — this matches the `ALLURE` variable already set in `makefile`.
+3. If you install it somewhere else (or on Windows), update the `ALLURE`
+   variable at the top of `makefile` to point to your actual `allure`
+   binary path — the Windows path is already there, commented out.
+
+```bash
+make test     # runs pytest, writes results to tests/allure-results/reports
+make allure   # builds tests/allure-results/html/index.html from those results
+```
+
 ## Installation
 
+Requires Poetry itself to be installed first (it's a global tool, not a
+project dependency). Recommended via [pipx](https://pipx.pypa.io/):
+
+```bash
+pipx install poetry
+```
+
+or via the official installer:
+
+```bash
+curl -sSL https://install.python-poetry.org | python3 -
+```
+
+Then install the project dependencies:
+
 ```bash
 poetry install
 ```
 
+## Codebase knowledge graph (graphify)
+
+The repo can be explored as a navigable knowledge graph via the
+[graphify](https://github.com/safishamsi/graphify) Claude Code skill —
+useful for onboarding or tracing how a change ripples across modules.
+It lives in its own Poetry group (`claude`) so it's never installed in
+`web`/`daemon`/prod images:
+
+```bash
+poetry install --with claude
+```
+
+Then, inside a Claude Code session in this repo, run:
+
+```text
+/graphify
+```
+
+This builds `graphify-out/graph.html` (open directly in a browser),
+`graphify-out/GRAPH_REPORT.md` (god nodes, surprising cross-module
+connections, suggested questions), and `graphify-out/graph.json` (raw
+graph data). Ask follow-up questions about the codebase directly — once
+`graphify-out/graph.json` exists, Claude answers from the graph instead
+of rebuilding it. `graphify-out/` is gitignored: it's regenerable local
+output, not part of the codebase.
+
 ## Running migrations
 
 ```bash

src/daemons/base.py

@@ -0,0 +1,9 @@
+from abc import ABC, abstractmethod
+
+
+class BaseDaemon(ABC):
+    name: str
+
+    @abstractmethod
+    async def run(self) -> None:
+        ...

src/daemons/email_daemons.py

@@ -0,0 +1,19 @@
+from src.daemons.base import BaseDaemon
+from src.messaging.consumers.consumers import (
+    reset_email_consumer,
+    welcome_email_consumer,
+)
+
+
+class WelcomeEmailDaemon(BaseDaemon):
+    name = "welcome_email"
+
+    async def run(self) -> None:
+        await welcome_email_consumer.start_consuming()
+
+
+class ResetEmailDaemon(BaseDaemon):
+    name = "reset_email"
+
+    async def run(self) -> None:
+        await reset_email_consumer.start_consuming()

src/daemons/registry.py

@@ -0,0 +1,6 @@
+from src.daemons.email_daemons import ResetEmailDaemon, WelcomeEmailDaemon
+
+DAEMONS = {
+    "welcome_email": WelcomeEmailDaemon,
+    "reset_email": ResetEmailDaemon,
+}

src/fake/__init__.py

@@ -1 +0,0 @@
-'''fake data for tests'''

src/logging/__init__.py

@@ -2,8 +2,14 @@
 
 import logging
 
-from .logger import LoggerDB
+from .logger import LoggerDaemon, LoggerDB
 
 sql_logger = logging.getLogger("sqlalchemy.engine")
 sql_logger.setLevel(logging.INFO)
 sql_logger.addHandler(LoggerDB())
+sql_logger.addHandler(logging.StreamHandler())
+
+daemon_logger=logging.getLogger("daemon")
+daemon_logger.setLevel(logging.INFO)
+daemon_logger.addHandler(LoggerDaemon())
+daemon_logger.addHandler(logging.StreamHandler())

src/logging/http_logger.py

@@ -0,0 +1,63 @@
+import json
+from time import perf_counter
+from typing import cast
+from uuid import uuid4
+
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from starlette.concurrency import iterate_in_threadpool
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import Response, StreamingResponse
+
+from src.logging.logger import log_queue, request_id_ctx
+
+
+class ProcessingTimeMiddleware(BaseHTTPMiddleware, ):
+    async def dispatch(self, request: Request, call_next)->Response:
+        start_time = perf_counter()
+        response = await call_next(request)
+        process_time = perf_counter() - start_time
+        response.headers["X-Process-Time"] = str(process_time)
+        return response
+
+
+class LoggingMiddleware(BaseHTTPMiddleware):
+    
+    async def build_log_line(self, request_id, method, path, status_code, detail, client_ip) -> str:
+        return f"[{request_id}] [{method}] [{path}] [{status_code}] [{detail}] [{client_ip}]"
+
+    async def dispatch(self, request: Request, call_next) -> Response:
+        request_id=str(uuid4())
+        request_id_ctx.set(request_id)
+        client_ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip() or (request.client.host if request.client else 'unknown')
+        method = request.method
+        path=request.url.path
+        try:
+            response = await call_next(request)
+        except Exception as exc:  # noqa: BLE001
+            line=await self.build_log_line(request_id=request_id, method=method, path=path, status_code=500, detail=repr(exc), client_ip=client_ip)
+            log_queue.put_nowait(("endpoints", line))
+            return JSONResponse(
+                status_code=500,
+                content={"detail": "Internal Server Error", "request_id": request_id}
+            )
+            
+        streaming_response = cast(StreamingResponse, response)
+
+        chunks = []
+        async for chunk in streaming_response.body_iterator:
+            chunks.append(chunk.encode() if isinstance(chunk, str) else bytes(chunk))
+
+        body_bytes = b"".join(chunks)
+        streaming_response.body_iterator = iterate_in_threadpool(iter([body_bytes]))
+
+        try:
+            parsed = json.loads(body_bytes)
+            body = parsed.get("detail", None) if not isinstance(parsed, bool) else None
+        except (json.JSONDecodeError, TypeError):
+            body = None
+            
+        line=await self.build_log_line(request_id=request_id, method=method, path=path, status_code=response.status_code, detail=body, client_ip=client_ip)
+        log_queue.put_nowait(("endpoints", line))
+    
+        return response

src/logging/logger.py

@@ -1,71 +1,43 @@
 import asyncio
-import json
 import logging
-from time import gmtime, perf_counter, strftime
+from contextvars import ContextVar
-from typing import cast
+from time import gmtime, strftime
 
 import aiofiles
-from fastapi import Request
-from fastapi.responses import JSONResponse
-from starlette.concurrency import iterate_in_threadpool
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.responses import Response, StreamingResponse
 
+request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")
+message_id_ctx: ContextVar[str] = ContextVar("message_id", default="-")
 
-class ProcessingTimeMiddleware(BaseHTTPMiddleware):
+log_queue=asyncio.Queue()
-    async def dispatch(self, request: Request, call_next)->Response:
-        start_time = perf_counter()
-        response = await call_next(request)
-        process_time = perf_counter() - start_time
-        response.headers["X-Process-Time"] = str(process_time)
-        return response
 
+class LogWriter:
+    def __init__(self) -> None:
+        pass
         
-class LoggingMiddleware(BaseHTTPMiddleware):
+    async def log_writer(self)->None:
-    async def dispatch(self, request: Request, call_next) -> Response:
+        while True:
-        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
+            log_type, msg=await log_queue.get()
+            await self._write(msg, log_type)
+    
+    async def _write(self, msg:str, type:str)->None:
         file_time = strftime("%b_%Y", gmtime())
-        client_ip = request.headers.get('x-forwarded-for', '').split(',')[0].strip() or (request.client.host if request.client else 'unknown')
+        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
         
-        try:
+        async with aiofiles.open(f"./logs/{type}_log_{file_time}.txt", "a") as f:
-            response = await call_next(request)
+            await f.write(f"[{current_time}] {msg}\n")
-        except Exception as exc:  # noqa: BLE001
-            body = str(exc)
-            async with aiofiles.open(f"./logs/endpoints_log_{file_time}.txt", "a") as file:
-                await file.write(f"[{current_time}] [500] [{body}] [{client_ip}]\n")
-            return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})
-
-        streaming_response = cast(StreamingResponse, response)
-
-        chunks = []
-        async for chunk in streaming_response.body_iterator:
-            chunks.append(chunk.encode() if isinstance(chunk, str) else bytes(chunk))
-
-        body_bytes = b"".join(chunks)
-        streaming_response.body_iterator = iterate_in_threadpool(iter([body_bytes]))
-
-        try:
-            parsed = json.loads(body_bytes)
-            body = parsed.get("detail", None) if not isinstance(parsed, bool) else None
-        except (json.JSONDecodeError, TypeError):
-            body = None
-
-        async with aiofiles.open(f"./logs/endpoints_log_{file_time}.txt", "a") as file:
-            await file.write(f"[{current_time}] [{response.status_code}] [{body}] [{client_ip}]\n")
-
-        return response
             
 
 class LoggerDB(logging.Handler):
     
     def emit(self, record: logging.LogRecord) -> None:
         msg = self.format(record)
-        asyncio.create_task(self._write(msg))
+        rid=request_id_ctx.get()
+        log_queue.put_nowait(("sql",f"[{rid}] {msg}"))
 
-    async def _write(self, msg: str) -> None:
 
-        file_time = strftime("%b_%Y", gmtime())
+class LoggerDaemon(logging.Handler):
-        current_time = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime())
     
-        async with aiofiles.open(f"./logs/sql_log_{file_time}.txt", "a") as f:
+    def emit(self, record: logging.LogRecord)->None:
-            await f.write(f"[{current_time}] {msg}\n")
+        msg= self.format(record)
+        mid=message_id_ctx.get()
+        log_queue.put_nowait(("daemon", f"[{mid}], {msg}"))

src/messaging/__init__.py

@@ -0,0 +1 @@
+# rabbitmq code

src/messaging/consumers/consumers.py

@@ -0,0 +1,105 @@
+import json
+import smtplib
+from uuid import uuid4
+
+from src.logging import daemon_logger
+from src.logging.logger import message_id_ctx
+from src.messaging.rabbitmq_client import rabbitmq_client
+from src.service.email.email_reset import ResetEmailSender
+from src.service.email.email_welcome import DaemonEmailSender
+
+
+class WelcomeEmailConsumer:
+    
+    def __init__(self) -> None:
+        self.channel = None
+        self.queue = None
+        self.daemon=DaemonEmailSender()
+        
+    async def setup(self) -> None:
+        self.channel = await rabbitmq_client.get_channel()
+        await self.channel.set_qos(prefetch_count=10)
+        self.queue = await self.channel.get_queue("queue_welcome_email")
+
+    
+    async def process_message(self, message) -> None:
+        message_id_ctx.set(str(uuid4()))
+        async with message.process(ignore_processed=True):
+            data = json.loads(message.body)
+            daemon_logger.info(f"Обрабатываю: {data}, метка: {message.routing_key}")
+            try:
+                await self.daemon.send_email(data.get("email"))
+            except (
+                smtplib.SMTPServerDisconnected,
+                smtplib.SMTPConnectError,
+                TimeoutError,
+                ConnectionRefusedError,
+            ) as exc:
+                daemon_logger.exception(f"transient error, retrying: {exc!r}")
+                await message.nack(requeue=True)
+            except Exception as exc:  # noqa: BLE001
+                daemon_logger.exception(f"permanent error, sending to DLQ: {exc!r}")
+                await message.nack(requeue=False)
+
+    
+    async def start_consuming(self)->None:
+        
+        if self.queue is None:
+            await self.setup()
+
+        queue = self.queue
+        if queue is None:
+            raise RuntimeError("Failed to set up RabbitMQ queue")
+
+        async with queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                await self.process_message(message)
+
+class ResetEmailConsumer:
+
+    def __init__(self) -> None:
+        self.channel = None
+        self.queue = None
+        self.daemon = ResetEmailSender()
+
+
+    async def setup(self) -> None:
+        self.channel = await rabbitmq_client.get_channel()
+        await self.channel.set_qos(prefetch_count=10)
+        self.queue = await self.channel.get_queue("queue_reset_email")
+
+
+    async def process_message(self, message) -> None:
+        message_id_ctx.set(str(uuid4()))
+        async with message.process(ignore_processed=True):
+            data = json.loads(message.body)
+            daemon_logger.info(f"Обрабатываю: {data}, метка: {message.routing_key}")
+            try:
+                await self.daemon.send_email(data.get("email"), data.get("temp_password"))
+            except (
+                smtplib.SMTPServerDisconnected,
+                smtplib.SMTPConnectError,
+                TimeoutError,
+                ConnectionRefusedError,
+            ) as exc:
+                daemon_logger.exception(f"transient error, retrying: {exc!r}")
+                await message.nack(requeue=True)
+            except Exception as exc:  # noqa: BLE001
+                daemon_logger.exception(f"permanent error, sending to DLQ: {exc!r}")
+                await message.nack(requeue=False)
+    
+    async def start_consuming(self)->None:
+        
+        if self.queue is None:
+            await self.setup()
+
+        queue = self.queue
+        if queue is None:
+            raise RuntimeError("Failed to set up RabbitMQ queue")
+
+        async with queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                await self.process_message(message)
+                
+welcome_email_consumer=WelcomeEmailConsumer()
+reset_email_consumer=ResetEmailConsumer()

src/messaging/producers/producers.py

@@ -0,0 +1,41 @@
+import json
+
+import aio_pika
+
+from src.messaging.rabbitmq_client import rabbitmq_client
+
+
+class EmailProducer:
+    
+    def __init__(self) -> None:
+        self.channel = None
+        self.exchange = None
+    
+    async def setup(self) -> None:
+        self.channel = await rabbitmq_client.get_channel()
+        self.exchange = await self.channel.get_exchange("email")
+
+        
+    async def send_welcome_email(self, email:str)->None:
+        
+        await self._publish({"email":email}, routing_key="email.welcome")
+        
+    async def send_reset_email(self, email: str, temp_password: str) -> None:
+
+        await self._publish({"email": email, "temp_password": temp_password}, routing_key="email.reset")
+    
+    async def _publish(self,data:dict, routing_key:str)->None:
+        
+        if self.exchange is None:
+            await self.setup()
+
+        if self.exchange is None:
+            raise RuntimeError("Failed to set up RabbitMQ exchange")
+
+        message = aio_pika.Message(
+            body=json.dumps(data).encode(),
+            delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+        )
+        await self.exchange.publish(message=message, routing_key=routing_key)
+
+email_producer=EmailProducer()

src/messaging/rabbitmq_client.py

@@ -0,0 +1,47 @@
+import asyncio
+
+import aio_pika
+from aio_pika.abc import AbstractChannel, AbstractRobustConnection
+
+from src.models.configs_read.env import env_settings
+
+
+class RabbitMQClient:
+    def __init__(self) -> None:
+        self.connection: AbstractRobustConnection | None = None
+        self.channel: AbstractChannel | None = None
+
+    async def connect(self) -> None:
+        for attempt in range(1,6):
+            try:
+                if self.connection is None or self.connection.is_closed:
+                    self.connection = await aio_pika.connect_robust(
+                        host=env_settings.RABBITMQ_HOST,
+                        port=env_settings.RABBITMQ_PORT,
+                        login=env_settings.RABBITMQ_LOGIN,
+                        password=env_settings.RABBITMQ_PASSWORD,
+                    )
+                break
+            except Exception as exc:
+                if attempt==5:
+                    raise
+                print(f"RabbitMQ not ready yet (attempt {attempt}/5): {exc!r}, retrying...")
+                await asyncio.sleep(2**attempt)
+        if self.connection is None:
+            raise RuntimeError("Failed to esablish RabbitMQ connection. Check the server!")
+        if self.channel is None or self.channel.is_closed:
+            self.channel = await self.connection.channel()
+
+    async def get_channel(self) -> AbstractChannel:
+        await self.connect()
+        if self.channel is None:
+            raise RuntimeError("Failed to establish RabbitMQ channel")
+        return self.channel
+
+    async def close(self) -> None:
+        if self.channel and not self.channel.is_closed:
+            await self.channel.close()
+        if self.connection and not self.connection.is_closed:
+            await self.connection.close()
+
+rabbitmq_client = RabbitMQClient()

src/messaging/topology_setup.py

@@ -0,0 +1,22 @@
+from src.models.rabbitmq_models.email import Topology
+
+
+async def apply_topology(channel, topology: Topology) -> None:
+    dlx = await channel.declare_exchange("email.dlx", "direct")
+
+    for exch_spec in topology.exchanges:
+        exchange = await channel.declare_exchange(exch_spec.name, exch_spec.type)
+        for q in exch_spec.queues:
+            queue = await channel.declare_queue(
+                q.name,
+                durable=True,
+                arguments={
+                    "x-queue-type": "quorum",
+                    "x-dead-letter-exchange": "email.dlx",
+                },
+            )
+            await queue.bind(exchange, routing_key=q.routing_key)
+
+            if q.dlx_queue:
+                dlq = await channel.declare_queue(q.dlx_queue, durable=True)
+                await dlq.bind(dlx, routing_key=q.routing_key)

src/models/configs_read/daemons_json.py

@@ -0,0 +1,28 @@
+from pydantic_settings import (
+    BaseSettings,
+    JsonConfigSettingsSource,
+    PydanticBaseSettingsSource,
+    SettingsConfigDict,
+)
+
+from src.models.configs_read.env import Base
+
+
+class DaemonsConfig(Base):
+    
+    daemons: list[str]
+    
+    model_config = SettingsConfigDict(json_file="configs/daemons.json")
+
+    @classmethod
+    def settings_customise_sources(
+        cls,
+        settings_cls: type[BaseSettings],
+        init_settings: PydanticBaseSettingsSource,
+        env_settings: PydanticBaseSettingsSource,
+        dotenv_settings: PydanticBaseSettingsSource,
+        file_secret_settings: PydanticBaseSettingsSource,
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
+        return (JsonConfigSettingsSource(settings_cls),)
+    
+daemons_config = DaemonsConfig() # type: ignore[call-arg]

src/models/configs_read/env.py

@@ -20,6 +20,17 @@ class Env(Base):
     REDIS_PORT:int
     REDIS_HOST:str
     
+    RABBITMQ_PASSWORD:str
+    RABBITMQ_LOGIN:str
+    RABBITMQ_HOST:str
+    RABBITMQ_PORT:int
+    RABBITMQ_PORT_UI:int
+    
+    EMAIL_PORT:int
+    SMTP_SERVER:str
+    EMAIL_LOGIN:str
+    EMAIL_PASSWORD:str
+    
     PROD_MODE:bool
     
     model_config=SettingsConfigDict(env_file="configs/.env", extra=None)

src/models/rabbitmq_models/email.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel
+
+
+class QueueSpec(BaseModel):
+    name:str
+    routing_key:str
+    dlx_queue:str|None = None
+
+class ExchangeSpec(BaseModel):
+    name:str
+    type:str = "topic"
+    queues: list[QueueSpec]
+
+class Topology(BaseModel):
+    exchanges:list[ExchangeSpec]
+    
+email_topology=Topology(exchanges=[
+    ExchangeSpec(name="email", queues=[
+        QueueSpec(name="queue_welcome_email", routing_key="email.welcome", dlx_queue="queue_welcome_email.dlq"),
+        QueueSpec(name="queue_reset_email", routing_key="email.reset", dlx_queue="queue_reset_email.dlq"),
+    ]),
+])
+    
+    

src/service/email/email_reset.py

@@ -0,0 +1,37 @@
+import smtplib
+import ssl
+from email.message import EmailMessage
+
+from src.models.configs_read.env import env_settings
+from src.service.email.jinja_env import jinja_env
+
+
+class ResetEmailSender:
+
+    async def send_email(self, target_email: str, temp_password: str) -> None:
+        context = ssl.create_default_context()
+        with smtplib.SMTP_SSL(
+            env_settings.SMTP_SERVER, env_settings.EMAIL_PORT, context=context
+        ) as server:
+            server.login(env_settings.EMAIL_LOGIN, env_settings.EMAIL_PASSWORD)
+            msg = await self.build_email_message(target_email, temp_password)
+            server.send_message(msg)
+
+    async def build_email_message(self, target_email: str, temp_password: str) -> EmailMessage:
+        template = jinja_env.get_template("reset.html")
+        html_body = template.render(temp_password=temp_password)
+
+        text_body = (
+            "Пароль от вашей учётной записи в «The DisExcel» был сброшен.\n\n"
+            f"Временный пароль: {temp_password}\n\n"
+            "Рекомендуем сменить его на свой сразу после входа в аккаунт. "
+            "Если вы не запрашивали сброс пароля, срочно свяжитесь с поддержкой."
+        )
+
+        msg = EmailMessage()
+        msg["to"] = target_email
+        msg["from"] = env_settings.EMAIL_LOGIN
+        msg["subject"] = "Пароль сброшен"
+        msg.set_content(text_body)
+        msg.add_alternative(html_body, subtype="html")
+        return msg

src/service/email/email_welcome.py

@@ -0,0 +1,40 @@
+import smtplib
+import ssl
+from email.message import EmailMessage
+
+from src.models.configs_read.env import env_settings
+from src.service.email.jinja_env import jinja_env
+
+
+class DaemonEmailSender:
+    def __init__(self) -> None:
+        pass
+    
+    async def send_email(self, target_email:str)->None:
+        context = ssl.create_default_context()
+        with smtplib.SMTP_SSL(
+            env_settings.SMTP_SERVER, env_settings.EMAIL_PORT, context=context
+        ) as server:
+            server.login(env_settings.EMAIL_LOGIN, env_settings.EMAIL_PASSWORD)
+            msg=await self.build_email_message(target_email)
+            server.send_message(msg)
+            
+    async def build_email_message(self, target_email: str) -> EmailMessage:
+        template = jinja_env.get_template("welcome.html")
+        html_body = template.render(name=target_email)
+
+        text_body = (
+            f"Добро пожаловать, {target_email}!\n\n"
+            "Спасибо за регистрацию на проекте «The DisExcel». "
+            "Мы рады, что вы с нами — учётная запись уже готова к работе.\n\n"
+            "Если вы не регистрировались на The DisExcel, просто проигнорируйте это письмо."
+        )
+
+        msg = EmailMessage()
+        msg["to"] = target_email
+        msg["from"] = env_settings.EMAIL_LOGIN
+        msg["subject"] = "Добро пожаловать"
+        msg.set_content(text_body)
+        msg.add_alternative(html_body, subtype="html")
+        return msg
+

src/service/email/jinja_env.py

@@ -0,0 +1,6 @@
+from jinja2 import Environment, FileSystemLoader
+
+jinja_env = Environment(
+    loader=FileSystemLoader("src/service/email/templates"),
+    autoescape=True,
+)

src/service/email/templates/reset.html

@@ -0,0 +1,55 @@
+<!doctype html>
+<html lang="ru">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Пароль сброшен</title>
+</head>
+<body style="margin:0; padding:0; background-color:#f4f4f7; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;">
+  <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f4f4f7; padding:40px 0;">
+    <tr>
+      <td align="center">
+        <table role="presentation" width="480" cellpadding="0" cellspacing="0" style="background-color:#ffffff; border-radius:12px; overflow:hidden; box-shadow:0 2px 8px rgba(0,0,0,0.06);">
+
+          <tr>
+            <td style="background-color:#dc2626; padding:32px 40px; text-align:center;">
+              <h1 style="margin:0; color:#ffffff; font-size:22px; font-weight:600;">The DisExcel</h1>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:40px;">
+              <h2 style="margin:0 0 16px 0; color:#111827; font-size:20px;">Пароль был сброшен</h2>
+              <p style="margin:0 0 24px 0; color:#4b5563; font-size:15px; line-height:1.6;">
+                Пароль от вашей учётной записи в «The DisExcel» был сброшен. Ниже — временный пароль для входа. Рекомендуем сменить его на свой сразу после входа в аккаунт.
+              </p>
+
+              <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f9fafb; border:1px solid #e5e7eb; border-radius:8px;">
+                <tr>
+                  <td style="padding:20px; text-align:center;">
+                    <span style="display:block; margin:0 0 4px 0; color:#9ca3af; font-size:12px; text-transform:uppercase; letter-spacing:0.05em;">Новый пароль</span>
+                    <span style="display:inline-block; font-family: 'Courier New', monospace; font-size:20px; font-weight:700; color:#111827; letter-spacing:0.05em;">{{ temp_password }}</span>
+                  </td>
+                </tr>
+              </table>
+
+              <p style="margin:24px 0 0 0; color:#9ca3af; font-size:13px; line-height:1.5;">
+                Если вы не запрашивали сброс пароля, срочно свяжитесь с поддержкой — возможно, кто-то получил доступ к вашей учётной записи.
+              </p>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:24px 40px; background-color:#f9fafb; border-top:1px solid #e5e7eb;">
+              <p style="margin:0; color:#9ca3af; font-size:12px; line-height:1.5;">
+                Это автоматическое уведомление от The DisExcel. Не пересылайте это письмо никому.
+              </p>
+            </td>
+          </tr>
+
+        </table>
+      </td>
+    </tr>
+  </table>
+</body>
+</html>

src/service/email/templates/welcome.html

@@ -0,0 +1,52 @@
+<!doctype html>
+<html lang="ru">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>Добро пожаловать</title>
+</head>
+<body style="margin:0; padding:0; background-color:#f4f4f7; font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;">
+  <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background-color:#f4f4f7; padding:40px 0;">
+    <tr>
+      <td align="center">
+        <table role="presentation" width="480" cellpadding="0" cellspacing="0" style="background-color:#ffffff; border-radius:12px; overflow:hidden; box-shadow:0 2px 8px rgba(0,0,0,0.06);">
+
+          <tr>
+            <td style="background-color:#4f46e5; padding:32px 40px; text-align:center;">
+              <h1 style="margin:0; color:#ffffff; font-size:22px; font-weight:600;">The DisExcel</h1>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:40px;">
+              <h2 style="margin:0 0 16px 0; color:#111827; font-size:20px;">Добро пожаловать, {{ name }}</h2>
+              <p style="margin:0 0 24px 0; color:#4b5563; font-size:15px; line-height:1.6;">
+                Спасибо за регистрацию на проекте «The DisExcel». Мы рады, что вы с нами — учётная запись уже готова к работе.
+              </p>
+
+              <table role="presentation" cellpadding="0" cellspacing="0" align="center" style="margin:0 auto;">
+                <tr>
+                  <td style="border-radius:8px; background-color:#4f46e5;">
+                    <a href="#" style="display:inline-block; padding:12px 28px; color:#ffffff; font-size:14px; font-weight:600; text-decoration:none;">
+                      Перейти в аккаунт
+                    </a>
+                  </td>
+                </tr>
+              </table>
+            </td>
+          </tr>
+
+          <tr>
+            <td style="padding:24px 40px; background-color:#f9fafb; border-top:1px solid #e5e7eb;">
+              <p style="margin:0; color:#9ca3af; font-size:12px; line-height:1.5;">
+                Если вы не регистрировались на The DisExcel, просто проигнорируйте это письмо.
+              </p>
+            </td>
+          </tr>
+
+        </table>
+      </td>
+    </tr>
+  </table>
+</body>
+</html>

src/web/protected_routes/auth_routes.py

@@ -74,6 +74,7 @@ async def logout(response:Response,
     response.delete_cookie("refresh_token")
     return await auth.logout(refresh_token, access_token)
 
+ 
 @router.get("")
 async def protected(current_user:UserOut=Depends(require_permissions()))->dict:
     return {"protected router": "Hello, this is a protected router"}

src/web/protected_routes/protected_user_action_routes.py

@@ -1,5 +1,6 @@
 from fastapi import APIRouter, Depends
 
+from src.messaging.producers.producers import email_producer
 from src.models.pydantic_models.model import UserCreate, UserOut, UserUpdate
 from src.service.users_crud.users_crud import CrudService, crud_service
 from src.web.protected_routes.auth_routes import require_permissions
@@ -12,6 +13,7 @@ async def get_current_user_by_email(email:str, crud:CrudService=Depends(crud_ser
 
 @router.post("/create_user")
 async def create_user(data:UserCreate, crud:CrudService=Depends(crud_service), current_user=Depends(require_permissions("admin")))->UserOut:
+    await email_producer.send_welcome_email(current_user.email)  
     return await crud.create_user(data)
 
 @router.post("/delete_user_soft")

tests/e2e/test_auth.py

@@ -34,16 +34,18 @@ class TestPermissions:
         assert exc_info.value.response.status_code != 403
         assert exc_info.value.response.status_code == 401
     
+    
     @pytest.mark.parametrize("test_user_fixture", [([], [])], indirect=True)
     async def test_get_logout_positive(self, test_user_fixture, target_url: str) -> None:
         
         session = test_user_fixture[0]
-        with allure.step("get_root"), pytest.raises(HTTPStatusError) as exc_info:
+        with allure.step("get_root"):
             
             response = await session.get(f"{target_url}/protected/logout")
             response.raise_for_status()
-        assert exc_info.value.response.status_code != 403
+
-        assert exc_info.value.response.status_code != 401
+        assert response.status_code == 200
+
 
 
 class TestRedis:

tests/unit/test_consumers.py

@@ -0,0 +1,234 @@
+import asyncio
+import json
+import smtplib
+from unittest.mock import AsyncMock, MagicMock
+
+import allure
+import pytest
+
+import src.messaging.rabbitmq_client as rabbitmq_client_module
+from src.messaging.consumers.consumers import ResetEmailConsumer, WelcomeEmailConsumer
+from src.messaging.rabbitmq_client import RabbitMQClient
+
+
+@pytest.mark.unit
+class TestRabbitMQClient:
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="Connect_retries_then_succeeds")
+    ])
+    async def test_connect_retries_then_succeeds(self, monkeypatch):
+        
+        with allure.step("Mocking connection to the rabbitmq"):
+            fake_connection = MagicMock()
+            fake_connection.is_closed = False
+            fake_connection.channel = AsyncMock(return_value=MagicMock())
+            
+            mock_connect = AsyncMock(
+                side_effect=[ConnectionError(), ConnectionError(), fake_connection]
+            )
+            monkeypatch.setattr(rabbitmq_client_module.aio_pika, "connect_robust", mock_connect)
+            monkeypatch.setattr(asyncio, "sleep", AsyncMock())  # чтобы не ждать реально
+
+        with allure.step("Test client connection"):
+            client = RabbitMQClient()
+            await client.connect()
+
+            assert mock_connect.await_count == 3
+            assert client.connection is fake_connection
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_connect_gives_up_after_5_attempts")
+    ])
+    async def test_connect_gives_up_after_5_attempts(self, monkeypatch):
+        
+        with allure.step("Mocking connection to the rabbitmq"):
+            mock_connect = AsyncMock(side_effect=ConnectionError("still down"))
+            monkeypatch.setattr(rabbitmq_client_module.aio_pika, "connect_robust", mock_connect)
+            monkeypatch.setattr(asyncio, "sleep", AsyncMock())
+
+        with allure.step("Test client connection"):
+            client = RabbitMQClient()
+
+        with pytest.raises(ConnectionError):
+            await client.connect()
+
+        assert mock_connect.await_count == 5
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_get_channel_negative")
+    ])
+    async def test_get_channel_negative(self, monkeypatch)->None:
+        
+        with allure.step("Mocking connection to the rabbitmq"):
+            fake_connection = MagicMock()
+            fake_connection.is_closed = False
+            fake_connection.channel = AsyncMock(return_value=None)
+        
+            mock_connect = AsyncMock(return_value=fake_connection)
+            monkeypatch.setattr(rabbitmq_client_module.aio_pika, "connect_robust", mock_connect)
+            
+        with allure.step("Test client connection"):
+            client = RabbitMQClient()
+            
+        with pytest.raises(RuntimeError):
+            await client.get_channel()
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_get_channel_positive")
+    ])
+    async def test_get_channel_positive(self, monkeypatch)->None:
+        
+        with allure.step("Mocking connection to the rabbitmq"):
+            
+            fake_connection = MagicMock()
+            fake_connection.is_closed = False
+            fake_connection.channel = AsyncMock(return_value=MagicMock())
+            
+            mock_connect = AsyncMock(return_value=fake_connection)
+            monkeypatch.setattr(rabbitmq_client_module.aio_pika, "connect_robust", mock_connect)
+            
+        with allure.step("Test client connection"):
+            client = RabbitMQClient()
+            await client.get_channel()
+        
+        assert client.channel is fake_connection.channel.return_value
+    
+def make_fake_message(body: dict, routing_key: str = "email.welcome") -> MagicMock:
+    """Строим 'фальшивое' RabbitMQ-сообщение вручную, без реального брокера."""
+    message = MagicMock()
+    message.body = json.dumps(body).encode()
+    message.routing_key = routing_key
+    message.nack = AsyncMock()  # запоминает, вызвали ли nack и с какими аргументами
+
+    # message.process(...) в реальном aio_pika — async context manager,
+    # поэтому нужен объект с асинхронными __aenter__/__aexit__.
+    process_cm = MagicMock()
+    process_cm.__aenter__ = AsyncMock(return_value=None)
+    process_cm.__aexit__ = AsyncMock(return_value=False)
+    message.process = MagicMock(return_value=process_cm)
+    return message
+
+
+@pytest.mark.unit
+class TestEmailConsumers:
+    
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_start_consuming_negative")
+    ])
+    async def test_start_consuming_negative(self, monkeypatch)->None:
+        
+        with allure.step("Mocking connection to the rabbitmq"):
+            consumer = WelcomeEmailConsumer()
+            consumer.setup=AsyncMock()
+            
+        with allure.step("Test client connection"), pytest.raises(RuntimeError):
+            await consumer.start_consuming()
+        
+        
+    @pytest.mark.parametrize("error", [
+        pytest.param(smtplib.SMTPServerDisconnected(), id="smtplib.SMTPServerDisconnected"),
+        pytest.param(smtplib.SMTPConnectError(421, "Service not available"), id="smtplib.SMTPConnectError"),
+        pytest.param(TimeoutError(), id="TimeoutError"),
+        pytest.param(ConnectionRefusedError(), id="ConnectionRefusedError"),
+    ])
+    async def test_process_message_transient_errors_requeue(self, error) -> None:
+        consumer = WelcomeEmailConsumer()
+        consumer.daemon.send_email = AsyncMock(side_effect=error)
+
+        message = make_fake_message({"email": "user@example.com"})
+        await consumer.process_message(message)
+
+        message.nack.assert_awaited_once_with(requeue=True)
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_process_message_positive")
+    ])
+    async def test_process_message_positive(self) -> None:
+
+        with allure.step("Mocking dependencies"):
+            consumer = WelcomeEmailConsumer()
+            consumer.daemon.send_email = AsyncMock()
+
+        with allure.step("Test process_message"):
+            message = make_fake_message({"email": "user@example.com"})
+            await consumer.process_message(message)
+
+        consumer.daemon.send_email.assert_awaited_once_with("user@example.com")
+        message.nack.assert_not_called()
+
+
+@pytest.mark.unit
+class TestResetEmailConsumer:
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_reset_start_consuming_negative")
+    ])
+    async def test_start_consuming_negative(self, monkeypatch) -> None:
+
+        with allure.step("Mocking connection to the rabbitmq"):
+            consumer = ResetEmailConsumer()
+            consumer.setup = AsyncMock()
+
+        with allure.step("Test client connection"), pytest.raises(RuntimeError):
+            await consumer.start_consuming()
+
+
+    @pytest.mark.parametrize("error", [
+        pytest.param(smtplib.SMTPServerDisconnected(), id="smtplib.SMTPServerDisconnected"),
+        pytest.param(smtplib.SMTPConnectError(421, "Service not available"), id="smtplib.SMTPConnectError"),
+        pytest.param(TimeoutError(), id="TimeoutError"),
+        pytest.param(ConnectionRefusedError(), id="ConnectionRefusedError"),
+    ])
+    async def test_process_message_transient_errors_requeue(self, error) -> None:
+        consumer = ResetEmailConsumer()
+        consumer.daemon.send_email = AsyncMock(side_effect=error)
+
+        message = make_fake_message(
+            {"email": "user@example.com", "temp_password": "Temp1234!"},
+            routing_key="email.reset",
+        )
+        await consumer.process_message(message)
+
+        message.nack.assert_awaited_once_with(requeue=True)
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_reset_process_message_permanent_error_goes_to_dlq")
+    ])
+    async def test_process_message_permanent_error_goes_to_dlq(self) -> None:
+        consumer = ResetEmailConsumer()
+        consumer.daemon.send_email = AsyncMock(side_effect=ValueError("bad payload"))
+
+        message = make_fake_message(
+            {"email": "user@example.com", "temp_password": "Temp1234!"},
+            routing_key="email.reset",
+        )
+        await consumer.process_message(message)
+
+        message.nack.assert_awaited_once_with(requeue=False)
+
+
+    @pytest.mark.parametrize("",[
+        pytest.param(id="test_reset_process_message_positive")
+    ])
+    async def test_process_message_positive(self) -> None:
+
+        with allure.step("Mocking dependencies"):
+            consumer = ResetEmailConsumer()
+            consumer.daemon.send_email = AsyncMock()
+
+        with allure.step("Test process_message"):
+            message = make_fake_message(
+                {"email": "user@example.com", "temp_password": "Temp1234!"},
+                routing_key="email.reset",
+            )
+            await consumer.process_message(message)
+
+        consumer.daemon.send_email.assert_awaited_once_with("user@example.com", "Temp1234!")
+        message.nack.assert_not_called()
+