A production-grade microservices platform connecting Ugandan farmers and buyers. Farmers list produce, buyers place orders, and the system handles payments, messaging, notifications, and price intelligence — all through a single Nginx API gateway.

The platform is split into two independent but integrated stacks:

• Core stack — transactional services (auth, users, produce, orders, payments, messaging, blog, USSD)
• ML stack — price prediction and farmer/buyer matching (services/soko-ml/)

1. Architecture Overview
2. Core Services
3. ML Layer
4. How the Two Stacks Interact
5. API Reference
6. User Flows
7. Event System
8. Getting the ML Stack Running
9. Running Tests
10. Environment Variables
11. Project Structure

 ┌────────────────────────────────────────────────────────────────┐
 │                      CLIENT LAYER                              │
 │          Web App · Mobile App · USSD Handsets                  │
 └───────────────────────────┬────────────────────────────────────┘
                             │ HTTP / WebSocket
                             ▼
 ┌────────────────────────────────────────────────────────────────┐
 │               NGINX API GATEWAY  :80                           │
 │   Rate limiting (30 req/min) · CORS · JWT subrequest auth      │
 │   Routes: /auth/ /users/ /listings/ /orders/ /payments/        │
 │           /message/ /notifications/ /posts/ /ussd/             │
 │           /recommendations/                                     │
 └──┬────┬────┬────┬────┬────┬────┬────┬────┬────────────────────┘
    │    │    │    │    │    │    │    │    │
    ▼    ▼    ▼    ▼    ▼    ▼    ▼    ▼    ▼
  Auth User Prod Ord  Pay  Msg  Notif Blog USSD  Rec
 :8001:8002:8003:8004:8005:8006:8007 :8008:8009 :8010
    │    │    │    │    │    │    │    │    │     │
    └────┴────┴────┴────┴────┴────┴────┴────┴─────┘
                         │
              ┌──────────┴──────────┐
              │                     │
              ▼                     ▼
        PostgreSQL              RabbitMQ :5672
     (one DB per service)     (async events between
                               core services)

 ┌────────────────────────────────────────────────────────────────┐
 │                      ML STACK  (services/soko-ml/)             │
 │                                                                │
 │   ml-gateway-service :8000  ←  called by core services        │
 │        │                   ←  single entry point              │
 │        ├──► price-prediction-service :8001                     │
 │        │         Prophet .pkl models · Redis cache 24h         │
 │        └──► recommendation-service :8002                       │
 │                  Content scoring · Redis cache 1h              │
 │                                                                │
 │   kafka-agent  (no HTTP port)                                  │
 │        ├── consumes: soko.transactions                         │
 │        ├── consumes: soko.interactions                         │
 │        ├── consumes: soko.price.requests                       │
 │        └── produces: soko.price.results · soko.dlq             │
 │                                                                │
 │   Infrastructure: Kafka · Zookeeper · Redis                    │
 └────────────────────────────────────────────────────────────────┘

• Every external request enters through Nginx only — services are never exposed directly.
• Every call to the ML layer enters through ml-gateway-service only — price and recommendation services are never called directly by core services.
• Auth is enforced at the gateway via an internal /verify-token subrequest to the Auth service before the request reaches any protected service.
• Core services communicate asynchronously via RabbitMQ. The ML layer uses Kafka for its own event backbone.

Responsibility: Identity and access. Issues JWTs on login, exposes /verify-token which Nginx calls internally on every protected route to validate tokens and inject X-User-Id, X-User-Role, X-User-Email headers downstream.

Nginx route: /auth/ and /oauth/ (public — no auth guard)

Responsibility: User profiles and account management. Receives the authenticated user context (X-User-Id, X-User-Role) from Nginx — it never validates tokens itself.

Nginx route: /users/ (JWT required)

Responsibility: Produce listings — creation, search, stock management. Farmers create listings; buyers browse them. Publishes produce.listed events to RabbitMQ so the recommendation service can index new listings.

Nginx route: /listings/ (JWT required, 20 MB upload limit for images)

Responsibility: Order lifecycle from placement to completion. Buyers place orders against listings; farmers accept or reject; status advances through a defined state machine.

Nginx route: /orders/ (JWT required)

Order state machine:

placed → pending
          ├─► confirmed  (farmer accepts)
          │       └─► completed  (farmer marks done → review unlocked)
          ├─► rejected   (farmer declines)
          └─► cancelled  (buyer withdraws)

Responsibility: Payment initiation and reconciliation via PesaPal (MTN Mobile Money / Airtel Money). The /webhook/ endpoint is public so PesaPal can POST payment confirmations without authentication.

