Razorpay Payment Gateway.

# api/idempotency.py

"""
Idempotent write APIs. The key, a hash of the request, and the stored
response live in the same database as the effect, written in the same
transaction, so a crash can never leave "effect done, key not recorded".
"""

import hashlib
import json
from dataclasses import dataclass
from typing import Callable, Optional


@dataclass
class Stored:
    status: str              # IN_PROGRESS | DONE
    request_hash: str
    response: Optional[dict]


class IdempotencyMismatch(Exception):
    pass


class InProgress(Exception):
    pass


def request_hash(method: str, path: str, body: dict) -> str:
    return hashlib.sha256(f"{method} {path} {json.dumps(body, sort_keys=True)}".encode()).hexdigest()


class IdempotentHandler:
    def __init__(self, db):
        self.db = db

    def run(self, merchant_id: str, key: str, method: str, path: str, body: dict,
            effect: Callable[[object], dict]) -> dict:
        h = request_hash(method, path, body)
        with self.db.transaction() as tx:
            inserted = tx.execute(
                "INSERT INTO idem_keys (merchant_id, key, request_hash, status) "
                "VALUES (%s, %s, %s, 'IN_PROGRESS') ON CONFLICT DO NOTHING",
                (merchant_id, key, h),
            ).rowcount
            if inserted == 0:
                row: Stored = tx.fetch_one(
                    "SELECT status, request_hash, response FROM idem_keys "
                    "WHERE merchant_id = %s AND key = %s FOR UPDATE", (merchant_id, key))
                if row.request_hash != h:
                    raise IdempotencyMismatch(key)            # same key, different request
                if row.status == "DONE":
                    return row.response                       # a retry: same answer
                raise InProgress(key)                         # first attempt still running

            response = effect(tx)                             # creates the order, in this tx
            tx.execute(
                "UPDATE idem_keys SET status = 'DONE', response = %s "
                "WHERE merchant_id = %s AND key = %s",
                (json.dumps(response), merchant_id, key),
            )
            return response


def transition(tx, payment_id: str, frm: str, to: str) -> bool:
    """The only way a payment changes state. False means someone else got there first."""
    return tx.execute(
        "UPDATE payments SET state = %s, updated_at = now() WHERE id = %s AND state = %s",
        (to, payment_id, frm),
    ).rowcount == 1

# routing/router.py

"""
Route a payment to the processor most likely to approve it, using
success rates per slice (method, network, issuer, processor). A slice
that drops far below its baseline is marked down for a while; a small
probe share keeps testing it so it can come back.
"""

import random
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, List, Optional, Tuple

WINDOW_S = 300
MIN_ATTEMPTS = 50
DOWN_FOR_S = 20 * 60           # the documented Optimizer behaviour uses 20 minutes
PROBE_SHARE = 0.02


@dataclass
class SliceStats:
    events: Deque[Tuple[float, bool]] = field(default_factory=deque)   # (ts, technical_ok)
    baseline: float = 0.9
    down_until: float = 0.0

    def rate(self, now: float) -> Optional[float]:
        while self.events and self.events[0][0] < now - WINDOW_S:
            self.events.popleft()
        if len(self.events) < MIN_ATTEMPTS:
            return None
        return sum(ok for _, ok in self.events) / len(self.events)


