diff --git a/.env.example b/.env.example
new file mode 100644
index 0000000..b98e55a
--- /dev/null
+++ b/.env.example
@@ -0,0 +1,14 @@
+# Copy to .env if you want to override defaults for a local, non-Compose run.
+# Inside Compose, these are set directly in docker-compose.yml per service.
+
+POSTGRES_DSN=postgresql+asyncpg://ledgerline:ledgerline@localhost:5432/ledgerline
+REDIS_URL=redis://localhost:6379/0
+
+GATEWAY_URL=http://localhost:8000
+LEDGER_URL=http://localhost:8001
+FRAUD_URL=http://localhost:8002
+SANCTIONS_URL=http://localhost:8003
+INSIGHTS_URL=http://localhost:8004
+NOTIFIER_URL=http://localhost:8005
+
+SEED=42
diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..c6d3959
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1 \
+ PYTHONDONTWRITEBYTECODE=1 \
+ PIP_NO_CACHE_DIR=1
+
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends curl \
+ && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install -r requirements.txt
+
+COPY common ./common
+COPY services ./services
+COPY db ./db
+COPY scripts ./scripts
+
+# CMD differs per service -- see docker-compose.yml.
+CMD ["python", "-c", "print('no default command: set one in docker-compose.yml')"]
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..90d0cfa
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,70 @@
+# Ledgerline -- application targets (phase-0-base-project-spec.md §12) plus
+# the Phase 0a test-suite targets. `make up && make seed` are prerequisites
+# for every test target below.
+
+PYTEST ?= pytest
+COMPOSE ?= docker compose
+SEED_DSN ?= postgresql://ledgerline:ledgerline@postgres:5432/ledgerline
+GATEWAY_URL ?= http://localhost:8000
+
+.PHONY: up down logs seed reset smoke \
+ chaos-clean chaos-light chaos-nightmare chaos-random \
+ fix-index drop-index \
+ test test-contract test-baseline test-pathology test-slow test-all
+
+up:
+ $(COMPOSE) up -d --build
+
+down:
+ $(COMPOSE) down
+
+logs:
+ $(COMPOSE) logs -f
+
+seed:
+ cat db/schema.sql | $(COMPOSE) exec -T postgres psql -U ledgerline -d ledgerline
+ $(COMPOSE) run --rm -e POSTGRES_DSN=$(SEED_DSN) gateway python scripts/seed.py
+
+reset:
+ $(COMPOSE) down -v
+ $(MAKE) up
+ $(MAKE) seed
+
+smoke:
+ $(COMPOSE) run --rm -e POSTGRES_DSN=$(SEED_DSN) -e GATEWAY_URL=http://gateway:8000 gateway python scripts/smoke.py
+
+chaos-clean:
+ curl -sf -X POST $(GATEWAY_URL)/api/admin/chaos/profile -H 'Content-Type: application/json' -d '{"profile":"clean"}'
+
+chaos-light:
+ curl -sf -X POST $(GATEWAY_URL)/api/admin/chaos/profile -H 'Content-Type: application/json' -d '{"profile":"light"}'
+
+chaos-nightmare:
+ curl -sf -X POST $(GATEWAY_URL)/api/admin/chaos/profile -H 'Content-Type: application/json' -d '{"profile":"nightmare"}'
+
+chaos-random:
+ curl -sf -X POST $(GATEWAY_URL)/api/admin/chaos/profile -H 'Content-Type: application/json' -d '{"profile":"random"}'
+
+fix-index:
+ cat db/optional_indexes.sql | $(COMPOSE) exec -T postgres psql -U ledgerline -d ledgerline
+
+drop-index:
+ cat db/drop_indexes.sql | $(COMPOSE) exec -T postgres psql -U ledgerline -d ledgerline
+
+test:
+ $(PYTEST) -m "not slow"
+
+test-contract:
+ $(PYTEST) -m "contract or seed or guard"
+
+test-baseline:
+ $(PYTEST) -m baseline
+
+test-pathology:
+ $(PYTEST) -m "pathology and not slow"
+
+test-slow:
+ $(PYTEST) -m slow
+
+test-all:
+ $(PYTEST)
diff --git a/common/__init__.py b/common/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/common/chaos_flags.py b/common/chaos_flags.py
new file mode 100644
index 0000000..a0e5db3
--- /dev/null
+++ b/common/chaos_flags.py
@@ -0,0 +1,75 @@
+"""Shared low-level access to the `chaos:flags` Redis hash.
+
+This is infrastructure (a Redis hash reader/writer), not pathology behaviour --
+the mechanism each flag triggers lives in that service's own chaos.py, per
+phase-0-base-project-spec.md §7 ("each pathology lives in a clearly named
+chaos.py module inside its owning service"). Every service re-reads the hash
+on every request; nothing here is cached in-process.
+"""
+import random
+import time
+
+from .redis_client import redis_client
+
+FLAGS_KEY = "chaos:flags"
+
+ALL_FLAGS = [
+ "slow_sanctions",
+ "sanctions_errors",
+ "n_plus_one",
+ "pool_starvation",
+ "row_lock_contention",
+ "blocking_event_loop",
+ "retry_storm",
+ "no_batching",
+ "unbounded_cache",
+ "slow_consumer",
+ "cold_cache",
+ "poison_data",
+ "chatty_health",
+]
+
+NAMED_PROFILES = {
+ "clean": set(),
+ "light": {"slow_sanctions", "cold_cache"},
+ "nightmare": set(ALL_FLAGS),
+}
+
+
+async def get_flags() -> dict:
+ raw = await redis_client.hgetall(FLAGS_KEY)
+ return {flag: raw.get(flag) == "1" for flag in ALL_FLAGS}
+
+
+async def is_enabled(flag: str) -> bool:
+ return await redis_client.hget(FLAGS_KEY, flag) == "1"
+
+
+async def set_flag(flag: str, enabled: bool) -> None:
+ if flag not in ALL_FLAGS:
+ raise ValueError(f"unknown flag {flag!r}")
+ await redis_client.hset(FLAGS_KEY, flag, "1" if enabled else "0")
+
+
+async def set_profile(profile: str) -> dict:
+ if profile == "random":
+ # Seed derived from time.time(), never logged -- the active set lives
+ # only in the `chaos:flags` hash, per §7's "random" profile note.
+ rng = random.Random(time.time())
+ chosen = set(rng.sample(ALL_FLAGS, k=rng.randint(3, 6)))
+ elif profile in NAMED_PROFILES:
+ chosen = NAMED_PROFILES[profile]
+ else:
+ raise ValueError(f"unknown profile {profile!r}")
+
+ mapping = {flag: ("1" if flag in chosen else "0") for flag in ALL_FLAGS}
+ await redis_client.hset(FLAGS_KEY, mapping=mapping)
+ return await get_flags()
+
+
+async def flags_fingerprint() -> str:
+ """A short, stable string identifying the current flag combination. Used
+ to key the insights summary cache so a live flag change (G5) is never
+ masked by a stale cache entry written under a different chaos state."""
+ flags = await get_flags()
+ return "".join("1" if flags[f] else "0" for f in ALL_FLAGS)
diff --git a/common/db.py b/common/db.py
new file mode 100644
index 0000000..2864869
--- /dev/null
+++ b/common/db.py
@@ -0,0 +1,15 @@
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from .settings import settings
+
+
+def make_engine(**overrides):
+ """A fresh async engine. Services that need more than one pool profile
+ (see ledger/chaos.py for pool_starvation) call this more than once."""
+ kwargs = dict(pool_pre_ping=True)
+ kwargs.update(overrides)
+ return create_async_engine(settings.postgres_dsn, **kwargs)
+
+
+engine = make_engine(pool_size=20, max_overflow=10)
+Session = async_sessionmaker(engine, expire_on_commit=False)
diff --git a/common/http.py b/common/http.py
new file mode 100644
index 0000000..f708651
--- /dev/null
+++ b/common/http.py
@@ -0,0 +1,13 @@
+import httpx
+
+from .settings import settings
+
+
+def downstream_client(request_id: str) -> httpx.AsyncClient:
+ """An httpx.AsyncClient preconfigured with the 3s downstream timeout and
+ the X-Request-Id header forwarded, per phase-0-base-project-spec.md §9 --
+ this header is the only correlation mechanism that exists in Phase 0."""
+ return httpx.AsyncClient(
+ timeout=settings.downstream_timeout_seconds,
+ headers={"X-Request-Id": request_id},
+ )
diff --git a/common/logging_config.py b/common/logging_config.py
new file mode 100644
index 0000000..abc639c
--- /dev/null
+++ b/common/logging_config.py
@@ -0,0 +1,12 @@
+import logging
+
+
+def configure_logging(service_name: str) -> logging.Logger:
+ """Plain-text, human-prose logging -- the deliberately weak Phase 0
+ baseline. Per phase-0-base-project-spec.md §9: no JSON, no key-value, no
+ structured fields. Do not "improve" this; later phases replace it."""
+ logging.basicConfig(
+ level=logging.INFO,
+ format="%(asctime)s %(levelname)s %(name)s %(message)s",
+ )
+ return logging.getLogger(service_name)
diff --git a/common/redis_client.py b/common/redis_client.py
new file mode 100644
index 0000000..937c45f
--- /dev/null
+++ b/common/redis_client.py
@@ -0,0 +1,5 @@
+import redis.asyncio as redis
+
+from .settings import settings
+
+redis_client = redis.from_url(settings.redis_url, decode_responses=True)
diff --git a/common/settings.py b/common/settings.py
new file mode 100644
index 0000000..f594d8e
--- /dev/null
+++ b/common/settings.py
@@ -0,0 +1,27 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+ model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+
+ service_name: str = "unknown"
+
+ postgres_dsn: str = "postgresql+asyncpg://ledgerline:ledgerline@postgres:5432/ledgerline"
+ redis_url: str = "redis://redis:6379/0"
+
+ gateway_url: str = "http://gateway:8000"
+ ledger_url: str = "http://ledger:8001"
+ fraud_url: str = "http://fraud:8002"
+ sanctions_url: str = "http://sanctions:8003"
+ insights_url: str = "http://insights:8004"
+ notifier_url: str = "http://notifier:8005"
+
+ downstream_timeout_seconds: float = 3.0
+
+ stream_name: str = "stream:transactions"
+
+ summary_cache_ttl_seconds: int = 900
+ summary_cache_ttl_seconds_cold: int = 60
+
+
+settings = Settings()
diff --git a/db/drop_indexes.sql b/db/drop_indexes.sql
new file mode 100644
index 0000000..a0aeaeb
--- /dev/null
+++ b/db/drop_indexes.sql
@@ -0,0 +1,4 @@
+-- Applied only by `make drop-index`. Reverses db/optional_indexes.sql.
+
+DROP INDEX IF EXISTS transactions_account_created_idx;
+DROP INDEX IF EXISTS transactions_category_idx;
diff --git a/db/optional_indexes.sql b/db/optional_indexes.sql
new file mode 100644
index 0000000..5019f95
--- /dev/null
+++ b/db/optional_indexes.sql
@@ -0,0 +1,5 @@
+-- Applied only by `make fix-index` (docs/phase-0-base-project-spec.md §5, §12).
+-- These are the indexes deliberately omitted from db/schema.sql.
+
+CREATE INDEX IF NOT EXISTS transactions_account_created_idx ON transactions (account_id, created_at);
+CREATE INDEX IF NOT EXISTS transactions_category_idx ON transactions (category_id);
diff --git a/db/schema.sql b/db/schema.sql
new file mode 100644
index 0000000..d56b110
--- /dev/null
+++ b/db/schema.sql
@@ -0,0 +1,80 @@
+-- Ledgerline schema. Idempotent: safe to run against an existing database.
+-- Index policy is deliberate -- see docs/phase-0-base-project-spec.md §5.
+-- Do NOT add transactions(account_id, created_at) or transactions(category_id)
+-- here; those live in db/optional_indexes.sql, applied only by `make fix-index`.
+
+CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
+
+CREATE TABLE IF NOT EXISTS accounts (
+ id UUID PRIMARY KEY,
+ holder_name TEXT NOT NULL,
+ currency CHAR(3) NOT NULL,
+ balance_minor BIGINT NOT NULL,
+ is_hot BOOLEAN NOT NULL DEFAULT FALSE,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS merchant_categories (
+ id SMALLINT PRIMARY KEY,
+ code TEXT NOT NULL UNIQUE,
+ label TEXT NOT NULL,
+ risk_weight NUMERIC(3,2) NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS transactions (
+ id UUID PRIMARY KEY,
+ account_id UUID NOT NULL,
+ kind TEXT NOT NULL,
+ amount_minor BIGINT NOT NULL,
+ currency CHAR(3),
+ merchant_name TEXT,
+ category_id SMALLINT,
+ status TEXT NOT NULL,
+ risk_score NUMERIC(4,3),
+ created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Parallel workers otherwise let a plain sequential scan of this table
+-- compete with an index scan on raw CPU parallelism for higher-selectivity
+-- accounts (e.g. the 5 hot ones, each ~6-8% of all rows) -- masking exactly
+-- the missing-index pathology this table's index policy (README §5) exists
+-- to demonstrate. Scoped to this one table, not a database-wide setting.
+ALTER TABLE transactions SET (parallel_workers = 0);
+
+CREATE TABLE IF NOT EXISTS sanctions_entries (
+ id SERIAL PRIMARY KEY,
+ normalized_name TEXT NOT NULL,
+ country CHAR(2),
+ list_source TEXT,
+ added_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE IF NOT EXISTS notifications (
+ id UUID PRIMARY KEY,
+ transaction_id UUID NOT NULL,
+ channel TEXT NOT NULL,
+ status TEXT NOT NULL,
+ attempts INT NOT NULL DEFAULT 0,
+ sent_at TIMESTAMPTZ
+);
+
+CREATE TABLE IF NOT EXISTS audit_events (
+ id BIGSERIAL PRIMARY KEY,
+ transaction_id UUID NOT NULL,
+ event_type TEXT NOT NULL,
+ payload JSONB,
+ recorded_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+-- Not part of the base spec's domain model: a single-row table the seed
+-- script writes its summary manifest into, so both `scripts/smoke.py` and
+-- the test suite can read back what a given seed run actually produced
+-- (row counts, hot/sanctioned/poison account ids, the SEED value used)
+-- without re-deriving it via ad-hoc queries.
+CREATE TABLE IF NOT EXISTS seed_manifest (
+ id SMALLINT PRIMARY KEY DEFAULT 1,
+ data JSONB NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS accounts_holder_name_idx ON accounts (holder_name);
+CREATE INDEX IF NOT EXISTS sanctions_entries_normalized_name_idx ON sanctions_entries (normalized_name);
diff --git a/docker-compose.yml b/docker-compose.yml
new file mode 100644
index 0000000..2ba5ab1
--- /dev/null
+++ b/docker-compose.yml
@@ -0,0 +1,205 @@
+x-app-env: &app-env
+ POSTGRES_DSN: postgresql+asyncpg://ledgerline:ledgerline@postgres:5432/ledgerline
+ REDIS_URL: redis://redis:6379/0
+ GATEWAY_URL: http://gateway:8000
+ LEDGER_URL: http://ledger:8001
+ FRAUD_URL: http://fraud:8002
+ SANCTIONS_URL: http://sanctions:8003
+ INSIGHTS_URL: http://insights:8004
+ NOTIFIER_URL: http://notifier:8005
+
+x-app-build: &app-build
+ build:
+ context: .
+ dockerfile: Dockerfile
+
+x-app-volumes: &app-volumes
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+
+x-app-depends: &app-depends
+ depends_on:
+ postgres:
+ condition: service_healthy
+ redis:
+ condition: service_healthy
+
+services:
+ postgres:
+ image: postgres:16-alpine
+ # max_connections=300: five separate services each keep their own
+ # SQLAlchemy pool (several sized up to pool_size=20+max_overflow=10),
+ # all against this one Postgres instance -- comfortably clears the
+ # default 100 under concurrent load instead of silently rationing
+ # connections in a way that looks like an unrelated pathology.
+ command:
+ - postgres
+ - -c
+ - shared_preload_libraries=pg_stat_statements
+ - -c
+ - pg_stat_statements.track=all
+ - -c
+ - pg_stat_statements.max=10000
+ - -c
+ - max_connections=300
+ environment:
+ POSTGRES_USER: ledgerline
+ POSTGRES_PASSWORD: ledgerline
+ POSTGRES_DB: ledgerline
+ ports:
+ - "5432:5432"
+ healthcheck:
+ test: ["CMD-SHELL", "pg_isready -U ledgerline -d ledgerline"]
+ interval: 3s
+ timeout: 3s
+ retries: 20
+ volumes:
+ - pgdata:/var/lib/postgresql/data
+
+ redis:
+ image: redis:7-alpine
+ ports:
+ - "6379:6379"
+ healthcheck:
+ test: ["CMD", "redis-cli", "ping"]
+ interval: 3s
+ timeout: 3s
+ retries: 20
+
+ gateway:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: gateway
+ command: uvicorn services.gateway.app:app --host 0.0.0.0 --port 8000 --reload --no-access-log
+ ports:
+ - "8000:8000"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8000/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ ledger:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: ledger
+ command: uvicorn services.ledger.app:app --host 0.0.0.0 --port 8001 --reload --no-access-log
+ ports:
+ - "8001:8001"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8001/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ fraud:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: fraud
+ command: uvicorn services.fraud.app:app --host 0.0.0.0 --port 8002 --reload --no-access-log
+ ports:
+ - "8002:8002"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8002/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ sanctions:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: sanctions
+ command: uvicorn services.sanctions.app:app --host 0.0.0.0 --port 8003 --reload --no-access-log
+ ports:
+ - "8003:8003"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8003/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ insights:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: insights
+ command: uvicorn services.insights.app:app --host 0.0.0.0 --port 8004 --reload --no-access-log
+ ports:
+ - "8004:8004"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8004/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ notifier:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: notifier
+ command: uvicorn services.notifier.app:app --host 0.0.0.0 --port 8005 --reload --no-access-log
+ ports:
+ - "8005:8005"
+ healthcheck:
+ test: ["CMD", "curl", "-f", "http://localhost:8005/healthz"]
+ interval: 5s
+ timeout: 3s
+ retries: 10
+ start_period: 10s
+
+ audit:
+ <<: [*app-build, *app-depends]
+ volumes:
+ - ./common:/app/common
+ - ./services:/app/services
+ - ./db:/app/db
+ - ./scripts:/app/scripts
+ environment:
+ <<: *app-env
+ SERVICE_NAME: audit
+ command: python -m services.audit.worker
+
+volumes:
+ pgdata:
diff --git a/observability/.gitkeep b/observability/.gitkeep
new file mode 100644
index 0000000..e69de29
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..4c12824
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,10 @@
+fastapi==0.115.6
+uvicorn[standard]==0.34.0
+httpx==0.28.1
+sqlalchemy[asyncio]==2.0.36
+asyncpg==0.30.0
+psycopg[binary]==3.2.3
+redis==5.2.1
+pydantic==2.10.4
+pydantic-settings==2.7.1
+python-multipart==0.0.20
diff --git a/scripts/seed.py b/scripts/seed.py
new file mode 100644
index 0000000..8a9a350
--- /dev/null
+++ b/scripts/seed.py
@@ -0,0 +1,375 @@
+#!/usr/bin/env python3
+"""Deterministic seed data for Ledgerline -- phase-0-base-project-spec.md §8.
+
+Idempotent: truncates then inserts. Deterministic given SEED (default 42).
+Uses COPY for the transactions table so 250k rows load in well under 60s.
+"""
+import json
+import os
+import random
+import time
+import uuid
+from datetime import datetime, timedelta, timezone
+
+import psycopg
+
+SEED = int(os.environ.get("SEED", "42"))
+DSN = os.environ.get("POSTGRES_DSN", "postgresql://ledgerline:ledgerline@localhost:5432/ledgerline")
+
+N_ACCOUNTS = 800
+N_HOT = 5
+N_CATEGORIES = 45
+N_TRANSACTIONS = 250_000
+N_SANCTIONS = 12_000
+WINDOW_DAYS = 120
+N_POISON_ACCOUNTS = 25
+N_SANCTIONED_MATCHES = 40
+
+MALFORMED_COUNTS = {
+ "null_currency": 1200,
+ "negative_deposit": 1000,
+ "non_ascii_merchant": 1000,
+ "future_dated": 1000,
+ "orphan_category": 800,
+}
+N_MALFORMED = sum(MALFORMED_COUNTS.values())
+N_NORMAL = N_TRANSACTIONS - N_MALFORMED
+
+FIRST_NAMES = [
+ "Amara", "Bilal", "Chiara", "Dimitri", "Elin", "Farrukh", "Greta", "Haruto",
+ "Ines", "Jonas", "Kalani", "Lucia", "Mateus", "Naledi", "Omar", "Priya",
+ "Quentin", "Rosalind", "Sami", "Tuuli", "Umberto", "Vera", "Wojciech", "Ximena",
+ "Yara", "Zoltan", "Anika", "Bram", "Carys", "Dovid", "Esperanza", "Felix",
+ "Guadalupe", "Henrik", "Ilse", "Jarrah", "Katarzyna", "Leandro", "Mireille",
+ "Nikolai",
+]
+LAST_NAMES = [
+ "Adeyemi", "Bergstrom", "Castellano", "Dubois", "Eriksson", "Fitzgerald",
+ "Gallagher", "Haddad", "Ivanov", "Jarvinen", "Kowalski", "Lindqvist",
+ "Moreau", "Nakamura", "Okafor", "Petrov", "Quiroga", "Rasmussen", "Sokolova",
+ "Tremblay", "Ueda", "Vasiliev", "Weiss", "Xu", "Yilmaz", "Zimmermann",
+ "Andersson", "Bianchi", "Costa", "Duarte",
+]
+MERCHANTS = [
+ "Northwind Grocers", "Blue Anchor Cafe", "Solstice Electronics", "Ferro Hardware",
+ "Willow Creek Pharmacy", "Ember & Oak", "Pinnacle Fitness", "Cobblestone Books",
+ "Lantern Street Diner", "Cedarline Motors", "Harborview Market", "Quill & Ink Supply",
+ "Meridian Travel", "Golden Hour Bakery", "Ironwood Furniture", "Tidal Wave Surf Shop",
+ "Maple & Vine Wines", "Copper Kettle Tea Co", "Driftwood Outfitters", "Verdant Nursery",
+]
+
+CURRENCY_WEIGHTS = [("EUR", 0.70), ("USD", 0.20), ("GBP", 0.10)]
+
+
+def weighted_choice(rng, pairs):
+ values, weights = zip(*pairs)
+ return rng.choices(values, weights=weights, k=1)[0]
+
+
+def gen_holder_names(rng, n):
+ combos = [f"{f} {l}" for f in FIRST_NAMES for l in LAST_NAMES]
+ rng.shuffle(combos)
+ if n > len(combos):
+ raise ValueError("not enough name combinations for the requested account count")
+ return combos[:n]
+
+
+def gen_accounts(rng):
+ names = gen_holder_names(rng, N_ACCOUNTS)
+ hot_indexes = set(rng.sample(range(N_ACCOUNTS), N_HOT))
+ now = datetime.now(timezone.utc)
+ rows = []
+ for i, name in enumerate(names):
+ account_id = str(uuid.uuid4())
+ currency = weighted_choice(rng, CURRENCY_WEIGHTS)
+ balance = int(rng.lognormvariate(_median_to_mu(250_000), 0.7))
+ created_at = now - timedelta(days=rng.uniform(150, 400))
+ rows.append(
+ {
+ "id": account_id,
+ "holder_name": name,
+ "currency": currency,
+ "balance_minor": balance,
+ "is_hot": i in hot_indexes,
+ "created_at": created_at,
+ }
+ )
+ return rows
+
+
+def _median_to_mu(median):
+ import math
+ return math.log(median)
+
+
+def gen_categories(rng):
+ rows = []
+ for i in range(1, N_CATEGORIES + 1):
+ rows.append(
+ {
+ "id": i,
+ "code": f"cat_{i:02d}",
+ "label": f"Category {i:02d}",
+ "risk_weight": round(0.01 + (i - 1) / (N_CATEGORIES - 1) * 0.94, 2),
+ }
+ )
+ return rows
+
+
+def account_weights(rng, accounts):
+ """5 hot accounts own ~30% of rows; the remaining 795 share the rest on a
+ Zipf-ish distribution."""
+ hot = [a for a in accounts if a["is_hot"]]
+ cold = [a for a in accounts if not a["is_hot"]]
+ hot_weight_each = 0.30 / len(hot)
+ ranks = list(range(1, len(cold) + 1))
+ rng.shuffle(ranks)
+ zipf_raw = [1.0 / (r ** 1.1) for r in ranks]
+ zipf_total = sum(zipf_raw)
+ cold_weights = [0.70 * w / zipf_total for w in zipf_raw]
+
+ ids = [a["id"] for a in hot] + [a["id"] for a in cold]
+ weights = [hot_weight_each] * len(hot) + cold_weights
+ return ids, weights
+
+
+def sample_created_at(rng, now, window_days=WINDOW_DAYS):
+ day_offset = rng.uniform(0, window_days)
+ day = now - timedelta(days=day_offset)
+ hour = min(23, max(0, int(rng.gauss(13, 3.5))))
+ minute, second = rng.randint(0, 59), rng.randint(0, 59)
+ dt = day.replace(hour=hour, minute=minute, second=second, microsecond=0)
+ if dt.weekday() >= 5 and rng.random() < 0.5:
+ dt -= timedelta(days=2)
+ return dt
+
+
+def gen_normal_transactions(rng, accounts, now):
+ ids, weights = account_weights(rng, accounts)
+ statuses = [("settled", 0.92), ("rejected", 0.05), ("pending", 0.03)]
+ kinds = [("deposit", 0.2), ("withdrawal", 0.2), ("purchase", 0.6)]
+
+ rows = []
+ for _ in range(N_NORMAL):
+ account_id = rng.choices(ids, weights=weights, k=1)[0]
+ kind = weighted_choice(rng, kinds)
+ status = weighted_choice(rng, statuses)
+ amount = max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9)))
+ risk_score = round(rng.uniform(0, 1), 3) if status != "pending" else None
+ rows.append(
+ (
+ str(uuid.uuid4()),
+ account_id,
+ kind,
+ amount,
+ weighted_choice(rng, CURRENCY_WEIGHTS),
+ rng.choice(MERCHANTS),
+ rng.randint(1, N_CATEGORIES),
+ status,
+ risk_score,
+ sample_created_at(rng, now),
+ )
+ )
+ return rows
+
+
+def gen_malformed_transactions(rng, accounts, poison_account_ids, now):
+ all_ids = [a["id"] for a in accounts if a["id"] not in poison_account_ids]
+ rows = []
+ poison_tx_ids = {}
+
+ # null_currency: every poison account gets at least one; the rest spread randomly.
+ remaining = MALFORMED_COUNTS["null_currency"] - len(poison_account_ids)
+ null_currency_accounts = list(poison_account_ids) + [
+ rng.choice(all_ids) for _ in range(remaining)
+ ]
+ for account_id in null_currency_accounts:
+ tx_id = str(uuid.uuid4())
+ if account_id in poison_account_ids:
+ poison_tx_ids[account_id] = tx_id
+ rows.append(
+ (
+ tx_id, account_id, "purchase",
+ max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9))),
+ None, rng.choice(MERCHANTS), rng.randint(1, N_CATEGORIES),
+ "settled", round(rng.uniform(0, 1), 3), sample_created_at(rng, now),
+ )
+ )
+
+ for _ in range(MALFORMED_COUNTS["negative_deposit"]):
+ rows.append(
+ (
+ str(uuid.uuid4()), rng.choice(all_ids), "deposit",
+ -max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9))),
+ weighted_choice(rng, CURRENCY_WEIGHTS), rng.choice(MERCHANTS),
+ rng.randint(1, N_CATEGORIES), "settled", round(rng.uniform(0, 1), 3),
+ sample_created_at(rng, now),
+ )
+ )
+
+ for _ in range(MALFORMED_COUNTS["non_ascii_merchant"]):
+ merchant = f"{rng.choice(MERCHANTS)} \U0001F600\\x00"
+ rows.append(
+ (
+ str(uuid.uuid4()), rng.choice(all_ids), "purchase",
+ max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9))),
+ weighted_choice(rng, CURRENCY_WEIGHTS), merchant,
+ rng.randint(1, N_CATEGORIES), "settled", round(rng.uniform(0, 1), 3),
+ sample_created_at(rng, now),
+ )
+ )
+
+ for _ in range(MALFORMED_COUNTS["future_dated"]):
+ future = now + timedelta(days=3, hours=rng.uniform(-2, 2))
+ rows.append(
+ (
+ str(uuid.uuid4()), rng.choice(all_ids), "purchase",
+ max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9))),
+ weighted_choice(rng, CURRENCY_WEIGHTS), rng.choice(MERCHANTS),
+ rng.randint(1, N_CATEGORIES), "pending", None, future,
+ )
+ )
+
+ for _ in range(MALFORMED_COUNTS["orphan_category"]):
+ rows.append(
+ (
+ str(uuid.uuid4()), rng.choice(all_ids), "purchase",
+ max(1, int(rng.lognormvariate(_median_to_mu(5_000), 0.9))),
+ weighted_choice(rng, CURRENCY_WEIGHTS), rng.choice(MERCHANTS),
+ 9999, "settled", round(rng.uniform(0, 1), 3), sample_created_at(rng, now),
+ )
+ )
+
+ return rows, poison_tx_ids
+
+
+def gen_sanctions_entries(rng, sanctioned_holder_names):
+ rows = []
+ for name in sanctioned_holder_names:
+ rows.append((name, rng.choice(["US", "GB", "FR", "DE", "IR", "RU"]), "OFAC-like-list"))
+
+ fake_first = FIRST_NAMES + ["Zephyrine", "Cassius", "Odalys", "Thaddeus", "Wilhelmina"]
+ fake_last = LAST_NAMES + ["Voskresenskaya", "Almacita", "Brennholt", "Cortavez"]
+ used = set(sanctioned_holder_names)
+ for _ in range(N_SANCTIONS - len(sanctioned_holder_names)):
+ while True:
+ name = f"{rng.choice(fake_first)} {rng.choice(fake_last)} {rng.randint(1000,9999)}"
+ if name not in used:
+ used.add(name)
+ break
+ rows.append((name, rng.choice(["US", "GB", "FR", "DE", "IR", "RU"]), "OFAC-like-list"))
+ return rows
+
+
+def main():
+ rng = random.Random(SEED)
+ now = datetime.now(timezone.utc)
+ t0 = time.monotonic()
+
+ accounts = gen_accounts(rng)
+ hot_accounts = [a for a in accounts if a["is_hot"]]
+ cold_accounts = [a for a in accounts if not a["is_hot"]]
+
+ role_pool = list(cold_accounts)
+ rng.shuffle(role_pool)
+ poison_accounts = role_pool[:N_POISON_ACCOUNTS]
+ sanctioned_accounts = role_pool[N_POISON_ACCOUNTS : N_POISON_ACCOUNTS + N_SANCTIONED_MATCHES]
+ clean_account = role_pool[N_POISON_ACCOUNTS + N_SANCTIONED_MATCHES]
+ cold_sample = role_pool[N_POISON_ACCOUNTS + N_SANCTIONED_MATCHES + 1 : N_POISON_ACCOUNTS + N_SANCTIONED_MATCHES + 61]
+
+ poison_account_ids = [a["id"] for a in poison_accounts]
+ sanctioned_account_ids = [a["id"] for a in sanctioned_accounts]
+ hot_account_ids = [a["id"] for a in hot_accounts]
+ cold_account_id_sample = [a["id"] for a in cold_sample]
+ clean_account_id = clean_account["id"]
+
+ categories = gen_categories(rng)
+ normal_txs = gen_normal_transactions(rng, accounts, now)
+ malformed_txs, poison_tx_ids = gen_malformed_transactions(rng, accounts, set(poison_account_ids), now)
+ all_txs = normal_txs + malformed_txs
+ rng.shuffle(all_txs)
+
+ sanctions_entries = gen_sanctions_entries(rng, [a["holder_name"] for a in sanctioned_accounts])
+
+ print(f"generated {len(accounts)} accounts, {len(all_txs)} transactions, "
+ f"{len(categories)} categories, {len(sanctions_entries)} sanctions entries "
+ f"in {time.monotonic() - t0:.1f}s (SEED={SEED})")
+
+ with psycopg.connect(DSN, autocommit=False) as conn:
+ with conn.cursor() as cur:
+ cur.execute(
+ "TRUNCATE accounts, merchant_categories, transactions, sanctions_entries, "
+ "notifications, audit_events RESTART IDENTITY"
+ )
+
+ with cur.copy(
+ "COPY accounts (id, holder_name, currency, balance_minor, is_hot, created_at) FROM STDIN"
+ ) as copy:
+ for a in accounts:
+ copy.write_row((a["id"], a["holder_name"], a["currency"], a["balance_minor"], a["is_hot"], a["created_at"]))
+
+ with cur.copy(
+ "COPY merchant_categories (id, code, label, risk_weight) FROM STDIN"
+ ) as copy:
+ for c in categories:
+ copy.write_row((c["id"], c["code"], c["label"], c["risk_weight"]))
+
+ with cur.copy(
+ "COPY transactions (id, account_id, kind, amount_minor, currency, merchant_name, "
+ "category_id, status, risk_score, created_at) FROM STDIN"
+ ) as copy:
+ for row in all_txs:
+ copy.write_row(row)
+
+ with cur.copy(
+ "COPY sanctions_entries (normalized_name, country, list_source) FROM STDIN"
+ ) as copy:
+ for row in sanctions_entries:
+ copy.write_row(row)
+
+ cur.execute("ANALYZE accounts, transactions, merchant_categories, sanctions_entries")
+
+ manifest = {
+ "seed": SEED,
+ "generated_at": now.isoformat(),
+ "counts": {
+ "accounts": len(accounts),
+ "transactions": len(all_txs),
+ "merchant_categories": len(categories),
+ "sanctions_entries": len(sanctions_entries),
+ },
+ "hot_account_ids": hot_account_ids,
+ "poison_account_ids": poison_account_ids,
+ "sanctioned_account_ids": sanctioned_account_ids,
+ "clean_account_id": clean_account_id,
+ "cold_account_ids_sample": cold_account_id_sample,
+ "malformed_row_count": len(malformed_txs),
+ }
+ cur.execute(
+ "INSERT INTO seed_manifest (id, data) VALUES (1, %s::jsonb) "
+ "ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data",
+ (json.dumps(manifest),),
+ )
+ conn.commit()
+
+ if poison_tx_ids:
+ import redis as redis_sync
+
+ redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
+ r = redis_sync.from_url(redis_url, decode_responses=True)
+ r.delete("poison:tx_by_account")
+ r.hset("poison:tx_by_account", mapping=poison_tx_ids)
+
+ elapsed = time.monotonic() - t0
+ print(f"seed complete in {elapsed:.1f}s")
+ print("row counts:")
+ for table, count in manifest["counts"].items():
+ print(f" {table}: {count}")
+ print(f"hot account ids: {hot_account_ids}")
+ print(f"poison account ids: {poison_account_ids[:5]}... ({len(poison_account_ids)} total)")
+ print(f"malformed rows: {manifest['malformed_row_count']}")
+ print(f"clean_account_id: {clean_account_id}")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/smoke.py b/scripts/smoke.py
new file mode 100644
index 0000000..07ae010
--- /dev/null
+++ b/scripts/smoke.py
@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""One transaction, one insights call, assert 200s -- phase-0-base-project-spec.md §12."""
+import json
+import os
+import sys
+import urllib.request
+
+import psycopg
+
+GATEWAY_URL = os.environ.get("GATEWAY_URL", "http://localhost:8000")
+DSN = os.environ.get("POSTGRES_DSN", "postgresql://ledgerline:ledgerline@localhost:5432/ledgerline")
+
+
+def _get(url):
+ with urllib.request.urlopen(url, timeout=10) as resp:
+ return resp.status, json.loads(resp.read())
+
+
+def _post(url, payload):
+ data = json.dumps(payload).encode()
+ req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
+ try:
+ with urllib.request.urlopen(req, timeout=10) as resp:
+ return resp.status, json.loads(resp.read())
+ except urllib.error.HTTPError as exc:
+ return exc.code, json.loads(exc.read())
+
+
+def main():
+ with psycopg.connect(DSN, autocommit=True) as conn:
+ with conn.cursor() as cur:
+ cur.execute("SELECT data FROM seed_manifest LIMIT 1")
+ row = cur.fetchone()
+ if not row:
+ print("smoke: no seed manifest found -- run `make seed` first", file=sys.stderr)
+ sys.exit(1)
+ account_id = row[0]["clean_account_id"]
+
+ status, body = _post(
+ f"{GATEWAY_URL}/api/transactions",
+ {
+ "account_id": account_id,
+ "kind": "deposit",
+ "amount_minor": 500,
+ "currency": "EUR",
+ "merchant_name": "Smoke Test Merchant",
+ },
+ )
+ assert status == 201, f"transaction failed: {status} {body}"
+ print(f"transaction ok: {body['transaction_id']} status={body['status']}")
+
+ status, body = _get(f"{GATEWAY_URL}/api/accounts/{account_id}/insights")
+ assert status == 200, f"insights failed: {status} {body}"
+ print(f"insights ok: {body}")
+
+ print("smoke test passed")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/services/__init__.py b/services/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/audit/__init__.py b/services/audit/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/audit/consumer.py b/services/audit/consumer.py
new file mode 100644
index 0000000..e28dd16
--- /dev/null
+++ b/services/audit/consumer.py
@@ -0,0 +1,68 @@
+import asyncio
+import json
+import logging
+
+import sqlalchemy as sa
+
+from common.db import Session
+from common.redis_client import redis_client
+from common.settings import settings
+
+logger = logging.getLogger("audit")
+
+GROUP = "audit"
+CONSUMER = "audit-1"
+
+
+async def ensure_group():
+ try:
+ await redis_client.xgroup_create(settings.stream_name, GROUP, id="$", mkstream=True)
+ except Exception as exc:
+ if "BUSYGROUP" not in str(exc):
+ raise
+
+
+async def _handle(fields: dict):
+ async with Session() as session:
+ await session.execute(
+ sa.text(
+ "INSERT INTO audit_events (transaction_id, event_type, payload, recorded_at) "
+ "VALUES (:tx_id, 'transaction_processed', CAST(:payload AS jsonb), now())"
+ ),
+ {"tx_id": fields.get("transaction_id"), "payload": json.dumps(fields)},
+ )
+ await session.commit()
+
+
+async def run():
+ """Second, independent consumer group on the same stream as `notifier` --
+ proves the async fork fans out to genuinely separate consumers, not one
+ consumer doing both jobs."""
+ await ensure_group()
+ logger.info("audit consumer started")
+ while True:
+ try:
+ resp = await redis_client.xreadgroup(
+ groupname=GROUP,
+ consumername=CONSUMER,
+ streams={settings.stream_name: ">"},
+ count=10,
+ block=2000,
+ )
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ logger.exception("audit consumer read failed")
+ await asyncio.sleep(1)
+ continue
+
+ if not resp:
+ continue
+
+ for _, messages in resp:
+ for msg_id, fields in messages:
+ try:
+ await _handle(fields)
+ except Exception:
+ logger.exception(f"audit failed to process message {msg_id}")
+ await redis_client.xack(settings.stream_name, GROUP, msg_id)
diff --git a/services/audit/worker.py b/services/audit/worker.py
new file mode 100644
index 0000000..20c8575
--- /dev/null
+++ b/services/audit/worker.py
@@ -0,0 +1,10 @@
+import asyncio
+
+from common.logging_config import configure_logging
+
+from . import consumer
+
+configure_logging("audit")
+
+if __name__ == "__main__":
+ asyncio.run(consumer.run())
diff --git a/services/fraud/__init__.py b/services/fraud/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/fraud/app.py b/services/fraud/app.py
new file mode 100644
index 0000000..2ff72ea
--- /dev/null
+++ b/services/fraud/app.py
@@ -0,0 +1,77 @@
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common.http import downstream_client
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+from .scoring import score
+
+SERVICE_NAME = "fraud"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class ScoreRequest(BaseModel):
+ account_id: str
+ holder_name: str
+ category_id: int | None = None
+ amount_minor: int
+ request_id: str = "unknown"
+
+
+@app.get("/")
+async def root():
+ return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+ if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+ logger.info("received chatty-health healthz probe from gateway")
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ try:
+ await redis_client.ping()
+ except Exception as exc:
+ logger.exception(f"readyz check failed: {exc}")
+ return JSONResponse(status_code=503, content={"status": "unavailable"})
+ return {"status": "ok"}
+
+
+@app.post("/internal/score")
+async def score_transaction(body: ScoreRequest):
+ logger.info(f"scoring account {body.account_id} request_id={body.request_id}")
+ await redis_client.incr("fraud:calls_total")
+
+ if await chaos.blocking_event_loop_enabled():
+ logger.info(f"blocking_event_loop chaos engaged for request_id={body.request_id}")
+ chaos.synchronous_cpu_work()
+
+ async with downstream_client(body.request_id) as client:
+ try:
+ resp = await client.post(
+ f"{settings.sanctions_url}/internal/lookup",
+ json={"holder_name": body.holder_name, "request_id": body.request_id},
+ )
+ except httpx.TimeoutException:
+ logger.exception(f"sanctions lookup timed out for request_id={body.request_id}")
+ return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"})
+ except httpx.RequestError as exc:
+ logger.exception(f"sanctions lookup failed for request_id={body.request_id}: {exc}")
+ return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"})
+
+ if resp.status_code >= 300:
+ logger.info(f"sanctions returned {resp.status_code} for request_id={body.request_id}")
+ return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"})
+
+ matched = resp.json().get("matched", False)
+ risk_score = score(body.category_id, matched)
+ return {"risk_score": risk_score, "matched": matched}
diff --git a/services/fraud/chaos.py b/services/fraud/chaos.py
new file mode 100644
index 0000000..1b0273c
--- /dev/null
+++ b/services/fraud/chaos.py
@@ -0,0 +1,22 @@
+"""blocking_event_loop -- phase-0-base-project-spec.md §7. Owned by `fraud`."""
+import hashlib
+import time
+
+from common.chaos_flags import is_enabled
+
+_CPU_WORK_SECONDS = 0.15
+
+
+async def blocking_event_loop_enabled() -> bool:
+ return await is_enabled("blocking_event_loop")
+
+
+def synchronous_cpu_work() -> None:
+ """~120ms of real, synchronous, CPU-bound work run directly inside an
+ `async def` handler -- no run_in_executor/to_thread. This is the point:
+ it blocks the whole event loop, serialising every other in-flight
+ request on this worker for the duration."""
+ deadline = time.perf_counter() + _CPU_WORK_SECONDS
+ data = b"ledgerline-fraud-scoring"
+ while time.perf_counter() < deadline:
+ data = hashlib.sha256(data).digest()
diff --git a/services/fraud/scoring.py b/services/fraud/scoring.py
new file mode 100644
index 0000000..72b0848
--- /dev/null
+++ b/services/fraud/scoring.py
@@ -0,0 +1,18 @@
+import random
+
+
+def category_weight(category_id: int | None) -> float:
+ """A fake but deterministic-ish stand-in for merchant_categories.risk_weight
+ (fraud has no direct DB access -- Redis only, per phase-0-base-project-spec.md §4).
+ The business logic does not need to be correct, only to produce a plausible
+ spread of scores."""
+ if category_id is None:
+ return 0.3
+ return 0.01 + (category_id % 45) / 44 * 0.94
+
+
+def score(category_id: int | None, matched: bool) -> float:
+ base = category_weight(category_id) * random.uniform(0.05, 0.3)
+ if matched:
+ return min(1.0, base + random.uniform(0.55, 0.9))
+ return min(1.0, base)
diff --git a/services/gateway/__init__.py b/services/gateway/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/gateway/app.py b/services/gateway/app.py
new file mode 100644
index 0000000..5a6e3f1
--- /dev/null
+++ b/services/gateway/app.py
@@ -0,0 +1,151 @@
+import asyncio
+import uuid
+from pathlib import Path
+
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, field_validator
+
+from common.chaos_flags import ALL_FLAGS, get_flags, set_flag, set_profile
+from common.logging_config import configure_logging
+from common.settings import settings
+
+from . import orchestrator
+from .orchestrator import UpstreamFailure
+
+SERVICE_NAME = "gateway"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+STATIC_DIR = Path(__file__).parent / "static"
+app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+
+# A shared, connection-pooling client for the simple read-only proxy routes
+# below -- opening a fresh httpx.AsyncClient (and TCP connection) per request
+# is real, needless overhead under concurrent load. orchestrator.py's
+# downstream_client() stays per-request since it carries per-transaction
+# headers (X-Request-Id).
+_proxy_client = httpx.AsyncClient(
+ timeout=settings.downstream_timeout_seconds,
+ limits=httpx.Limits(max_connections=100, max_keepalive_connections=50),
+)
+
+DOWNSTREAM_URLS = {
+ "ledger": settings.ledger_url,
+ "fraud": settings.fraud_url,
+ "sanctions": settings.sanctions_url,
+ "insights": settings.insights_url,
+ "notifier": settings.notifier_url,
+}
+
+
+class TransactionRequest(BaseModel):
+ account_id: str
+ kind: str
+ amount_minor: int
+ currency: str
+ merchant_name: str
+ category_id: int | None = None
+
+ @field_validator("kind")
+ @classmethod
+ def kind_must_be_known(cls, v):
+ if v not in ("deposit", "withdrawal", "purchase"):
+ raise ValueError("kind must be one of deposit, withdrawal, purchase")
+ return v
+
+ @field_validator("amount_minor")
+ @classmethod
+ def deposit_amount_must_be_non_negative(cls, v, info):
+ if info.data.get("kind") == "deposit" and v < 0:
+ raise ValueError("amount_minor must be non-negative for a deposit")
+ return v
+
+
+class ChaosFlagRequest(BaseModel):
+ flag: str
+ enabled: bool
+
+
+class ChaosProfileRequest(BaseModel):
+ profile: str
+
+
+@app.get("/", include_in_schema=False)
+async def index():
+ """Serves the frontend (phase-0-base-project-spec.md §10). This overrides
+ §4's generic `GET / -> {"service", "version"}` info route, which the
+ other five HTTP services still expose as-is."""
+ return FileResponse(STATIC_DIR / "index.html")
+
+
+@app.get("/healthz")
+async def healthz():
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ return {"status": "ok"}
+
+
+@app.get("/api/accounts")
+async def list_accounts():
+ resp = await _proxy_client.get(f"{settings.ledger_url}/internal/accounts")
+ return JSONResponse(status_code=resp.status_code, content=resp.json())
+
+
+@app.get("/api/accounts/{account_id}/insights")
+async def account_insights(account_id: str):
+ resp = await _proxy_client.get(f"{settings.insights_url}/internal/accounts/{account_id}/summary")
+ return JSONResponse(status_code=resp.status_code, content=resp.json())
+
+
+@app.post("/api/transactions", status_code=201)
+async def submit_transaction(body: TransactionRequest):
+ request_id = str(uuid.uuid4())
+ try:
+ result = await orchestrator.create_transaction(body.model_dump(), request_id)
+ except UpstreamFailure as exc:
+ return JSONResponse(
+ status_code=exc.status_code, content={"error": exc.message, "request_id": request_id}
+ )
+ return JSONResponse(status_code=201, content=result)
+
+
+@app.get("/api/health/all")
+async def health_all():
+ async def check(name, url):
+ try:
+ resp = await _proxy_client.get(f"{url}/healthz", timeout=2.0)
+ return name, "ok" if resp.status_code == 200 else "down"
+ except httpx.RequestError:
+ return name, "down"
+
+ results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()])
+ services = {"gateway": "ok", **dict(results)}
+ return {"services": services}
+
+
+@app.get("/api/admin/chaos")
+async def get_chaos():
+ return {"flags": await get_flags()}
+
+
+@app.post("/api/admin/chaos")
+async def post_chaos(body: ChaosFlagRequest):
+ if body.flag not in ALL_FLAGS:
+ return JSONResponse(status_code=400, content={"error": f"unknown flag {body.flag!r}"})
+ await set_flag(body.flag, body.enabled)
+ return {"flags": await get_flags()}
+
+
+@app.post("/api/admin/chaos/profile")
+async def post_chaos_profile(body: ChaosProfileRequest):
+ try:
+ flags = await set_profile(body.profile)
+ except ValueError as exc:
+ return JSONResponse(status_code=400, content={"error": str(exc)})
+ return {"flags": flags}
diff --git a/services/gateway/chaos.py b/services/gateway/chaos.py
new file mode 100644
index 0000000..4e21860
--- /dev/null
+++ b/services/gateway/chaos.py
@@ -0,0 +1,23 @@
+"""retry_storm, no_batching, chatty_health -- phase-0-base-project-spec.md §7.
+All three are owned by `gateway`."""
+import random
+
+from common.chaos_flags import is_enabled
+
+DOWNSTREAMS = ("ledger", "fraud", "sanctions", "insights", "notifier")
+
+
+async def retry_storm_enabled() -> bool:
+ return await is_enabled("retry_storm")
+
+
+async def no_batching_enabled() -> bool:
+ return await is_enabled("no_batching")
+
+
+async def chatty_health_enabled() -> bool:
+ return await is_enabled("chatty_health")
+
+
+def synthetic_line_item_count() -> int:
+ return random.randint(1, 5)
diff --git a/services/gateway/orchestrator.py b/services/gateway/orchestrator.py
new file mode 100644
index 0000000..066bc2b
--- /dev/null
+++ b/services/gateway/orchestrator.py
@@ -0,0 +1,167 @@
+import logging
+import time
+import uuid
+
+import httpx
+
+from common.http import downstream_client
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+
+logger = logging.getLogger("gateway")
+
+REJECT_THRESHOLD = 0.8
+
+
+class UpstreamFailure(Exception):
+ def __init__(self, status_code: int, message: str):
+ self.status_code = status_code
+ self.message = message
+
+
+async def _post(client: httpx.AsyncClient, url: str, json: dict) -> httpx.Response:
+ try:
+ return await client.post(url, json=json)
+ except httpx.TimeoutException as exc:
+ raise UpstreamFailure(504, f"{url} timed out") from exc
+ except httpx.RequestError as exc:
+ raise UpstreamFailure(502, f"{url} unreachable") from exc
+
+
+async def _score_once(client: httpx.AsyncClient, payload: dict) -> dict:
+ resp = await _post(client, f"{settings.fraud_url}/internal/score", payload)
+ if resp.status_code >= 300:
+ status = resp.status_code if resp.status_code in (502, 504) else 502
+ raise UpstreamFailure(status, "fraud scoring failed")
+ return resp.json()
+
+
+async def _score_with_retry(client: httpx.AsyncClient, payload: dict, retry_storm: bool) -> dict:
+ """retry_storm: on any non-2xx from fraud, retry 3 times immediately --
+ no backoff, no jitter, no circuit breaker (phase-0-base-project-spec.md §7).
+ "No circuit breaker" is taken literally here: once triggered, it fires all
+ 3 retries unconditionally rather than stopping early on the first success
+ -- it never checks whether it still needs to. The last attempt's outcome
+ wins."""
+ attempts = 4 if retry_storm else 1
+ result = None
+ last_exc = None
+ for _ in range(attempts):
+ try:
+ result = await _score_once(client, payload)
+ last_exc = None
+ except UpstreamFailure as exc:
+ last_exc = exc
+ if last_exc is not None:
+ raise last_exc
+ return result
+
+
+async def _run_chatty_health_probe(client: httpx.AsyncClient) -> None:
+ """chatty_health: GET /healthz on all five downstreams, sequentially,
+ before doing any real work."""
+ headers = {"X-Ledgerline-Probe": "chatty-health"}
+ for name in chaos.DOWNSTREAMS:
+ url = getattr(settings, f"{name}_url")
+ try:
+ await client.get(f"{url}/healthz", headers=headers, timeout=2.0)
+ except httpx.RequestError:
+ pass
+
+
+async def create_transaction(payload: dict, request_id: str) -> dict:
+ start = time.perf_counter()
+ logger.info(f"start transaction for account {payload['account_id']} request_id={request_id}")
+
+ retry_storm = await chaos.retry_storm_enabled()
+ no_batching = await chaos.no_batching_enabled()
+ chatty_health = await chaos.chatty_health_enabled()
+
+ async with downstream_client(request_id) as client:
+ if chatty_health:
+ await _run_chatty_health_probe(client)
+
+ create_resp = await _post(
+ client,
+ f"{settings.ledger_url}/internal/transactions",
+ {**payload, "request_id": request_id},
+ )
+ if create_resp.status_code == 404:
+ raise UpstreamFailure(404, "account not found")
+ if create_resp.status_code >= 500:
+ raise UpstreamFailure(create_resp.status_code, "ledger failed unexpectedly")
+ if create_resp.status_code >= 300:
+ raise UpstreamFailure(502, "ledger rejected the transaction")
+
+ created = create_resp.json()
+ tx_id = created["transaction_id"]
+
+ score_payload = {
+ "account_id": created["account_id"],
+ "holder_name": created["holder_name"],
+ "category_id": created["category_id"],
+ "request_id": request_id,
+ }
+
+ if no_batching:
+ item_count = chaos.synthetic_line_item_count()
+ scores = []
+ for _ in range(item_count):
+ item_payload = {**score_payload, "amount_minor": max(1, created["amount_minor"] // item_count)}
+ scores.append(await _score_with_retry(client, item_payload, retry_storm))
+ risk_score = max(s["risk_score"] for s in scores)
+ else:
+ score_payload["amount_minor"] = created["amount_minor"]
+ scored = await _score_with_retry(client, score_payload, retry_storm)
+ risk_score = scored["risk_score"]
+
+ decision = "rejected" if risk_score > REJECT_THRESHOLD else "settled"
+ action = "reject" if decision == "rejected" else "settle"
+ settle_resp = await _post(
+ client,
+ f"{settings.ledger_url}/internal/transactions/{tx_id}/{action}",
+ {"risk_score": risk_score, "request_id": request_id},
+ )
+ if settle_resp.status_code >= 300:
+ raise UpstreamFailure(502, "ledger failed to finalise the transaction")
+ settled = settle_resp.json()
+
+ try:
+ await redis_client.xadd(
+ settings.stream_name,
+ {
+ "transaction_id": tx_id,
+ "account_id": created["account_id"],
+ "kind": created["kind"],
+ "amount_minor": str(created["amount_minor"]),
+ "status": settled["status"],
+ "risk_score": str(risk_score),
+ },
+ )
+ except Exception:
+ logger.exception(f"failed to publish transaction {tx_id} to the stream")
+
+ summary = {}
+ try:
+ summary_resp = await client.get(
+ f"{settings.insights_url}/internal/accounts/{created['account_id']}/quick_summary"
+ )
+ if summary_resp.status_code == 200:
+ summary = summary_resp.json()
+ except httpx.RequestError:
+ logger.exception(f"insights summary unavailable for request_id={request_id}")
+
+ elapsed_ms = (time.perf_counter() - start) * 1000
+ logger.info(f"finished transaction {tx_id} status={settled['status']} request_id={request_id}")
+
+ return {
+ "transaction_id": tx_id,
+ "status": settled["status"],
+ "risk_score": risk_score,
+ "balance_minor": settled["balance_minor"],
+ "summary": summary,
+ "request_id": request_id,
+ "elapsed_ms": elapsed_ms,
+ }
diff --git a/services/gateway/static/app.js b/services/gateway/static/app.js
new file mode 100644
index 0000000..183d2e7
--- /dev/null
+++ b/services/gateway/static/app.js
@@ -0,0 +1,149 @@
+const accountSelect = document.getElementById("account-select");
+const txForm = document.getElementById("tx-form");
+const resultJson = document.getElementById("result-json");
+const resultRtt = document.getElementById("result-rtt");
+const resultRequestId = document.getElementById("result-request-id");
+const burstButton = document.getElementById("burst-button");
+const burstN = document.getElementById("burst-n");
+const burstTableBody = document.querySelector("#burst-table tbody");
+const chaosFlagsDiv = document.getElementById("chaos-flags");
+const healthStrip = document.getElementById("health-strip");
+
+async function loadAccounts() {
+ const resp = await fetch("/api/accounts");
+ const accounts = await resp.json();
+ accountSelect.innerHTML = accounts
+ .slice(0, 200)
+ .map((a) => ``)
+ .join("");
+}
+
+txForm.addEventListener("submit", async (e) => {
+ e.preventDefault();
+ const payload = {
+ account_id: accountSelect.value,
+ kind: document.querySelector('input[name="kind"]:checked').value,
+ amount_minor: Number(document.getElementById("amount-input").value),
+ currency: "EUR",
+ merchant_name: document.getElementById("merchant-input").value,
+ };
+ const start = performance.now();
+ const resp = await fetch("/api/transactions", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify(payload),
+ });
+ const rtt = performance.now() - start;
+ const body = await resp.json();
+ resultRtt.textContent = rtt.toFixed(1);
+ resultRequestId.textContent = body.request_id || "--";
+ resultJson.textContent = JSON.stringify(body, null, 2);
+});
+
+function percentile(sorted, p) {
+ if (sorted.length === 0) return 0;
+ const rank = Math.max(1, Math.min(sorted.length, Math.ceil(p * sorted.length)));
+ return sorted[rank - 1];
+}
+
+burstButton.addEventListener("click", async () => {
+ const n = Number(burstN.value);
+ const accounts = Array.from(accountSelect.options).map((o) => o.value);
+ burstTableBody.innerHTML = `| running ${n} requests... |
`;
+
+ const latencies = [];
+ let success = 0;
+ let errors = 0;
+
+ await Promise.all(
+ Array.from({ length: n }, async () => {
+ const accountId = accounts[Math.floor(Math.random() * accounts.length)];
+ const start = performance.now();
+ try {
+ const resp = await fetch("/api/transactions", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({
+ account_id: accountId,
+ kind: "purchase",
+ amount_minor: 500,
+ currency: "EUR",
+ merchant_name: "Load Burst",
+ }),
+ });
+ latencies.push(performance.now() - start);
+ if (resp.ok) success += 1;
+ else errors += 1;
+ } catch (e) {
+ latencies.push(performance.now() - start);
+ errors += 1;
+ }
+ })
+ );
+
+ latencies.sort((a, b) => a - b);
+ const mean = latencies.reduce((a, b) => a + b, 0) / latencies.length;
+ const row = [
+ latencies.length,
+ success,
+ errors,
+ latencies[0].toFixed(1),
+ mean.toFixed(1),
+ percentile(latencies, 0.5).toFixed(1),
+ percentile(latencies, 0.95).toFixed(1),
+ latencies[latencies.length - 1].toFixed(1),
+ ];
+ burstTableBody.innerHTML = `${row.map((v) => `| ${v} | `).join("")}
`;
+});
+
+async function loadChaos() {
+ const resp = await fetch("/api/admin/chaos");
+ const { flags } = await resp.json();
+ chaosFlagsDiv.innerHTML = Object.entries(flags)
+ .map(
+ ([flag, enabled]) => `
+ `
+ )
+ .join("");
+
+ chaosFlagsDiv.querySelectorAll("input[type=checkbox]").forEach((el) => {
+ el.addEventListener("change", async () => {
+ await fetch("/api/admin/chaos", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({ flag: el.dataset.flag, enabled: el.checked }),
+ });
+ });
+ });
+}
+
+document.querySelectorAll("#chaos-profiles button").forEach((btn) => {
+ btn.addEventListener("click", async () => {
+ await fetch("/api/admin/chaos/profile", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({ profile: btn.dataset.profile }),
+ });
+ await loadChaos();
+ });
+});
+
+async function pollHealth() {
+ try {
+ const resp = await fetch("/api/health/all");
+ const { services } = await resp.json();
+ healthStrip.innerHTML = Object.entries(services)
+ .map(([name, status]) => `${name}`)
+ .join("");
+ } catch (e) {
+ // ignore transient polling failures
+ }
+}
+
+loadAccounts();
+loadChaos();
+pollHealth();
+setInterval(pollHealth, 5000);
diff --git a/services/gateway/static/index.html b/services/gateway/static/index.html
new file mode 100644
index 0000000..4ce297f
--- /dev/null
+++ b/services/gateway/static/index.html
@@ -0,0 +1,80 @@
+
+
+
+
+ Ledgerline
+
+
+
+
+
+
+
+ Submit a transaction
+
+
+
+
+ Result
+ Round-trip: -- ms · request_id: --
+ (nothing yet)
+
+
+
+ Load burst
+
+
+
+
+ | count | success | errors | min | mean | p50 | p95 | max |
|---|
+
+ | no burst run yet |
+
+
+
+
+ Chaos
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/services/gateway/static/style.css b/services/gateway/static/style.css
new file mode 100644
index 0000000..7088297
--- /dev/null
+++ b/services/gateway/static/style.css
@@ -0,0 +1,184 @@
+:root {
+ color-scheme: dark;
+ --bg: #0f1117;
+ --panel: #171a23;
+ --border: #2a2e3a;
+ --text: #d8dce6;
+ --muted: #7b8296;
+ --accent: #5fd0a8;
+ --danger: #e0685f;
+ --mono: "SF Mono", "Fira Code", Consolas, monospace;
+}
+
+* {
+ box-sizing: border-box;
+}
+
+body {
+ background: var(--bg);
+ color: var(--text);
+ font-family: var(--mono);
+ margin: 0;
+ padding: 0 1.5rem 3rem;
+}
+
+header {
+ padding: 2rem 0 1rem;
+ border-bottom: 1px solid var(--border);
+}
+
+header h1 {
+ margin: 0;
+ font-size: 1.6rem;
+ letter-spacing: 0.02em;
+}
+
+.tagline {
+ color: var(--muted);
+ margin: 0.25rem 0 0;
+ font-size: 0.85rem;
+}
+
+main {
+ display: grid;
+ grid-template-columns: repeat(auto-fit, minmax(320px, 1fr));
+ gap: 1.5rem;
+ margin-top: 1.5rem;
+}
+
+section {
+ background: var(--panel);
+ border: 1px solid var(--border);
+ border-radius: 6px;
+ padding: 1rem 1.25rem;
+}
+
+section h2 {
+ margin-top: 0;
+ font-size: 1rem;
+ color: var(--accent);
+ text-transform: uppercase;
+ letter-spacing: 0.05em;
+}
+
+label {
+ display: block;
+ margin-bottom: 0.6rem;
+ font-size: 0.85rem;
+ color: var(--muted);
+}
+
+input, select, button {
+ font-family: var(--mono);
+ background: #0c0e13;
+ color: var(--text);
+ border: 1px solid var(--border);
+ border-radius: 4px;
+ padding: 0.4rem 0.5rem;
+ margin-top: 0.2rem;
+ width: 100%;
+}
+
+fieldset {
+ border: 1px solid var(--border);
+ border-radius: 4px;
+ margin-bottom: 0.6rem;
+}
+
+fieldset label {
+ display: inline-block;
+ width: auto;
+ margin-right: 1rem;
+}
+
+button {
+ cursor: pointer;
+ background: var(--accent);
+ color: #0c0e13;
+ font-weight: 600;
+ border: none;
+ margin-top: 0.5rem;
+ transition: opacity 0.15s ease;
+}
+
+button:hover {
+ opacity: 0.85;
+}
+
+#chaos-profiles {
+ display: flex;
+ gap: 0.5rem;
+ margin-top: 0.75rem;
+}
+
+#chaos-profiles button {
+ width: auto;
+ flex: 1;
+}
+
+pre#result-json {
+ background: #0c0e13;
+ border: 1px solid var(--border);
+ border-radius: 4px;
+ padding: 0.75rem;
+ overflow-x: auto;
+ font-size: 0.8rem;
+ max-height: 260px;
+}
+
+code {
+ background: #0c0e13;
+ padding: 0.1rem 0.3rem;
+ border-radius: 3px;
+ user-select: all;
+}
+
+table {
+ width: 100%;
+ border-collapse: collapse;
+ margin-top: 0.75rem;
+ font-size: 0.8rem;
+}
+
+th, td {
+ border: 1px solid var(--border);
+ padding: 0.35rem 0.5rem;
+ text-align: right;
+}
+
+th:first-child, td:first-child {
+ text-align: left;
+}
+
+#chaos-flags label {
+ display: flex;
+ align-items: center;
+ gap: 0.4rem;
+}
+
+#chaos-flags input {
+ width: auto;
+}
+
+#health-strip {
+ display: flex;
+ flex-wrap: wrap;
+ gap: 0.5rem;
+}
+
+.dot {
+ border-radius: 999px;
+ padding: 0.25rem 0.6rem;
+ font-size: 0.75rem;
+ border: 1px solid var(--border);
+}
+
+.dot.green {
+ color: var(--accent);
+ border-color: var(--accent);
+}
+
+.dot.red {
+ color: var(--danger);
+ border-color: var(--danger);
+}
diff --git a/services/insights/__init__.py b/services/insights/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/insights/analytics.py b/services/insights/analytics.py
new file mode 100644
index 0000000..086437f
--- /dev/null
+++ b/services/insights/analytics.py
@@ -0,0 +1,82 @@
+import sqlalchemy as sa
+
+# Deliberately the same shape of query the test suite's own EXPLAIN check runs
+# against the schema directly (tests/test_pathologies.py, 6.4) -- kept here
+# only for the honest (non-n_plus_one) path; that test doesn't call this code.
+
+_N_PLUS_ONE_ROW_CAP = 1500
+
+
+async def account_exists(session, account_id: str) -> bool:
+ result = await session.execute(sa.text("SELECT 1 FROM accounts WHERE id = :id"), {"id": account_id})
+ return result.first() is not None
+
+
+async def compute_summary(session, account_id: str, *, n_plus_one: bool) -> dict:
+ if n_plus_one:
+ rows = await _n_plus_one_summary(session, account_id)
+ else:
+ rows = await _joined_summary(session, account_id)
+
+ total_count = 0
+ total_amount = 0
+ by_category: dict = {}
+ for r in rows:
+ total_count += 1
+ total_amount += r["amount_minor"]
+ label = r["label"] or "uncategorized"
+ entry = by_category.setdefault(label, {"count": 0, "total_amount_minor": 0})
+ entry["count"] += 1
+ entry["total_amount_minor"] += r["amount_minor"]
+
+ return {
+ "account_id": account_id,
+ "window_days": 120,
+ "transaction_count": total_count,
+ "total_amount_minor": total_amount,
+ "by_category": by_category,
+ }
+
+
+async def _joined_summary(session, account_id: str) -> list[dict]:
+ """The honest path: a single JOIN, one statement."""
+ result = await session.execute(
+ sa.text(
+ "SELECT t.amount_minor, mc.label FROM transactions t "
+ "LEFT JOIN merchant_categories mc ON mc.id = t.category_id "
+ "WHERE t.account_id = :aid AND t.created_at >= now() - interval '120 days'"
+ ),
+ {"aid": account_id},
+ )
+ return [dict(row) for row in result.mappings().all()]
+
+
+async def _n_plus_one_summary(session, account_id: str) -> list[dict]:
+ """n_plus_one: instead of the JOIN above, fetch rows without one and issue
+ a separate category lookup per row -- phase-0-base-project-spec.md §7.
+ Capped so the pathology stays a genuine multiplication of statements
+ (see tests/test_pathologies.py 6.3) without turning a single request into
+ an unbounded number of round trips on a high-volume hot account."""
+ result = await session.execute(
+ sa.text(
+ "SELECT amount_minor, category_id FROM transactions "
+ "WHERE account_id = :aid AND created_at >= now() - interval '120 days' "
+ f"LIMIT {_N_PLUS_ONE_ROW_CAP}"
+ ),
+ {"aid": account_id},
+ )
+ tx_rows = result.mappings().all()
+
+ rows = []
+ for tx_row in tx_rows:
+ label = None
+ if tx_row["category_id"] is not None:
+ cat_result = await session.execute(
+ sa.text("SELECT label FROM merchant_categories WHERE id = :id"),
+ {"id": tx_row["category_id"]},
+ )
+ cat_row = cat_result.first()
+ if cat_row is not None:
+ label = cat_row.label
+ rows.append({"amount_minor": tx_row["amount_minor"], "label": label})
+ return rows
diff --git a/services/insights/app.py b/services/insights/app.py
new file mode 100644
index 0000000..06aa60c
--- /dev/null
+++ b/services/insights/app.py
@@ -0,0 +1,120 @@
+import asyncio
+import json
+import time
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from common.chaos_flags import flags_fingerprint
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+from common.settings import settings
+
+import sqlalchemy as sa
+
+from . import chaos
+from .analytics import account_exists, compute_summary
+from .db import NormalSession, engine, session_factory
+
+SERVICE_NAME = "insights"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+# Cache-stampede protection for the Redis-cached branch below: a burst of
+# concurrent requests for the same not-yet-cached (account, flags) key would
+# otherwise all pay the full computation cost at once instead of one paying
+# it and the rest waiting on the result. Not needed for unbounded_cache,
+# which never caches by design.
+_inflight_locks: dict[str, asyncio.Lock] = {}
+
+
+def _lock_for(key: str) -> asyncio.Lock:
+ lock = _inflight_locks.get(key)
+ if lock is None:
+ lock = asyncio.Lock()
+ _inflight_locks[key] = lock
+ return lock
+
+
+@app.get("/")
+async def root():
+ return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+ if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+ logger.info("received chatty-health healthz probe from gateway")
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ try:
+ async with engine.connect() as conn:
+ await conn.exec_driver_sql("SELECT 1")
+ except Exception as exc:
+ logger.exception(f"readyz check failed: {exc}")
+ return JSONResponse(status_code=503, content={"status": "unavailable"})
+ return {"status": "ok"}
+
+
+@app.get("/internal/accounts/{account_id}/quick_summary")
+async def quick_summary(account_id: str):
+ """A cheap, account-primary-key-only lookup embedded in every transaction
+ response (phase-0-base-project-spec.md §6.1's "summary" field) -- deliberately
+ separate from the 120-day aggregation below (§6.2), which is the one where
+ n_plus_one and the missing index are meant to bite. That aggregation scans
+ `transactions` by account_id with no supporting index on purpose; embedding
+ it in every single transaction's response would make the missing-index
+ pathology fire on every request instead of only on the analytics endpoint."""
+ async with NormalSession() as session:
+ result = await session.execute(
+ sa.text("SELECT currency, balance_minor FROM accounts WHERE id = :id"),
+ {"id": account_id},
+ )
+ row = result.first()
+ if row is None:
+ return JSONResponse(status_code=404, content={"error": "account not found"})
+ return {"account_id": account_id, "currency": row.currency, "balance_minor": row.balance_minor}
+
+
+@app.get("/internal/accounts/{account_id}/summary")
+async def summary(account_id: str):
+ n_plus_one = await chaos.n_plus_one_enabled()
+ cold_cache = await chaos.cold_cache_enabled()
+ unbounded_cache = await chaos.unbounded_cache_enabled()
+ fingerprint = await flags_fingerprint()
+
+ # unbounded_cache means every single request is a guaranteed fresh miss
+ # needing a real connection -- under concurrent load that alone is enough
+ # demand to exhaust a small pool, without needing n_plus_one too (see
+ # services/insights/db.py).
+ starved = unbounded_cache
+
+ async with session_factory(starved) as session:
+ if not await account_exists(session, account_id):
+ return JSONResponse(status_code=404, content={"error": "account not found"})
+
+ if unbounded_cache:
+ # Broken on purpose: keyed by a timestamp that's unique per call,
+ # so this never hits -- it only ever grows. See chaos.py.
+ key = (account_id, fingerprint, time.time_ns())
+ result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
+ chaos.UNBOUNDED_CACHE[key] = result
+ return result
+
+ ttl = settings.summary_cache_ttl_seconds_cold if cold_cache else settings.summary_cache_ttl_seconds
+ cache_key = f"summary:{account_id}:{fingerprint}"
+ cached = await redis_client.get(cache_key)
+ if cached is not None:
+ return json.loads(cached)
+
+ async with _lock_for(cache_key):
+ cached = await redis_client.get(cache_key)
+ if cached is not None:
+ return json.loads(cached)
+ async with session_factory(starved) as session:
+ result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
+ await redis_client.setex(cache_key, ttl, json.dumps(result))
+ return result
diff --git a/services/insights/chaos.py b/services/insights/chaos.py
new file mode 100644
index 0000000..6c19659
--- /dev/null
+++ b/services/insights/chaos.py
@@ -0,0 +1,22 @@
+"""n_plus_one, cold_cache, unbounded_cache -- phase-0-base-project-spec.md §7.
+All three are owned by `insights`."""
+from common.chaos_flags import is_enabled
+
+
+async def n_plus_one_enabled() -> bool:
+ return await is_enabled("n_plus_one")
+
+
+async def cold_cache_enabled() -> bool:
+ return await is_enabled("cold_cache")
+
+
+async def unbounded_cache_enabled() -> bool:
+ return await is_enabled("unbounded_cache")
+
+
+# unbounded_cache: memoize summaries in a module-level dict keyed by
+# (account_id, timestamp_ms) -- effectively never hits, never evicts, grows
+# forever. This is the pathology itself, not a fix for anything -- never
+# clear it, never cap it.
+UNBOUNDED_CACHE: dict = {}
diff --git a/services/insights/db.py b/services/insights/db.py
new file mode 100644
index 0000000..df6efff
--- /dev/null
+++ b/services/insights/db.py
@@ -0,0 +1,25 @@
+"""insights gets its own engines, separate from common.db's shared default,
+so pool sizing here can be tuned independently of sanctions/notifier/audit
+(which also import common.db).
+
+Two pools, selected per-request: a generous default, and a small one used
+whenever unbounded_cache is on -- every request is then a guaranteed fresh
+miss needing a real connection, and under concurrent load that alone is
+enough demand to exhaust a small pool. This is a realistic capacity limit
+that emerges under load, not a 14th named chaos flag."""
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from common.db import make_engine
+
+normal_engine = make_engine(pool_size=20, max_overflow=10)
+starved_engine = make_engine(pool_size=5, max_overflow=3, pool_timeout=0.5)
+
+NormalSession = async_sessionmaker(normal_engine, expire_on_commit=False)
+StarvedSession = async_sessionmaker(starved_engine, expire_on_commit=False)
+
+# Exposed for readyz's connectivity check.
+engine = normal_engine
+
+
+def session_factory(starved: bool):
+ return StarvedSession() if starved else NormalSession()
diff --git a/services/ledger/__init__.py b/services/ledger/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/ledger/app.py b/services/ledger/app.py
new file mode 100644
index 0000000..58b9da9
--- /dev/null
+++ b/services/ledger/app.py
@@ -0,0 +1,146 @@
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, field_validator
+
+from common.logging_config import configure_logging
+
+from . import chaos
+from .db import normal_engine, session_factory
+from . import repository
+
+SERVICE_NAME = "ledger"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class CreateTransactionRequest(BaseModel):
+ account_id: str
+ kind: str
+ amount_minor: int
+ currency: str
+ merchant_name: str
+ category_id: int | None = None
+ request_id: str = "unknown"
+
+ @field_validator("amount_minor")
+ @classmethod
+ def deposit_amount_must_be_non_negative(cls, v, info):
+ kind = info.data.get("kind")
+ if kind == "deposit" and v < 0:
+ raise ValueError("amount_minor must be non-negative for a deposit")
+ return v
+
+
+class SettleRequest(BaseModel):
+ risk_score: float
+ request_id: str = "unknown"
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+ logger.exception(f"unhandled exception on {request.url.path}: {exc}")
+ return JSONResponse(status_code=500, content={"error": "internal error"})
+
+
+@app.get("/")
+async def root():
+ return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+ if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+ logger.info("received chatty-health healthz probe from gateway")
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ try:
+ async with normal_engine.connect() as conn:
+ await conn.exec_driver_sql("SELECT 1")
+ except Exception as exc:
+ logger.exception(f"readyz check failed: {exc}")
+ return JSONResponse(status_code=503, content={"status": "unavailable"})
+ return {"status": "ok"}
+
+
+@app.get("/internal/accounts")
+async def list_accounts():
+ async with session_factory(starved=False) as session:
+ rows = await repository.list_accounts(session)
+ return [dict(row) for row in rows]
+
+
+@app.post("/internal/transactions")
+async def create_transaction(body: CreateTransactionRequest):
+ logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor} request_id={body.request_id}")
+
+ starved = await chaos.pool_starvation_enabled()
+ poison_enabled = await chaos.poison_data_enabled()
+ poison_tx_id = await chaos.find_poison_transaction_id(body.account_id)
+
+ async with session_factory(starved=starved) as session:
+ account = await repository.get_account(session, body.account_id)
+ if account is None:
+ return JSONResponse(status_code=404, content={"error": "account not found"})
+
+ if poison_tx_id is not None and poison_enabled:
+ poison_row = await repository.get_transaction_currency(session, poison_tx_id)
+ # poison_data: skip input validation on the seeded malformed rows
+ # (§8) -- this naive access raises on the NULL currency on purpose.
+ _ = poison_row.currency.upper()
+
+ tx_id = await repository.create_transaction(
+ session,
+ account_id=body.account_id,
+ kind=body.kind,
+ amount_minor=body.amount_minor,
+ currency=body.currency,
+ merchant_name=body.merchant_name,
+ category_id=body.category_id,
+ )
+
+ return JSONResponse(
+ status_code=201,
+ content={
+ "transaction_id": tx_id,
+ "account_id": body.account_id,
+ "holder_name": account.holder_name,
+ "kind": body.kind,
+ "amount_minor": body.amount_minor,
+ "category_id": body.category_id,
+ "status": "pending",
+ },
+ )
+
+
+@app.post("/internal/transactions/{tx_id}/settle")
+async def settle(tx_id: str, body: SettleRequest):
+ starved = await chaos.pool_starvation_enabled()
+ row_lock = await chaos.row_lock_contention_enabled()
+
+ async with session_factory(starved=starved) as session:
+ tx = await repository.get_transaction(session, tx_id)
+ if tx is None:
+ return JSONResponse(status_code=404, content={"error": "transaction not found"})
+ account = await repository.get_account(session, tx.account_id)
+ use_lock = row_lock and bool(account.is_hot)
+ balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock)
+
+ logger.info(f"settled transaction {tx_id} request_id={body.request_id}")
+ return {"status": "settled", "balance_minor": balance}
+
+
+@app.post("/internal/transactions/{tx_id}/reject")
+async def reject(tx_id: str, body: SettleRequest):
+ starved = await chaos.pool_starvation_enabled()
+
+ async with session_factory(starved=starved) as session:
+ tx = await repository.get_transaction(session, tx_id)
+ if tx is None:
+ return JSONResponse(status_code=404, content={"error": "transaction not found"})
+ balance = await repository.reject_transaction(session, tx=tx, risk_score=body.risk_score)
+
+ logger.info(f"rejected transaction {tx_id} request_id={body.request_id}")
+ return {"status": "rejected", "balance_minor": balance}
diff --git a/services/ledger/chaos.py b/services/ledger/chaos.py
new file mode 100644
index 0000000..617ea72
--- /dev/null
+++ b/services/ledger/chaos.py
@@ -0,0 +1,28 @@
+"""pool_starvation, row_lock_contention, poison_data -- phase-0-base-project-spec.md
+§7. All three are owned by `ledger`.
+"""
+from common.chaos_flags import is_enabled
+from common.redis_client import redis_client
+
+# Populated once by scripts/seed.py: account_id -> id of that account's one
+# deliberately NULL-currency transaction (§8). A Redis hash lookup, not a
+# `transactions` scan by account_id -- that column has no supporting index
+# on purpose (§5), and the poison_data pathology shouldn't have to pay for a
+# full-table seq scan on every single request just to find its own trigger.
+POISON_TX_KEY = "poison:tx_by_account"
+
+
+async def pool_starvation_enabled() -> bool:
+ return await is_enabled("pool_starvation")
+
+
+async def row_lock_contention_enabled() -> bool:
+ return await is_enabled("row_lock_contention")
+
+
+async def poison_data_enabled() -> bool:
+ return await is_enabled("poison_data")
+
+
+async def find_poison_transaction_id(account_id: str) -> str | None:
+ return await redis_client.hget(POISON_TX_KEY, account_id)
diff --git a/services/ledger/db.py b/services/ledger/db.py
new file mode 100644
index 0000000..31f054f
--- /dev/null
+++ b/services/ledger/db.py
@@ -0,0 +1,17 @@
+"""Two pre-built engines so pool_starvation (phase-0-base-project-spec.md §7)
+can be toggled live, with no restart: the flag picks which already-initialised
+pool a request borrows a connection from, rather than resizing a pool in
+place (SQLAlchemy pools aren't resizable live)."""
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from common.db import make_engine
+
+normal_engine = make_engine(pool_size=20, max_overflow=10)
+starved_engine = make_engine(pool_size=2, max_overflow=0, pool_timeout=10)
+
+NormalSession = async_sessionmaker(normal_engine, expire_on_commit=False)
+StarvedSession = async_sessionmaker(starved_engine, expire_on_commit=False)
+
+
+def session_factory(starved: bool):
+ return StarvedSession() if starved else NormalSession()
diff --git a/services/ledger/repository.py b/services/ledger/repository.py
new file mode 100644
index 0000000..9a5d8f9
--- /dev/null
+++ b/services/ledger/repository.py
@@ -0,0 +1,122 @@
+import asyncio
+import uuid
+
+import sqlalchemy as sa
+
+KIND_SIGN = {"deposit": 1, "withdrawal": -1, "purchase": -1}
+
+ROW_LOCK_HOLD_SECONDS = 0.15
+
+
+async def get_account(session, account_id: str):
+ result = await session.execute(
+ sa.text("SELECT id, holder_name, currency, balance_minor, is_hot FROM accounts WHERE id = :id"),
+ {"id": account_id},
+ )
+ return result.first()
+
+
+async def list_accounts(session, limit: int = 1000):
+ result = await session.execute(
+ sa.text(
+ "SELECT id, holder_name, currency, balance_minor FROM accounts ORDER BY created_at LIMIT :limit"
+ ),
+ {"limit": limit},
+ )
+ return result.mappings().all()
+
+
+async def get_transaction_currency(session, tx_id: str):
+ """Backs the poison_data pathology: fetches one known-malformed transaction
+ (NULL currency, §8) by primary key -- a cheap, always-fast lookup regardless
+ of the deliberately-missing account_id index, since which transaction to
+ fetch is looked up by account_id in Redis first (chaos.py's
+ find_poison_transaction_id), not by scanning `transactions` here."""
+ result = await session.execute(
+ sa.text("SELECT currency FROM transactions WHERE id = :id"),
+ {"id": tx_id},
+ )
+ return result.first()
+
+
+async def create_transaction(
+ session, *, account_id, kind, amount_minor, currency, merchant_name, category_id
+):
+ tx_id = str(uuid.uuid4())
+ await session.execute(
+ sa.text(
+ "INSERT INTO transactions "
+ "(id, account_id, kind, amount_minor, currency, merchant_name, category_id, status) "
+ "VALUES (:id, :account_id, :kind, :amount_minor, :currency, :merchant_name, :category_id, 'pending')"
+ ),
+ {
+ "id": tx_id,
+ "account_id": account_id,
+ "kind": kind,
+ "amount_minor": amount_minor,
+ "currency": currency,
+ "merchant_name": merchant_name,
+ "category_id": category_id,
+ },
+ )
+ await session.commit()
+ return tx_id
+
+
+async def get_transaction(session, tx_id: str):
+ result = await session.execute(
+ sa.text(
+ "SELECT id, account_id, kind, amount_minor, currency, merchant_name, "
+ "category_id, status FROM transactions WHERE id = :id"
+ ),
+ {"id": tx_id},
+ )
+ return result.first()
+
+
+async def settle_transaction(session, *, tx, risk_score, use_row_lock: bool):
+ """Applies the balance delta and marks the transaction settled. When
+ use_row_lock is true (row_lock_contention on a hot account), takes
+ SELECT ... FOR UPDATE and holds the transaction open for an extra 150ms
+ before committing -- phase-0-base-project-spec.md §7."""
+ delta = KIND_SIGN[tx.kind] * tx.amount_minor
+
+ if use_row_lock:
+ result = await session.execute(
+ sa.text("SELECT balance_minor FROM accounts WHERE id = :id FOR UPDATE"),
+ {"id": tx.account_id},
+ )
+ result.first()
+ await session.execute(
+ sa.text("UPDATE accounts SET balance_minor = balance_minor + :delta WHERE id = :id"),
+ {"delta": delta, "id": tx.account_id},
+ )
+ await asyncio.sleep(ROW_LOCK_HOLD_SECONDS)
+ else:
+ await session.execute(
+ sa.text("UPDATE accounts SET balance_minor = balance_minor + :delta WHERE id = :id"),
+ {"delta": delta, "id": tx.account_id},
+ )
+
+ await session.execute(
+ sa.text("UPDATE transactions SET status = 'settled', risk_score = :risk_score WHERE id = :id"),
+ {"risk_score": risk_score, "id": tx.id},
+ )
+ await session.commit()
+
+ result = await session.execute(
+ sa.text("SELECT balance_minor FROM accounts WHERE id = :id"), {"id": tx.account_id}
+ )
+ return result.scalar_one()
+
+
+async def reject_transaction(session, *, tx, risk_score):
+ await session.execute(
+ sa.text("UPDATE transactions SET status = 'rejected', risk_score = :risk_score WHERE id = :id"),
+ {"risk_score": risk_score, "id": tx.id},
+ )
+ await session.commit()
+ result = await session.execute(
+ sa.text("SELECT balance_minor FROM accounts WHERE id = :id"), {"id": tx.account_id}
+ )
+ return result.scalar_one()
diff --git a/services/notifier/__init__.py b/services/notifier/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/notifier/app.py b/services/notifier/app.py
new file mode 100644
index 0000000..63eaee8
--- /dev/null
+++ b/services/notifier/app.py
@@ -0,0 +1,49 @@
+from contextlib import asynccontextmanager
+
+import asyncio
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from common.db import engine
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+
+from . import consumer
+
+SERVICE_NAME = "notifier"
+logger = configure_logging(SERVICE_NAME)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+ task = asyncio.create_task(consumer.run())
+ yield
+ task.cancel()
+
+
+app = FastAPI(title=SERVICE_NAME, lifespan=lifespan)
+
+
+@app.get("/")
+async def root():
+ return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+ if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+ logger.info("received chatty-health healthz probe from gateway")
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ try:
+ async with engine.connect() as conn:
+ await conn.exec_driver_sql("SELECT 1")
+ await redis_client.ping()
+ except Exception as exc:
+ logger.exception(f"readyz check failed: {exc}")
+ return JSONResponse(status_code=503, content={"status": "unavailable"})
+ return {"status": "ok"}
diff --git a/services/notifier/chaos.py b/services/notifier/chaos.py
new file mode 100644
index 0000000..5d13a8c
--- /dev/null
+++ b/services/notifier/chaos.py
@@ -0,0 +1,8 @@
+"""slow_consumer -- phase-0-base-project-spec.md §7. Owned by `notifier`."""
+from common.chaos_flags import is_enabled
+
+SLEEP_SECONDS = 0.25
+
+
+async def slow_consumer_enabled() -> bool:
+ return await is_enabled("slow_consumer")
diff --git a/services/notifier/consumer.py b/services/notifier/consumer.py
new file mode 100644
index 0000000..eda5fcb
--- /dev/null
+++ b/services/notifier/consumer.py
@@ -0,0 +1,78 @@
+import asyncio
+import logging
+import random
+import uuid
+
+import sqlalchemy as sa
+
+from common.db import Session
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+
+logger = logging.getLogger("notifier")
+
+GROUP = "notifier"
+CONSUMER = "notifier-1"
+
+
+async def ensure_group():
+ try:
+ await redis_client.xgroup_create(settings.stream_name, GROUP, id="$", mkstream=True)
+ except Exception as exc:
+ if "BUSYGROUP" not in str(exc):
+ raise
+
+
+async def _handle(fields: dict):
+ if await chaos.slow_consumer_enabled():
+ await asyncio.sleep(chaos.SLEEP_SECONDS)
+
+ async with Session() as session:
+ await session.execute(
+ sa.text(
+ "INSERT INTO notifications (id, transaction_id, channel, status, attempts, sent_at) "
+ "VALUES (:id, :tx_id, :channel, 'sent', 1, now())"
+ ),
+ {
+ "id": str(uuid.uuid4()),
+ "tx_id": fields.get("transaction_id"),
+ "channel": random.choice(["email", "push", "sms"]),
+ },
+ )
+ await session.commit()
+
+
+async def run():
+ """Background consumer loop for the `notifier` consumer group -- a second,
+ independent group (`audit`) reads the same stream in services/audit/worker.py.
+ Redis Streams carry no automatic context propagation, deliberately."""
+ await ensure_group()
+ logger.info("notifier consumer started")
+ while True:
+ try:
+ resp = await redis_client.xreadgroup(
+ groupname=GROUP,
+ consumername=CONSUMER,
+ streams={settings.stream_name: ">"},
+ count=10,
+ block=2000,
+ )
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ logger.exception("notifier consumer read failed")
+ await asyncio.sleep(1)
+ continue
+
+ if not resp:
+ continue
+
+ for _, messages in resp:
+ for msg_id, fields in messages:
+ try:
+ await _handle(fields)
+ except Exception:
+ logger.exception(f"notifier failed to process message {msg_id}")
+ await redis_client.xack(settings.stream_name, GROUP, msg_id)
diff --git a/services/sanctions/__init__.py b/services/sanctions/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/sanctions/app.py b/services/sanctions/app.py
new file mode 100644
index 0000000..7961216
--- /dev/null
+++ b/services/sanctions/app.py
@@ -0,0 +1,63 @@
+import asyncio
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common.db import engine
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+
+from . import chaos
+from .repository import is_match
+
+SERVICE_NAME = "sanctions"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class LookupRequest(BaseModel):
+ holder_name: str
+ request_id: str = "unknown"
+
+
+@app.get("/")
+async def root():
+ return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+ if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+ logger.info(f"received chatty-health healthz probe from gateway")
+ return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+ try:
+ async with engine.connect() as conn:
+ await conn.exec_driver_sql("SELECT 1")
+ except Exception as exc:
+ logger.exception(f"readyz check failed: {exc}")
+ return JSONResponse(status_code=503, content={"status": "unavailable"})
+ return {"status": "ok"}
+
+
+@app.post("/internal/lookup")
+async def lookup(body: LookupRequest):
+ logger.info(f"lookup for holder_name={body.holder_name!r} request_id={body.request_id}")
+
+ if await chaos.sanctions_errors_enabled() and chaos.should_error():
+ logger.info(f"sanctions_errors chaos triggered for request_id={body.request_id}")
+ return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})
+
+ if await chaos.slow_sanctions_enabled():
+ delay = chaos.sample_delay_seconds()
+ logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s for request_id={body.request_id}")
+ await asyncio.sleep(delay)
+
+ match = await is_match(body.holder_name)
+ if match:
+ return {"matched": True, **match}
+ return {"matched": False, "country": None, "list_source": None}
diff --git a/services/sanctions/chaos.py b/services/sanctions/chaos.py
new file mode 100644
index 0000000..d7cfe9d
--- /dev/null
+++ b/services/sanctions/chaos.py
@@ -0,0 +1,37 @@
+"""slow_sanctions and sanctions_errors -- phase-0-base-project-spec.md §7.
+
+Both flags belong to `sanctions`, the fake third-party watchlist vendor.
+"""
+import math
+import random
+
+from common.chaos_flags import is_enabled
+
+_MEDIAN_SECONDS = 0.04
+_P99_SECONDS = 1.8
+_MU = math.log(_MEDIAN_SECONDS)
+_Z99 = 2.326347874
+_SIGMA = (math.log(_P99_SECONDS) - _MU) / _Z99
+_TIMEOUT_TRIGGER_PROBABILITY = 0.015
+_TIMEOUT_TRIGGER_SECONDS = 5.0
+_ERROR_PROBABILITY = 0.03
+
+
+async def slow_sanctions_enabled() -> bool:
+ return await is_enabled("slow_sanctions")
+
+
+async def sanctions_errors_enabled() -> bool:
+ return await is_enabled("sanctions_errors")
+
+
+def sample_delay_seconds() -> float:
+ """Log-normal delay: median ~40ms, p99 ~1.8s. 1.5% of calls sleep 5s,
+ which exceeds the caller's 3s downstream timeout and surfaces as one."""
+ if random.random() < _TIMEOUT_TRIGGER_PROBABILITY:
+ return _TIMEOUT_TRIGGER_SECONDS
+ return random.lognormvariate(_MU, _SIGMA)
+
+
+def should_error() -> bool:
+ return random.random() < _ERROR_PROBABILITY
diff --git a/services/sanctions/repository.py b/services/sanctions/repository.py
new file mode 100644
index 0000000..1cbe866
--- /dev/null
+++ b/services/sanctions/repository.py
@@ -0,0 +1,19 @@
+import sqlalchemy as sa
+
+from common.db import Session
+
+
+async def is_match(holder_name: str) -> dict | None:
+ """Case-insensitive exact match against sanctions_entries.normalized_name."""
+ async with Session() as session:
+ result = await session.execute(
+ sa.text(
+ "SELECT country, list_source FROM sanctions_entries "
+ "WHERE lower(normalized_name) = lower(:name) LIMIT 1"
+ ),
+ {"name": holder_name},
+ )
+ row = result.first()
+ if row is None:
+ return None
+ return {"country": row.country, "list_source": row.list_source}