Capstone: Food Delivery (Swiggy / DoorDash)

DAU              = 20_000_000
orders_per_day   = 5_000_000
restaurants      = 200_000
couriers_online  = 100_000          # at peak
SECONDS_PER_DAY  = 86_400

# --- Orders: surprisingly small ---
orders_avg = orders_per_day / SECONDS_PER_DAY                 # ~58/s
orders_peak = orders_avg * 10        # lunch + dinner spikes are brutal, not 3x
print(f"orders: {orders_avg:.0f}/s avg, {orders_peak:.0f}/s peak")

# --- Browse: 20-40x the order volume (people look, then leave) ---
sessions_per_day = DAU * 1.5
views_per_session = 20
browse_avg = sessions_per_day * views_per_session / SECONDS_PER_DAY
print(f"browse: {browse_avg:,.0f}/s avg, {browse_avg * 8:,.0f}/s peak")

# --- Courier pings: the real firehose ---
ping_every_seconds = 4
pings = couriers_online / ping_every_seconds
print(f"courier pings: {pings:,.0f}/s")

# --- Storage per year ---
order_row_bytes = 1_500              # order + items + denormalised snapshot
orders_gb_year = orders_per_day * 365 * order_row_bytes / 1e9
ping_bytes = 40                      # courier_id, lat, lng, ts, accuracy
pings_tb_year = pings * SECONDS_PER_DAY * 365 * ping_bytes / 1e12
print(f"orders: {orders_gb_year:,.0f} GB/year   raw pings: {pings_tb_year:,.1f} TB/year")

# --- Live tracking fan-out (WebSocket connections held) ---
in_flight = orders_per_day / SECONDS_PER_DAY * 30 * 60        # ~30 min per order
print(f"orders in flight at once: {in_flight:,.0f} -> ~{in_flight * 2 / 1000:,.0f}k sockets")

## Discovery (read-heavy, cacheable, eventually consistent)

GET /api/v1/feed?lat=12.97&lng=77.59&cursor=<opaque>
  -> 200 { sections: [{ title, restaurants: [RestaurantCard] }], next_cursor }
     Cache-Control: private, max-age=60
     # RestaurantCard is a denormalised read model: name, image, rating,
     # eta_minutes, price_band, is_open, promos. One document, zero joins.

GET /api/v1/search?q=biryani&lat=&lng=&filters=veg,rating_4plus&cursor=
  -> 200 { results: [RestaurantCard], facets: {...}, next_cursor }

GET /api/v1/restaurants/{id}/menu
  -> 200 { version: 412, categories: [{ name, items: [{ id, name, price_minor,
            currency, is_available, customisations: [...] }] }] }
     ETag: "menu-412"          # client revalidates cheaply with If-None-Match

## Cart and pricing (server authoritative — never trust a client price)

POST /api/v1/carts/{cart_id}/items      { item_id, qty, customisations[] }
  -> 200 Cart | 409 ITEM_UNAVAILABLE | 409 RESTAURANT_CLOSED

POST /api/v1/carts/{cart_id}/quote      { address_id, tip_minor, promo_code? }
  -> 200 { subtotal_minor, delivery_fee_minor, surge_fee_minor, taxes_minor,
           discount_minor, total_minor, eta_minutes, quote_id, expires_at }
  # quote_id is signed and short-lived (2 min). Order placement replays it,
  # so price cannot change between "Pay" and the charge.

## Order placement — the one endpoint that must be exactly-once

POST /api/v1/orders
  Headers: Idempotency-Key: 7f1c... (client-generated UUID, required)
  { cart_id, quote_id, address_id, payment_method_id, instructions? }
  -> 201 { order_id, state: "CREATED", eta_minutes, total_minor,
           payment: { status: "REQUIRES_ACTION", client_secret } }
     409 QUOTE_EXPIRED | 409 ITEM_UNAVAILABLE | 402 PAYMENT_DECLINED
     429 + Retry-After   | 503 during load shedding
  # A retry with the same Idempotency-Key returns the SAME 201 body.
  # It never creates a second order and never charges twice.

