CLAUDE.md

@@ -38,32 +38,123 @@ workers. Python >=3.13,<4.0, Poetry for dependency management.
 
 - `RabbitMQClient` — shared class, lazy `connect()` (can't be async `__init__`),
   holds one `connection` + one `channel`, `get_channel()` ensures setup.
-- Topic exchange named `"email"`. Each message type gets its own routing key
-  (`email.welcome`, `email.reset`) and its own durable queue
-  (`queue_welcome_email`, `queue_reset_email`), each queue explicitly bound
-  to the exchange with its own key.
-- **Important**: both producer and consumer must declare/bind the queues —
-  if only the consumer does it and the consumer has never run, `publish`
-  on a not-yet-existing queue silently loses the message. `EmailProducer.setup()`
-  also declares+binds both queues defensively.
-- `message.process()` async context manager = manual ack (auto-ack on
-  success, requeue/nack on exception) — this is the right choice for email,
-  not `no_ack=True`.
-- Email templates: Jinja2, inline CSS (email clients don't support `<style>`
-  reliably), `EmailMessage` with `set_content()` (plain-text fallback) +
-  `add_alternative(html, subtype="html")`.
+  `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")`.
+- **Known gap**: the reset-password flow is a stub. `ResetEmailConsumer.process_message`
+  only prints and acks — it never calls a sender — and
+  `src/service/email/email_reset.py` is empty. `templates/reset.html` still
+  has a hardcoded placeholder password. Don't assume reset emails actually
+  send until this is wired up like `WelcomeEmailConsumer`/`DaemonEmailSender`.
+
+## 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.
-- `run_daemon.py` (project root) is the single entrypoint: `python
-  run_daemon.py <name>` runs one daemon, `python run_daemon.py --all` reads
+- `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`.
+  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",
-  "run_daemon.py", "<name>"]`), same pattern as the `migration` service.
+  "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
@@ -76,7 +167,10 @@ workers. Python >=3.13,<4.0, Poetry for dependency management.
   psycopg2-binary, alembic, greenlet).
 - `web` group: fastapi, uvicorn, gunicorn, python-multipart — only needed
   by the API server.
-- `daemon` group: worker-only deps (currently empty, grows as needed).
+- `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.
 - Dockerfile has parallel builder→final stage pairs: `builder`→`prod`
   (installs `main,web`) and `worker-builder`→`worker` (installs

src/messaging/rabbitmq_client.py

@@ -1,4 +1,5 @@
 import asyncio
+
 import aio_pika
 from aio_pika.abc import AbstractChannel, AbstractRobustConnection