IRCTC Tatkal Ticket Booking System.

# inventory/hot_pool_admission.py

"""
Fast admission control for a single hot (train, date, class, quota) pool.

Design principle: the pool counter must answer ONE question, as fast as
possible, for as many callers as possible. It does not allocate berths,
does not touch passenger data, and does not talk to a durable database.
"""

from dataclasses import dataclass
from typing import Optional
import time


ADMISSION_SCRIPT = """
-- KEYS[1]: pool counter key, e.g. "pool:12301:2026-08-10:3A:TQ"
-- ARGV[1]: this request's unique attempt id (for idempotent retries)

-- Idempotent retry: this attempt was already admitted, so do NOT
-- decrement again (a retried request must not burn a second seat)
if redis.call('SISMEMBER', KEYS[1] .. ':admitted', ARGV[1]) == 1 then
    return -3
end

local remaining = tonumber(redis.call('GET', KEYS[1]))
if remaining == nil then
    return -2  -- pool not initialized / booking not open yet
end

if remaining <= 0 then
    return -1  -- sold out, reject immediately
end

redis.call('DECR', KEYS[1])
redis.call('SADD', KEYS[1] .. ':admitted', ARGV[1])
return remaining - 1
"""


@dataclass
class AdmissionResult:
    admitted: bool
    reason: str
    remaining_hint: int = 0


class HotPoolAdmissionGate:
    """
    Guards a single (train, date, class, quota) pool from the full
    force of Tatkal traffic. This is the ONLY thing 2 million
    requests are allowed to touch directly.
    """

    def __init__(self, redis_client, metrics):
        self.redis = redis_client
        self.metrics = metrics
        self._script_sha = None

    async def initialize_pool(self, pool_key: str, seat_count: int):
        """Called once, moments before the Tatkal window opens."""
        self._script_sha = await self.redis.script_load(ADMISSION_SCRIPT)
        await self.redis.set(pool_key, seat_count, nx=True)

    async def try_admit(self, pool_key: str, attempt_id: str) -> AdmissionResult:
        """
        The ONLY call 2 million requests make. Must resolve in
        single-digit milliseconds under load.
        """
        result = await self.redis.evalsha(
            self._script_sha, 1, pool_key, attempt_id
        )

        if result == -2:
            self.metrics.increment("admission_not_open", pool_key)
            return AdmissionResult(False, "BOOKING_NOT_OPEN")

        if result == -1:
            self.metrics.increment("admission_sold_out", pool_key)
            return AdmissionResult(False, "SOLD_OUT")

        if result == -3:
            # Same attempt retried: it already holds its ticket to try
            return AdmissionResult(True, "ALREADY_ADMITTED")

        self.metrics.increment("admission_granted", pool_key)
        return AdmissionResult(True, "PROCEED_TO_ALLOCATION", remaining_hint=result)

# waitingroom/queue_service.py

"""
Virtual Waiting Room: absorbs the thundering herd before it ever
reaches the booking core, and admits requests at a rate the
downstream system has actually signaled it can handle.
"""

from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
import hashlib


@dataclass
class DownstreamHealth:
    """Live signal from the booking core, refreshed every second."""
    db_latency_ms: float
    payment_success_rate: float
    booking_core_cpu_pct: float


@dataclass
class QueueTicket:
    session_id: str
    bucket: int          # coarse time-bucket for fair shuffling
    shuffle_key: str      # deterministic but unpredictable ordering
    issued_at: datetime