GET  /api/v1/orders/{id}            -> 200 Order (full state + courier summary)
GET  /api/v1/orders/{id}/events     -> 200 { events: [...] }   # audit timeline
POST /api/v1/orders/{id}/cancel     { reason } -> 200 | 409 TOO_LATE_TO_CANCEL

## Live tracking — push, never poll at 200k clients

WS /api/v1/orders/{id}/track        # subscribe; server pushes:
  { type: "state", state: "PICKED_UP", at: "..." }
  { type: "courier_position", lat, lng, bearing, eta_minutes }
  # Fallback for restricted networks: SSE, then long-poll every 5s.

## Restaurant-facing

WS   /api/v1/partner/orders/stream          # new orders pushed to the tablet
POST /api/v1/partner/orders/{id}/accept     { prep_minutes } -> 200 | 409
POST /api/v1/partner/orders/{id}/reject     { reason }       -> 200 | 409
POST /api/v1/partner/orders/{id}/ready      -> 200 | 409
PATCH /api/v1/partner/menu/items/{id}       { is_available } -> 200

## Courier-facing

POST /api/v1/courier/location       { lat, lng, bearing, accuracy_m, ts }
  -> 204     # batched: the app sends up to 5 samples per call, fire-and-forget
POST /api/v1/courier/offers/{offer_id}/accept  -> 200 { order } | 409 TAKEN
POST /api/v1/courier/orders/{id}/picked_up     -> 200 | 409
POST /api/v1/courier/orders/{id}/delivered     { proof? } -> 200 | 409

-- =========================================================
-- CATALOGUE  (Postgres, source of truth; read path is cached)
-- =========================================================
restaurants (
  id            BIGINT PRIMARY KEY,
  name          TEXT NOT NULL,
  city_id       INT NOT NULL,          -- the natural shard / partition key
  lat           DOUBLE PRECISION NOT NULL,
  lng           DOUBLE PRECISION NOT NULL,
  geohash6      CHAR(6) NOT NULL,      -- ~1.2 km cell, for coarse proximity
  status        SMALLINT NOT NULL,     -- 0 offline, 1 live, 2 paused
  prep_minutes  SMALLINT NOT NULL,
  rating_avg    NUMERIC(2,1),          -- denormalised rollup, eventually consistent
  rating_count  INT,
  menu_version  INT NOT NULL           -- bump on any menu change -> cache key
);
CREATE INDEX ON restaurants (city_id, status);
CREATE INDEX ON restaurants (geohash6) WHERE status = 1;

menu_items (
  id            BIGINT PRIMARY KEY,
  restaurant_id BIGINT NOT NULL REFERENCES restaurants(id),
  category      TEXT NOT NULL,
  name          TEXT NOT NULL,
  price_minor   INT NOT NULL,
  is_available  BOOLEAN NOT NULL DEFAULT TRUE,   -- toggled hundreds of times/day
  tags          TEXT[]                            -- veg, spicy, bestseller
);
CREATE INDEX ON menu_items (restaurant_id) INCLUDE (name, price_minor);