Nginx routes: /payments/ (JWT required) · /webhook/ (public)

Responsibility: Real-time direct messaging between farmers and buyers over WebSocket. The WebSocket upgrade (/message/ws/) bypasses the JWT subrequest — token is validated by the service itself on connection.

Nginx routes: /message/ (JWT required) · /message/ws/ (WebSocket, service-auth)

Responsibility: Push notifications delivered in real-time over WebSocket. Consumes events from RabbitMQ (order updates, payment confirmations) and pushes them to connected clients.

Nginx routes: /notifications/ (JWT required) · /notifications/ws/ (WebSocket, service-auth)

Responsibility: Agri-knowledge articles and market commentary. Supports image uploads up to 10 MB.

Nginx route: /posts/ (JWT required, 10 MB upload limit)

Responsibility: USSD session handler for feature-phone users. Completely public — USSD networks don't carry HTTP auth headers. Allows farmers with basic handsets to check prices and receive order notifications.

Nginx route: /ussd/ (public — no auth guard)

Responsibility: Personalised produce feed for buyers based on order history, category preferences, and produce quality scores. Consumes produce.listed, order.placed, and quality.scored events from RabbitMQ to keep its index fresh. This is the existing rule-based recommendation service, separate from the ML farmer/buyer matching layer.

Nginx route: /recommendations/ (public)

The ML layer lives in services/soko-ml/ and runs as a separate Docker Compose stack. It has four services of its own, plus Kafka and a dedicated Redis instance.

The single entry point for all ML capabilities. No core service should ever call the price or recommendation ML services directly — they call this gateway, which adds:

• Request logging — service name, endpoint, latency, cache hit/miss
• Circuit breaking — if a downstream ML service is unreachable after 3 retries, returns a graceful fallback response instead of propagating a 500
• Health aggregation — GET /health polls all downstream services and returns a combined status

Serves 4-week price forecasts per market–crop pair in UGX using pre-trained Prophet models.

• Loads .pkl model files from models/ at startup (one model per market–crop pair, 48 total)
• Checks Redis on every request (price:v1:{market}:{crop}:{weeks}, TTL 24 h)
• Falls back to Uganda seasonal heuristics if no model file is present (always responds)
• Publishes a price.predicted event to soko.price.results after every inference
• Consumes soko.price.requests for async batch prediction jobs

Supported markets: Kisenyi_Kampala · Gulu · Mbarara · Mbale · Lira · Masaka

Supported crops: maize_grain · yellow_beans · irish_potatoes · tomatoes · matoke · cassava_chips · sorghum · millet

Recommends high-performing farmers to buyers and vice versa using a weighted content-based scoring model enriched in real-time from Kafka interaction events.

• Loads farmers.csv (200 profiles) and buyers.csv (300 profiles) at startup
• Scores farmer–buyer compatibility on crop overlap, market overlap, rating, fulfillment rate
• Boosts scores dynamically from Kafka events: view +0.02, inquiry +0.05, purchase +0.10, rating +0.04
• Caches results in Redis (rec:farmers:{buyer_id}:{top_n}, TTL 1 h)
• Invalidates cache when a relevant interaction event arrives

Scoring weights — farmers for buyer:

Scoring weights — buyers for farmer:

Long-running Python process — no HTTP port. Bridges the core Soko event stream with the ML layer.

Failed messages go to soko.dlq with full error context.

Kafka topics:

Redis cache keys:

Core Soko stack                         ML stack
─────────────────                       ─────────────────────────────────────
                                        ml-gateway-service :8000
recommendation_service :8010 ──HTTP──►  GET /recommend/farmers-for-buyer/
                                        GET /recommend/buyers-for-farmer/

produce_service :8003        ──HTTP──►  POST /price/predict
                                        (surface price context on listing pages)

order_service :8004          ──Kafka──► soko.transactions
                                        (kafka-agent listens, enriches to
                                         soko.interactions for rec boosts)

ussd_service :8009           ──HTTP──►  POST /price/predict
                                        (price checks on feature phones)

The ML stack is intentionally decoupled — the core stack calls ml-gateway-service over HTTP and publishes to Kafka topics. The ML layer never calls back into the core stack.

All requests enter via http://localhost:80 through Nginx. Protected routes require an Authorization: Bearer <token> header.

POST /auth/register         { "email": "...", "password": "...", "role": "farmer|buyer" }
POST /auth/login            { "email": "...", "password": "..." }  →  { "access_token": "..." }
GET  /auth/me               Authorization: Bearer <token>
POST /auth/refresh          Authorization: Bearer <token>