class Router:
    def __init__(self, rng: random.Random):
        self.stats: Dict[Tuple[str, str], SliceStats] = {}
        self.rng = rng

    def record(self, slice_key: str, processor: str, technical_ok: bool, now: float) -> None:
        """Called with each result. Customer declines (wrong OTP, no funds) are not recorded."""
        s = self.stats.setdefault((slice_key, processor), SliceStats())
        s.events.append((now, technical_ok))
        r = s.rate(now)
        if r is not None and r < s.baseline - 0.15:
            s.down_until = now + DOWN_FOR_S                      # move traffic away, for a while

    def choose(self, slice_key: str, allowed: List[Tuple[str, int]], now: float) -> str:
        """allowed: (processor, merchant priority), lower priority number first."""
        up, down = [], []
        for proc, prio in sorted(allowed, key=lambda a: a[1]):
            s = self.stats.get((slice_key, proc))
            (down if s and s.down_until > now else up).append((proc, prio, s))
        if down and (not up or self.rng.random() < PROBE_SHARE):
            return down[0][0]                                     # keep learning about down routes
        if not up:
            return sorted(allowed, key=lambda a: a[1])[0][0]

        def score(item):
            proc, prio, s = item
            r = (s.rate(now) if s else None) or (s.baseline if s else 0.9)
            return r - prio * 0.01                                # merchant priority breaks near-ties
        return max(up, key=score)[0]

# recon/reconciler.py

"""
Drives every UNKNOWN payment to a final state. Callbacks are handled
elsewhere; this worker covers the payments we heard nothing about,
within processors' limits on status checks, and falls back to
settlement files. Every action is a guarded transition.
"""

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

UPI_CHECK_AT = [timedelta(seconds=90), timedelta(seconds=180), timedelta(seconds=270)]


@dataclass
class Payment:
    id: str
    order_id: str
    method: str
    processor: str
    processor_ref: str
    state: str
    created_at: datetime
    checks_done: int = 0


class Reconciler:
    def __init__(self, db, adapters, orders, scheduler, notifier):
        self.db = db
        self.adapters = adapters
        self.orders = orders
        self.scheduler = scheduler
        self.notifier = notifier

    def on_unknown(self, p: Payment) -> None:
        for delay in UPI_CHECK_AT if p.method == "upi" else [timedelta(seconds=30), timedelta(minutes=5)]:
            self.scheduler.at(p.created_at + delay, "status_check", p.id)

    def status_check(self, payment_id: str) -> None:
        p = self.db.get_payment(payment_id)
        if p.state != "UNKNOWN":
            return                                     # a callback already settled it
        answer = self.adapters[p.processor].status(p.processor_ref)   # same reference every time
        self.db.increment_checks(p.id)
        if answer == "SUCCESS":
            self._settle(p, success=True)
        elif answer == "FAILED":
            self._settle(p, success=False)
        # "PENDING": wait for the next scheduled check, or the settlement file

    def on_settlement_file_row(self, processor_ref: str, amount_paise: int) -> None:
        p = self.db.get_payment_by_ref(processor_ref)
        if p.state == "UNKNOWN":
            self._settle(p, success=True)
        elif p.state == "FAILED":
            self._late_success(p)                      # money moved after we said no
        # CAPTURED / REFUNDED: matched, nothing to do except mark reconciled

    def close_unmatched(self, file_date: datetime) -> None:
        """After the file for a day is complete, UNKNOWN payments not in it failed."""
        for p in self.db.unknown_created_before(file_date):
            self._settle(p, success=False)

    def _settle(self, p: Payment, success: bool) -> None:
        with self.db.transaction() as tx:
            if success:
                if self.db.transition(tx, p.id, "UNKNOWN", "AUTHORIZED"):
                    self.db.outbox(tx, "payment.authorized", p.id)
            elif self.db.transition(tx, p.id, "UNKNOWN", "FAILED"):
                self.db.outbox(tx, "payment.failed", p.id)

    def _late_success(self, p: Payment) -> None:
        with self.db.transaction() as tx:
            if not self.db.transition(tx, p.id, "FAILED", "LATE_SUCCESS"):
                return
            if self.orders.is_open(p.order_id, tx=tx):
                self.db.transition(tx, p.id, "LATE_SUCCESS", "CAPTURED")
                self.db.outbox(tx, "payment.captured", p.id)
            else:
                self.db.transition(tx, p.id, "LATE_SUCCESS", "AUTO_REFUNDED")
                self.db.outbox(tx, "refund.created", p.id)
        self.notifier.customer_and_merchant(p.id)

# ledger/postings.py