-- =========================================================
-- ORDERS  (Postgres, strongly consistent, the money path)
-- =========================================================
orders (
  id               BIGINT PRIMARY KEY,          -- Snowflake: time-sortable (lesson 7)
  customer_id      BIGINT NOT NULL,
  restaurant_id    BIGINT NOT NULL,
  city_id          INT NOT NULL,                -- shard key when we shard
  courier_id       BIGINT NULL,
  state            TEXT NOT NULL,               -- see the state machine below
  version          INT NOT NULL DEFAULT 0,      -- optimistic concurrency control
  -- money, captured at order time and never recomputed:
  subtotal_minor   INT NOT NULL,
  delivery_fee_minor INT NOT NULL,
  surge_fee_minor  INT NOT NULL DEFAULT 0,
  taxes_minor      INT NOT NULL,
  discount_minor   INT NOT NULL DEFAULT 0,
  total_minor      INT NOT NULL,
  payment_id       TEXT NULL,
  delivery_address JSONB NOT NULL,              -- SNAPSHOT, not a foreign key:
  placed_at        TIMESTAMPTZ NOT NULL,        -- editing your address later must
  promised_at      TIMESTAMPTZ NOT NULL,        -- not rewrite last month's order
  delivered_at     TIMESTAMPTZ NULL
);
CREATE INDEX ON orders (customer_id, placed_at DESC);   -- "my orders"
CREATE INDEX ON orders (restaurant_id, state);           -- partner dashboard
CREATE INDEX ON orders (state) WHERE state IN ('PAID','ACCEPTED','PREPARING','READY');

order_items (                                   -- price snapshot per line
  order_id      BIGINT NOT NULL REFERENCES orders(id),
  menu_item_id  BIGINT NOT NULL,
  name_snapshot TEXT NOT NULL,
  qty           SMALLINT NOT NULL,
  unit_price_minor INT NOT NULL,
  customisations JSONB,
  PRIMARY KEY (order_id, menu_item_id, customisations)
);

order_events (                                  -- append-only audit log
  order_id   BIGINT NOT NULL,
  seq        INT NOT NULL,
  from_state TEXT, to_state TEXT NOT NULL,
  actor      TEXT NOT NULL,                     -- customer | restaurant | courier | system
  payload    JSONB, at TIMESTAMPTZ NOT NULL,
  PRIMARY KEY (order_id, seq)
);

idempotency_keys (                              -- the anti-double-charge table
  key          TEXT PRIMARY KEY,
  user_id      BIGINT NOT NULL,
  request_hash TEXT NOT NULL,                   -- reject key reuse with a new body
  state        TEXT NOT NULL,                   -- IN_PROGRESS | DONE
  response     JSONB NULL,
  created_at   TIMESTAMPTZ NOT NULL             -- TTL sweep after 24-48h
);

outbox (                                        -- transactional outbox pattern
  id BIGSERIAL PRIMARY KEY, topic TEXT NOT NULL,
  key TEXT NOT NULL, payload JSONB NOT NULL,
  created_at TIMESTAMPTZ NOT NULL, published_at TIMESTAMPTZ NULL
);
CREATE INDEX ON outbox (id) WHERE published_at IS NULL;

-- =========================================================
-- LIVE STATE  (Redis — ephemeral, high churn, never the source of truth)
-- =========================================================
GEOADD couriers:city:42            <lng> <lat> courier:9001   -- sorted-set geo index
HSET   courier:9001                state ONLINE order_id 0 battery 61 ts 1737
SETEX  track:order:88231           30  '{"lat":..,"lng":..,"eta":7}'
SETEX  menu:rest:1234:v412         600 '<serialised menu json>'
SETEX  feed:geo:tdr1y:v3           60  '<serialised feed page>'
INCR   inv:rest:1234:item:77                                  -- stock counters
SET    offer:lock:88231            courier:9001 NX EX 25      -- dispatch lock

-- =========================================================
-- SEARCH (Elasticsearch, fed by CDC — eventually consistent, ~1s behind)
-- =========================================================
doc: { restaurant_id, name, city_id, geo_point, cuisines[], dish_names[],
       rating_avg, is_open, price_band, popularity_score }

-- =========================================================
-- ANALYTICS (Kafka -> columnar warehouse; downsampled location history)
-- =========================================================
topics: order.events, courier.pings, search.queries, payment.events

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum

class State(str, Enum):
    CREATED = "CREATED";     PAID = "PAID";           ACCEPTED = "ACCEPTED"
    PREPARING = "PREPARING"; READY = "READY";         PICKED_UP = "PICKED_UP"
    DELIVERED = "DELIVERED"; CANCELLED = "CANCELLED"; REFUNDED = "REFUNDED"

# One table is the whole lifecycle rule set. A new rule edits this, not ten methods.
ALLOWED: dict[State, set[State]] = {
    State.CREATED:   {State.PAID, State.CANCELLED},
    State.PAID:      {State.ACCEPTED, State.REFUNDED, State.CANCELLED},
    State.ACCEPTED:  {State.PREPARING, State.REFUNDED},
    State.PREPARING: {State.READY, State.REFUNDED},
    State.READY:     {State.PICKED_UP},
    State.PICKED_UP: {State.DELIVERED},
    State.DELIVERED: set(), State.CANCELLED: set(), State.REFUNDED: set(),
}
# Who may cause which transition — authorisation is data, not scattered checks.
ACTOR_RIGHTS = {
    "customer":   {State.CANCELLED},
    "restaurant": {State.ACCEPTED, State.PREPARING, State.READY, State.REFUNDED},
    "courier":    {State.PICKED_UP, State.DELIVERED},
    "system":     set(State),
}

class IllegalTransition(Exception): pass
class Unauthorised(Exception): pass
class ConcurrentUpdate(Exception): pass
class DuplicateRequest(Exception):
    def __init__(self, stored_response): self.stored_response = stored_response

@dataclass
class Order:
    id: int
    customer_id: int
    restaurant_id: int
    city_id: int
    total_minor: int
    state: State = State.CREATED
    version: int = 0
    courier_id: int | None = None
    events: list[tuple] = field(default_factory=list)

    def can_transition(self, target: State, actor: str) -> None:
        if target not in ACTOR_RIGHTS.get(actor, set()):
            raise Unauthorised(f"{actor} may not set {target}")
        if target not in ALLOWED[self.state]:
            raise IllegalTransition(f"order {self.id}: {self.state} -> {target}")

# ---- ports (interfaces). The domain never mentions SQL, Redis or Stripe. ----

class OrderRepository(ABC):
    @abstractmethod
    def claim_idempotency_key(self, key: str, request_hash: str) -> dict | None:
        """Returns None if claimed by us; the stored response if it is a retry."""
    @abstractmethod
    def save_new(self, order: Order, outbox_event: dict) -> None:
        """ONE transaction: order + items + stock reservation + outbox row."""
    @abstractmethod
    def compare_and_set_state(self, order_id: int, expect: State,
                              target: State, expect_version: int) -> bool:
        """UPDATE ... WHERE id=? AND state=? AND version=?  -> rowcount == 1"""
    @abstractmethod
    def finish_idempotency(self, key: str, response: dict) -> None: ...

class PaymentGateway(ABC):
    @abstractmethod
    def authorise(self, order_id: int, amount_minor: int, method_id: str,
                  idempotency_key: str) -> str: ...
    @abstractmethod
    def void(self, payment_id: str) -> None: ...

class QuoteVerifier(ABC):
    @abstractmethod
    def verify(self, quote_id: str, cart_id: str) -> dict: ...

# ---- the application service ----