GET  /listings/             ?category=grains&district=Kampala&min_price=500&max_price=2000
POST /listings/             { "title", "category", "price_per_kg", "quantity_kg", "district" }
GET  /listings/{id}
PATCH /listings/{id}
DELETE /listings/{id}

POST  /orders/              { "listing_id": "...", "quantity_kg": 100 }
GET   /orders/
PATCH /orders/{id}/cancel
POST  /orders/{id}/review   { "rating": 5, "comment": "..." }
GET   /orders/incoming/                                (farmer)
PATCH /orders/{id}/status   { "new_status": "confirmed|completed|rejected" }  (farmer)

POST /payments/initiate     { "order_id": "...", "phone": "256700000000" }
GET  /payments/{id}/status
POST /webhook/pesapal       (PesaPal callback — public)

GET  /message/              List conversations
POST /message/              { "recipient_id": "...", "body": "..." }
WS   /message/ws/{token}    Real-time message stream

GET  /notifications/        List notifications
WS   /notifications/ws/{token}  Real-time push stream

POST http://localhost:8000/price/predict
     { "market": "Kisenyi_Kampala", "crop": "maize_grain", "weeks_ahead": 4 }

GET  http://localhost:8000/recommend/farmers-for-buyer/B0001?top_n=5
GET  http://localhost:8000/recommend/buyers-for-farmer/F0001?top_n=5
GET  http://localhost:8000/health
GET  http://localhost:8000/price/markets
GET  http://localhost:8000/price/crops

1. POST /auth/register  { role: "farmer" }
2. POST /auth/login     → JWT
3. POST /listings/      List produce with price and quantity
4. GET  /orders/incoming/    See buyer orders
5. PATCH /orders/{id}/status  { "new_status": "confirmed" }
6. PATCH /orders/{id}/status  { "new_status": "completed" }

1. POST /auth/register  { role: "buyer" }
2. POST /auth/login     → JWT
3. GET  /listings/      Browse produce (filter by district, crop, price)
4. POST /orders/        Place order
5. POST /payments/initiate   Pay via Mobile Money
6. POST /orders/{id}/review  Rate after completion
7. GET  /recommendations/    See personalised feed

1. Farmer dials USSD short code
2. ussd_service calls ml-gateway-service POST /price/predict
3. 4-week maize price forecast returned as plain text to handset

All queues are durable. Payload schemas are in CONTRACTS.md.

All commands run from the project root. Prerequisites: Docker 20+, Python 3.11, Make.

make install

Creates a .venv inside each ML service folder and installs its requirements.txt. Prophet pulls in pystan — expect 3–5 minutes on first run.

make generate-data

Writes three files to services/soko-ml/recommendation-service/data/raw/:

• crop_prices_raw.csv — 4 years of weekly UGX prices, 6 markets × 8 crops (~12,000 rows)
• farmers.csv — 200 synthetic farmer profiles with crop/market coverage, rating, fulfillment rate
• buyers.csv — 300 synthetic buyer profiles with preferred crops, markets, payment reliability

Verify the output:

wc -l services/soko-ml/recommendation-service/data/raw/*.csv
# expect: 12289 crop_prices_raw.csv  |  201 farmers.csv  |  301 buyers.csv

make train

Trains 48 Prophet models (6 markets × 8 crops) with Uganda bimodal seasonality and saves .pkl files to services/soko-ml/price-prediction-service/models/. Takes 5–15 minutes depending on CPU.

You can skip this step. The price-prediction-service has a built-in seasonal fallback that always returns a valid forecast. Skip make train for a faster first boot and run it later in the background.

Verify:

ls services/soko-ml/price-prediction-service/models/ | wc -l
# expect: 48

make up

Builds and starts 8 containers: Zookeeper, Kafka, kafka-init (topic creation), Redis, price-prediction-service, recommendation-service, ml-gateway-service, kafka-agent.

Watch the startup log until you see these three lines, then proceed:

make logs
# Look for:
# soko-ml-kafka-init  | All Kafka topics created.
# soko-ml-rec         | {"event": "recommendation_service_started", "farmers": 200, "buyers": 300}
# soko-ml-gateway     | {"event": "gateway_started"}

Kafka takes ~30 seconds to elect a leader — the ML services will retry automatically.

make health

All three services must return "ok" before proceeding:

=== ML Gateway ===
{
    "gateway": "ok",
    "services": { "price-prediction": "ok", "recommendation": "ok" },
    "circuit_breakers": { "price-prediction": "closed", "recommendation": "closed" }
}
=== Price Service ===
{ "status": "ok", "service": "price-prediction-service", "models_loaded": 48 }
=== Recommendation Service ===
{ "status": "ok", "service": "recommendation-service", "farmers_loaded": 200, "buyers_loaded": 300 }

make test

Runs pytest across all three FastAPI services — pure logic tests, no Redis or Kafka needed:

make test-price    # 11 tests — fallback predict, base UGX prices, model registry
make test-rec      # 14 tests — scoring, ranking, interaction boosts, cache invalidation
make test-gateway  # 11 tests — proxy, circuit breaker, health aggregation

make smoke-test

Fires three live HTTP calls through the gateway and prints the JSON. On the second run, the price response returns "cached": true — confirming Redis is working.

All targets run from the project root.

Start the core stack first:

docker compose up --build -d

Then run the integration suite (hits real services, no mocks):

pip install pytest httpx
pytest tests/integration/ -v

Covers: health checks, auth, user profiles, produce listings, order placement, stock reduction, reviews, recommendation event propagation.

No Docker required — runs against local code:

make install   # only needed once
make test

soko/
├── Makefile                          ← All ML commands (run from here)
├── docker-compose.yml                ← Core Soko stack
├── nginx/
│   └── nginx.conf                    ← API gateway routing + auth subrequests
├── services/
│   ├── auth/                         ← JWT auth, /verify-token
│   ├── user/                         ← User profiles          :8002
│   ├── produce/                      ← Listings, stock        :8003
│   ├── order/                        ← Orders, reviews        :8004
│   ├── payment/                      ← PesaPal integration    :8005
│   ├── message/                      ← WebSocket messaging    :8006
│   ├── notification/                 ← WebSocket push         :8007
│   ├── blog/                         ← Articles               :8008
│   ├── ussd/                         ← Feature-phone access   :8009
│   ├── recommendation/               ← RabbitMQ-driven feed   :8010
│   └── soko-ml/                      ← ML stack (own compose)
│       ├── docker-compose.yml
│       ├── docker-compose.dev.yml
│       ├── .env.example
│       ├── shared/
│       │   └── events.py             ← Kafka event dataclasses
│       ├── price-prediction-service/ ← Prophet + Redis + Kafka :8001
│       │   ├── src/
│       │   │   ├── main.py
│       │   │   ├── predictor.py
│       │   │   ├── cache.py
│       │   │   ├── kafka_producer.py
│       │   │   └── schemas.py
│       │   ├── models/               ← .pkl files (gitignored)
│       │   └── tests/
│       ├── recommendation-service/   ← Content scoring + Redis :8002
│       │   ├── src/
│       │   │   ├── main.py
│       │   │   ├── recommender.py
│       │   │   ├── interaction_store.py
│       │   │   ├── cache.py
│       │   │   ├── kafka_consumer.py
│       │   │   └── schemas.py
│       │   ├── data/raw/             ← farmers.csv, buyers.csv
│       │   └── tests/
│       ├── ml-gateway-service/       ← Proxy + circuit breaker :8000
│       │   ├── src/
│       │   │   ├── main.py
│       │   │   ├── proxy.py
│       │   │   └── logger.py
│       │   └── tests/
│       ├── kafka-agent/              ← Event backbone (no HTTP)
│       │   ├── src/
│       │   │   ├── agent.py
│       │   │   ├── consumers/
│       │   │   ├── producers/
│       │   │   └── dlq.py
│       │   └── tests/
│       └── data-generator/           ← One-shot CSV + model data
│           ├── generate_prices.py
│           └── generate_profiles.py
└── tests/
    └── integration/                  ← Core stack integration tests

Each core service follows the same internal layout:

service/
├── Dockerfile
├── requirements.txt
└── app/
    ├── main.py          ← FastAPI app + lifespan
    ├── config.py        ← pydantic-settings
    ├── database.py      ← SQLAlchemy engine
    ├── dependencies.py  ← JWT auth
    ├── messaging.py     ← RabbitMQ publisher / consumer
    ├── schemas.py       ← Pydantic models
    ├── models/          ← SQLAlchemy ORM
    └── routers/         ← Route handlers

• Alembic not wired — schema changes require dropping the affected DB volume
• Shared JWT secret — all services share one key; use a secrets manager in production
• /listings/{id}/reduce-stock is unauthenticated — secure with an internal API key in production
• No password reset — requires an email provider
• ML stack is a separate compose — it does not share the core stack's network or Redis; the two stacks communicate over localhost ports in development and would use a shared Docker network or service mesh in production