"""
Double-entry ledger. Every business event is one posting: a set of
entries that sums to zero, written in one transaction and never
updated. Balances are sums of entries (cached, but always rebuildable).
Amounts are integer paise; never floats.
"""

from dataclasses import dataclass
from typing import List

FEE_BPS = 200            # 2% (illustrative pricing)
GST_BPS = 1800           # 18% on the fee


@dataclass(frozen=True)
class Entry:
    account: str         # e.g. "merchant_payable:m_42"
    amount_paise: int    # debit positive, credit negative


class UnbalancedPosting(Exception):
    pass


class Ledger:
    def __init__(self, db):
        self.db = db

    def post(self, tx, posting_id: str, kind: str, entries: List[Entry]) -> None:
        if sum(e.amount_paise for e in entries) != 0:
            raise UnbalancedPosting(posting_id)
        # posting_id is unique: replaying the same event can never post twice
        tx.execute("INSERT INTO postings (id, kind) VALUES (%s, %s)", (posting_id, kind))
        for e in entries:
            tx.execute("INSERT INTO entries (posting_id, account, amount_paise) VALUES (%s, %s, %s)",
                       (posting_id, e.account, e.amount_paise))


def capture_entries(merchant: str, processor: str, amount: int) -> List[Entry]:
    fee = amount * FEE_BPS // 10_000
    gst = (fee * GST_BPS + 5_000) // 10_000      # rounded to the nearest paisa
    return [
        Entry(f"processor_receivable:{processor}", amount),
        Entry(f"merchant_payable:{merchant}", -amount),
        Entry(f"merchant_payable:{merchant}", fee + gst),
        Entry("fee_revenue", -fee),
        Entry("gst_payable", -gst),
    ]


def settlement_entries(merchant: str, payable: int) -> List[Entry]:
    return [
        Entry(f"merchant_payable:{merchant}", payable),
        Entry("escrow_cash", -payable),
    ]

# webhooks/sender.py

"""
At-least-once webhook delivery. Each merchant has its own queue and
concurrency limit, so one failing merchant can't slow the others.
Every delivery is signed and carries a stable event id.
"""

import hashlib
import hmac
import json
import random
from dataclasses import dataclass

TIMEOUT_S = 5
RETRY_WINDOW_S = 24 * 3600
BASE_S, CAP_S = 30, 3600


@dataclass
class Delivery:
    event_id: str              # stable across retries: the merchant dedupes on it
    merchant_id: str
    url: str
    body: dict
    created_at: float
    attempt: int = 0


def signature(secret: bytes, raw: bytes) -> str:
    return hmac.new(secret, raw, hashlib.sha256).hexdigest()


def next_delay(attempt: int, rng: random.Random) -> float:
    """Exponential backoff with jitter, capped, so retries don't arrive in waves."""
    return min(CAP_S, BASE_S * 2 ** attempt) * (0.5 + rng.random() / 2)


class Sender:
    def __init__(self, http, secrets, queues, alerts, rng):
        self.http = http
        self.secrets = secrets
        self.queues = queues          # per-merchant queues with concurrency limits
        self.alerts = alerts
        self.rng = rng

    def deliver(self, d: Delivery, now: float) -> None:
        raw = json.dumps(d.body, separators=(",", ":")).encode()
        headers = {
            "X-Event-Id": d.event_id,
            "X-Signature": signature(self.secrets[d.merchant_id], raw),
        }
        try:
            r = self.http.post(d.url, data=raw, headers=headers, timeout=TIMEOUT_S)
            if 200 <= r.status < 300:
                return                                            # done; never resent
        except TimeoutError:
            pass                                                  # may have been processed: hence event ids
        if now - d.created_at >= RETRY_WINDOW_S:
            self.queues.disable(d.merchant_id)
            self.alerts.merchant(d.merchant_id, "webhook disabled after 24 hours of failures")
            return
        d.attempt += 1
        self.queues.schedule(d.merchant_id, d, run_at=now + next_delay(d.attempt, self.rng))