class AdaptiveWaitingRoom:
    """
    Buffers arrivals and admits them to the booking core at a
    rate derived from real downstream health, not a static guess.
    """

    BASE_ADMIT_RATE = 1000      # requests/sec under healthy conditions
    MIN_ADMIT_RATE = 100        # never fully stop; keep some flow moving
    BUCKET_WIDTH_MS = 250

    def __init__(self, queue_store, session_binder, metrics):
        self.queue = queue_store          # ordered structure (e.g. Redis sorted set)
        self.sessions = session_binder    # enforces one active slot per user
        self.metrics = metrics

    async def enqueue(self, session_id: str) -> QueueTicket:
        """
        Called once per user session as they arrive. Idempotent:
        calling twice for the same active session returns the
        SAME ticket, never a second slot.
        """
        existing = await self.sessions.get_active_ticket(session_id)
        if existing:
            return existing

        now = datetime.utcnow()
        bucket = int(now.timestamp() * 1000) // self.BUCKET_WIDTH_MS

        # Shuffle key: hash of session + bucket, NOT arrival timestamp.
        # This defeats "fastest script wins" while staying deterministic.
        shuffle_key = hashlib.sha256(
            f"{session_id}:{bucket}".encode()
        ).hexdigest()

        ticket = QueueTicket(
            session_id=session_id,
            bucket=bucket,
            shuffle_key=shuffle_key,
            issued_at=now,
        )

        await self.queue.add(bucket, shuffle_key, session_id)
        await self.sessions.bind_active_ticket(session_id, ticket)
        self.metrics.increment("queue_enqueued")
        return ticket

    def compute_admit_rate(self, health: DownstreamHealth) -> int:
        """
        Adaptive throttling: shrink the front door the moment the
        booking core shows strain, rather than waiting for it to fail.
        """
        rate = self.BASE_ADMIT_RATE

        if health.db_latency_ms > 50:
            rate *= 0.5
        if health.db_latency_ms > 150:
            rate *= 0.3
        if health.payment_success_rate < 0.98:
            rate *= 0.6
        if health.booking_core_cpu_pct > 80:
            rate *= 0.4

        return max(self.MIN_ADMIT_RATE, int(rate))

    async def run_admission_tick(self, health: DownstreamHealth):
        """
        Runs every ~1 second. Pulls the next batch of tickets, in
        fair shuffled order, and releases them into the booking core.
        """
        admit_count = self.compute_admit_rate(health)
        batch = await self.queue.pop_next(admit_count)

        for session_id in batch:
            await self.sessions.mark_admitted(session_id)

        self.metrics.gauge("admit_rate", admit_count)
        self.metrics.gauge("queue_depth", await self.queue.size())
        return batch

# booking/saga_orchestrator.py

"""
Booking Saga Orchestrator.

Coordinates seat hold -> passenger/OTP -> payment -> PNR as one
logical transaction spanning multiple services, with compensating
actions if any step fails. Modeled the same way UPI's debit/credit
saga is: hold first, confirm second, and ALWAYS have a clean way back.
"""

from dataclasses import dataclass
from typing import Optional
from enum import Enum
from datetime import datetime, timedelta
import uuid


class BookingState(Enum):
    ADMITTED = "admitted"
    SEAT_HELD = "seat_held"
    PAYMENT_INITIATED = "payment_initiated"
    PNR_CONFIRMED = "pnr_confirmed"
    HOLD_EXPIRED = "hold_expired"
    HOLD_RELEASED = "hold_released"
    DEEMED_PENDING = "deemed_pending"
    COMPENSATED = "compensated"


@dataclass
class BookingAttempt:
    attempt_id: str
    pool_key: str                 # train:date:class:quota
    state: BookingState
    seat_ref: Optional[str] = None
    payment_ref: Optional[str] = None
    pnr: Optional[str] = None
    hold_expires_at: Optional[datetime] = None
    idempotency_key: str = ""