class OrderService:
    def __init__(self, repo: OrderRepository, payments: PaymentGateway,
                 quotes: QuoteVerifier, ids):
        self._repo, self._payments, self._quotes, self._ids = repo, payments, quotes, ids

    def place(self, cmd: dict, idempotency_key: str) -> dict:
        # 1. Exactly-once, before anything with a side effect.
        stored = self._repo.claim_idempotency_key(idempotency_key, hash_of(cmd))
        if stored is not None:
            raise DuplicateRequest(stored)          # handler returns 201 + this body

        # 2. Price is the server's, replayed from a signed quote.
        quote = self._quotes.verify(cmd["quote_id"], cmd["cart_id"])

        order = Order(id=self._ids.next(),          # Snowflake: sortable, shard-friendly
                      customer_id=cmd["customer_id"],
                      restaurant_id=quote["restaurant_id"],
                      city_id=quote["city_id"],
                      total_minor=quote["total_minor"])

        # 3. Order row, stock reservation and outbox event commit together or not at all.
        self._repo.save_new(order, outbox_event={
            "topic": "order.created", "key": str(order.id),
            "payload": {"order_id": order.id, "restaurant_id": order.restaurant_id},
        })

        # 4. Money last, and idempotent at the PSP too.
        try:
            payment_id = self._payments.authorise(
                order.id, order.total_minor, cmd["payment_method_id"], idempotency_key)
        except Exception:
            self.transition(order, State.CANCELLED, actor="system",
                            reason="payment_failed")   # releases the stock reservation
            raise

        # 5. Conditional write: a retried worker cannot pay twice or skip a state.
        if not self._repo.compare_and_set_state(order.id, State.CREATED,
                                                State.PAID, order.version):
            self._payments.void(payment_id)            # someone else moved it; undo
            raise ConcurrentUpdate(order.id)
        order.state, order.version = State.PAID, order.version + 1

        response = {"order_id": order.id, "state": order.state,
                    "total_minor": order.total_minor}
        self._repo.finish_idempotency(idempotency_key, response)
        return response

    def transition(self, order: Order, target: State, actor: str, **meta) -> Order:
        order.can_transition(target, actor)            # pure, unit-testable, no I/O
        if not self._repo.compare_and_set_state(order.id, order.state,
                                                target, order.version):
            raise ConcurrentUpdate(order.id)           # lost the race: re-read and retry
        order.events.append((order.state, target, actor, meta))
        order.state, order.version = target, order.version + 1
        return order


def hash_of(cmd: dict) -> str:
    import hashlib, json
    return hashlib.sha256(json.dumps(cmd, sort_keys=True).encode()).hexdigest()

# Demo with the pure domain object only — no database needed, which is the point.
o = Order(id=88231, customer_id=7, restaurant_id=1234, city_id=42, total_minor=48950)
print(o.state)
for actor, target in (("system", State.PAID), ("restaurant", State.ACCEPTED),
                      ("restaurant", State.PREPARING), ("restaurant", State.READY)):
    o.can_transition(target, actor); o.state = target
print(o.state)
for actor, target in (("customer", State.CANCELLED), ("courier", State.DELIVERED)):
    try:
        o.can_transition(target, actor)
    except (IllegalTransition, Unauthorised) as e:
        print(type(e).__name__, "-", e)

from dataclasses import dataclass

@dataclass
class Courier:
    id: int; distance_km: float; rating: float
    batch_compatible: bool      # already heading to the same area
    idle_minutes: float         # fairness: nudge work to whoever has waited longest
    accept_rate: float          # likelihood they take the offer

def score(c: Courier, weights: dict) -> float:
    """Higher is better. Every term is a product decision, so keep them named
    and weighted rather than buried in one expression."""
    return (
        weights["proximity"] * (1.0 / (1.0 + c.distance_km))
        + weights["quality"]   * (c.rating / 5.0)
        + weights["batching"]  * (1.0 if c.batch_compatible else 0.0)
        + weights["fairness"]  * min(c.idle_minutes / 30.0, 1.0)
        + weights["accept"]    * c.accept_rate
    )

WEIGHTS = {"proximity": 0.45, "quality": 0.10, "batching": 0.20,
           "fairness": 0.15, "accept": 0.10}

candidates = [
    Courier(901, 0.4, 4.9, False, 2,  0.95),
    Courier(902, 1.1, 4.6, True,  12, 0.80),
    Courier(903, 2.6, 4.8, False, 28, 0.60),
]
for c in sorted(candidates, key=lambda c: -score(c, WEIGHTS)):
    print(c.id, round(score(c, WEIGHTS), 3))