class BookingSagaOrchestrator:
    """
    Orchestrates the multi-step booking saga for one attempt.

    Guarantees:
    1. IDEMPOTENCY: retrying with the same idempotency_key never
       double-charges or double-allocates a seat.
    2. BOUNDED HOLDS: a held seat is always released within TTL,
       either by success, explicit abandonment, or expiry. (The TTL
       job only releases attempts still in SEAT_HELD; a DEEMED_PENDING
       hold is resolved by reconciliation instead, so a payment that
       turns out to have succeeded never lands on a resold seat.)
    3. COMPENSATION: any failure after a hold releases the seat and
       refunds any captured payment — automatically.
    """

    HOLD_TTL = timedelta(minutes=5)

    def __init__(self, db, seat_pool, payment_gateway, pnr_generator,
                 notifier, expiry_scheduler):
        self.db = db
        self.pool = seat_pool
        self.payments = payment_gateway
        self.pnr_generator = pnr_generator
        self.notifier = notifier
        self.expiry_scheduler = expiry_scheduler

    async def hold_seat(self, pool_key: str, idempotency_key: str) -> BookingAttempt:
        """Step 1: convert an admission ticket into a real, TTL-bound hold."""
        existing = await self.db.get_by_idempotency_key(idempotency_key)
        if existing:
            return existing  # same request retried; return the same attempt

        seat_ref = await self.pool.allocate_from_pool(pool_key)
        if seat_ref is None:
            return BookingAttempt(
                attempt_id=str(uuid.uuid4()),
                pool_key=pool_key,
                state=BookingState.HOLD_EXPIRED,
                idempotency_key=idempotency_key,
            )

        attempt = BookingAttempt(
            attempt_id=str(uuid.uuid4()),
            pool_key=pool_key,
            state=BookingState.SEAT_HELD,
            seat_ref=seat_ref,
            hold_expires_at=datetime.utcnow() + self.HOLD_TTL,
            idempotency_key=idempotency_key,
        )
        await self.db.save(attempt)

        # Schedule an automatic release exactly at TTL — never rely on
        # a client callback to free an abandoned seat.
        await self.expiry_scheduler.schedule(
            attempt.attempt_id, run_at=attempt.hold_expires_at
        )
        return attempt

    async def confirm_with_payment(
        self, attempt_id: str, payment_token: str
    ) -> BookingAttempt:
        """Step 2: attempt payment for a held seat, then confirm or compensate."""
        attempt = await self.db.get(attempt_id)

        if attempt.state != BookingState.SEAT_HELD:
            return attempt  # already resolved one way or another

        if datetime.utcnow() > attempt.hold_expires_at:
            await self._release_hold(attempt, reason="EXPIRED_BEFORE_PAYMENT")
            return attempt

        attempt.state = BookingState.PAYMENT_INITIATED
        await self.db.save(attempt)

        result = await self.payments.charge(
            token=payment_token,
            reference=attempt.attempt_id,
            idempotency_key=attempt.idempotency_key,
        )

        if result.status == "SUCCESS":
            pnr = await self.pnr_generator.generate(attempt)
            attempt.pnr = pnr
            attempt.payment_ref = result.reference
            attempt.state = BookingState.PNR_CONFIRMED
            await self.db.save(attempt)
            await self.notifier.send_confirmation(attempt)
            return attempt

        if result.status == "TIMEOUT":
            # Uncertain outcome — do NOT guess. Park it for reconciliation,
            # exactly like UPI's "deemed success" state.
            attempt.state = BookingState.DEEMED_PENDING
            await self.db.save(attempt)
            await self.expiry_scheduler.schedule_reconciliation(
                attempt.attempt_id, check_in=timedelta(minutes=10)
            )
            return attempt

        # Clean failure — compensate immediately
        await self._release_hold(attempt, reason="PAYMENT_FAILED")
        return attempt

    async def _release_hold(self, attempt: BookingAttempt, reason: str):
        """The Saga's compensating transaction: give the seat back."""
        await self.pool.release_to_pool(attempt.pool_key, attempt.seat_ref)
        attempt.state = BookingState.HOLD_RELEASED
        await self.db.save(attempt)

        if attempt.payment_ref:
            await self.payments.refund(attempt.payment_ref)

    async def reconcile_deemed_pending(self, attempt_id: str):
        """
        Called by the background reconciliation worker for any
        booking stuck in DEEMED_PENDING — resolves it one way,
        permanently, based on the payment gateway's true record.
        """
        attempt = await self.db.get(attempt_id)
        actual = await self.payments.check_status(attempt.attempt_id)

        if actual.status == "SUCCESS":
            pnr = await self.pnr_generator.generate(attempt)
            attempt.pnr = pnr
            attempt.payment_ref = actual.reference
            attempt.state = BookingState.PNR_CONFIRMED
            await self.db.save(attempt)
            await self.notifier.send_confirmation(attempt)
        elif actual.status == "PENDING":
            # Gateway still doesn't know: check again later, never guess
            await self.expiry_scheduler.schedule_reconciliation(
                attempt.attempt_id, check_in=timedelta(minutes=10)
            )
        else:
            await self._release_hold(attempt, reason="RECONCILED_AS_FAILED")

# security/tatkal_risk_gate.py

"""
Risk gate specific to the Tatkal booking window. Runs BEFORE the
booking saga is allowed to hold a seat, so that flagged traffic
never even competes for real inventory.
"""

from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime, time


@dataclass
class TatkalRiskSignals:
    user_id: str
    device_fingerprint: str
    is_authorized_agent: bool
    aadhaar_verified: bool
    otp_verified: bool

    # Velocity
    requests_per_min_this_device: int
    accounts_seen_on_this_device_today: int

    # Network
    is_datacenter_ip: bool
    ip_request_rate_per_sec: float

    # Behavioral
    form_fill_time_ms: int          # time between page load and submit
    tatkal_bookings_today: int      # this user, this opening window


class TatkalRiskGate:
    """
    Fast, explainable rules run first (cheap, deterministic);
    an ML model catches subtler patterns the rules miss.
    Target decision latency: < 50ms, same discipline as a real-time
    payment fraud check.
    """

    MIN_HUMAN_FILL_TIME_MS = 1200  # a script can submit in <100ms

    def __init__(self, ml_model, agent_window_config):
        self.model = ml_model
        self.agent_window = agent_window_config  # (start, end) per class

    async def evaluate(
        self, signals: TatkalRiskSignals, now: datetime, travel_class: str
    ) -> Tuple[str, List[str]]:
        """
        Returns: (decision, triggered_rules)
        decision: "ALLOW", "BLOCK", "STEP_UP" (extra OTP / CAPTCHA)
        """
        triggered: List[str] = []

        # RULE 1: Identity verification is mandatory, not optional
        if not (signals.aadhaar_verified and signals.otp_verified):
            return "BLOCK", ["IDENTITY_NOT_VERIFIED"]

        # RULE 2: Agent time-window restriction
        if signals.is_authorized_agent:
            start, end = self.agent_window.get(travel_class)
            if start <= now.time() <= end:
                return "BLOCK", ["AGENT_WINDOW_RESTRICTED"]

        # RULE 3: One Tatkal booking per user per window
        if signals.tatkal_bookings_today > 0:
            return "BLOCK", ["ALREADY_BOOKED_THIS_WINDOW"]

        # RULE 4: Device fan-out — many accounts, one device, same morning
        if signals.accounts_seen_on_this_device_today > 3:
            triggered.append("DEVICE_ACCOUNT_FANOUT")

        # RULE 5: Network origin
        if signals.is_datacenter_ip:
            triggered.append("DATACENTER_IP")
        if signals.ip_request_rate_per_sec > 5:
            triggered.append("HIGH_IP_VELOCITY")

        # RULE 6: Inhuman speed
        if signals.form_fill_time_ms < self.MIN_HUMAN_FILL_TIME_MS:
            triggered.append("SUBHUMAN_FILL_SPEED")

        # ML model catches the combinations rules can't enumerate
        ml_score = await self.model.predict(signals)

        if ml_score > 0.9 or len(triggered) >= 3:
            return "BLOCK", triggered + ["ML_HIGH_RISK"]

        if ml_score > 0.6 or len(triggered) >= 1:
            return "STEP_UP", triggered

        return "ALLOW", triggered