deploy: clean backend production release

.dockerignore

@@ -0,0 +1,19 @@
+.git
+.gitignore
+.env
+.env.*
+!.env.example
+target
+/frontend/node_modules
+/frontend/dist
+/frontend/playwright-report
+/frontend/test-results
+node_modules
+*.log
+backend.log
+backend_run.log
+cargo_check.log
+errors.log
+uploads
+scratch
+.DS_Store

.sqlx/query-5574d58fbcbd8d25f772c38a1aa87b08fb00ca58bd635570528271c95bfc9184.json

@@ -0,0 +1,19 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO idempotency_keys (key, merchant_id, action_scope, request_hash, response_data, status) VALUES ($1, $2, $3, $4, $5, $6)",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Varchar",
+        "Text",
+        "Varchar"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5574d58fbcbd8d25f772c38a1aa87b08fb00ca58bd635570528271c95bfc9184"
+}

.sqlx/query-af708ad96d5c6081d4d4ee814d5bd24008ea285217e846f547c8c3d9fec58f25.json

@@ -0,0 +1,54 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT key, merchant_id, action_scope, request_hash, response_data, status FROM idempotency_keys WHERE key = $1 AND merchant_id = $2 AND action_scope = $3",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "key",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "merchant_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "action_scope",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 3,
+        "name": "request_hash",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "response_data",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 5,
+        "name": "status",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false,
+      true,
+      false
+    ]
+  },
+  "hash": "af708ad96d5c6081d4d4ee814d5bd24008ea285217e846f547c8c3d9fec58f25"
+}

.sqlx/query-c8f23356524b430677693128e4956749f2644b6d8109a974bae1f9b4e32c3a57.json

@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE idempotency_keys SET response_data = $1, status = $2 WHERE key = $3 AND merchant_id = $4 AND action_scope = $5",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Varchar",
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "c8f23356524b430677693128e4956749f2644b6d8109a974bae1f9b4e32c3a57"
+}

Cargo.toml

@@ -0,0 +1,54 @@
+[package]
+name = "rtix"
+version = "0.2.0"
+edition = "2021"
+
+[dependencies]
+axum = { version = "0.7.5", features = ["ws", "macros"] }
+tokio = { version = "1.38", features = ["full"] }
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+uuid = { version = "1.8", features = ["v4", "serde"] }
+tower-http = { version = "0.5.2", features = ["cors", "trace", "compression-gzip", "compression-br"] }
+sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio-rustls", "postgres", "uuid", "macros", "chrono", "migrate"] }
+dotenvy = "0.15"
+jsonwebtoken = { version = "10.3.0", features = ["rust_crypto"] }
+argon2 = "0.5.3"
+rand = "0.8.5"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+base64 = "0.22"
+sha2 = "0.10"
+validator = { version = "0.18", features = ["derive"] }
+tower = { version = "0.4", features = ["util", "buffer", "limit"] }
+regex = "1.10"
+once_cell = "1.19"
+futures-util = "0.3"
+chrono = { version = "0.4", features = ["serde"] }
+thiserror = "1.0"
+async-trait = "0.1"
+metrics = "0.24"
+metrics-exporter-prometheus = "0.16"
+urlencoding = "2.1"
+parking_lot = "0.12"
+dashmap = "6.0"
+mimalloc = "0.1.43"
+smol_str = { version = "0.2.1", features = ["serde"] }
+reqwest = { version = "0.12", features = ["json"] }
+hmac = "0.12"
+hex = "0.4"
+aes-gcm = "0.10"
+aead = "0.5"
+opentelemetry = "0.21"
+opentelemetry_sdk = { version = "0.21", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.14", default-features = false, features = ["http-proto", "reqwest-client", "trace"] }
+tracing-opentelemetry = "0.22"
+tracing-core = "0.1.36"
+octocrab = "0.51.0"
+
+[profile.release]
+opt-level = 3
+lto = "thin"
+codegen-units = 16
+panic = "abort"
+strip = true

Dockerfile

@@ -0,0 +1,60 @@
+# Stage 1: Build dependencies & application
+FROM rust:1.94-slim-bookworm AS builder
+USER root
+
+# Install system dependencies required for compiling Rust libraries (like openssl, cmake, etc.)
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends pkg-config libssl-dev cmake g++ ca-certificates \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy dependency manifests first
+COPY Cargo.toml Cargo.lock ./
+
+# Create dummy source files so we can cache dependency compilation
+RUN mkdir -p src \
+    && echo "" > src/lib.rs \
+    && echo "fn main() {}" > src/main.rs
+
+# Build dependencies in release mode (caching layer)
+RUN cargo build --release
+
+# Copy the real source code
+COPY . .
+
+# Touch source files to ensure cargo compiles them with the real contents
+RUN touch src/lib.rs src/main.rs
+
+# Set SQLx offline compilation mode so we don't need a database connection at compile time
+ENV SQLX_OFFLINE=true
+
+# Recompile the actual application
+RUN cargo build --release --bin rtix
+
+
+# Stage 2: Minimal Runtime
+FROM debian:bookworm-slim AS runtime
+
+# Install runtime SSL dependencies and curl for health check
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends libssl3 ca-certificates curl \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+# Copy the compiled binary from builder
+COPY --from=builder /app/target/release/rtix /usr/local/bin/rtix
+
+# Hugging Face Spaces requires port 7860
+ENV PORT=7860
+ENV SERVER_PORT=7860
+# OAuth defaults (override with HF Secrets in Space settings)
+ENV OAUTH_REDIRECT_BASE=https://gowtham851-rtix.hf.space
+ENV FRONTEND_URL=https://rtix.vercel.app
+EXPOSE 7860
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+  CMD curl -fsS "http://127.0.0.1:7860/health" > /dev/null || exit 1
+
+CMD ["rtix"]

README.md

@@ -0,0 +1,17 @@
+---
+title: Rtix Secure API
+emoji: ⚡
+colorFrom: green
+colorTo: green
+sdk: docker
+app_port: 7860
+---
+
+# Rtix Secure Production API (Hugging Face Spaces)
+
+This is the production backend for the Rtix payment-sensitive storefront application.
+
+## Specifications
+* **Runtime**: Rust Axum (Dockerized)
+* **Port**: 3000
+* **Host**: 0.0.0.0

migrations/0001_baseline.sql

@@ -0,0 +1,145 @@
+-- 001_baseline.sql
+
+-- Merchants Table
+CREATE TABLE IF NOT EXISTS merchants (
+    merchant_id VARCHAR PRIMARY KEY,
+    email VARCHAR NOT NULL UNIQUE,
+    password_hash VARCHAR NOT NULL,
+    brand_name VARCHAR NOT NULL,
+    slug VARCHAR NOT NULL UNIQUE,
+    social_url VARCHAR,
+    upi_id VARCHAR,
+    recovery_key VARCHAR,
+    session_version BIGINT NOT NULL DEFAULT 1,
+    delivery_rate_per_km DOUBLE PRECISION NOT NULL DEFAULT 10.0,
+    delivery_base_fee DOUBLE PRECISION NOT NULL DEFAULT 20.0,
+    base_pincode VARCHAR NOT NULL DEFAULT '560001',
+    auto_settle_threshold DOUBLE PRECISION DEFAULT 0.5,
+    business_address TEXT,
+    logistics_config JSONB DEFAULT '{}'::jsonb,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Product Links Table
+CREATE TABLE IF NOT EXISTS product_links (
+    link_id VARCHAR PRIMARY KEY,
+    merchant_id VARCHAR NOT NULL,
+    product_name VARCHAR NOT NULL,
+    price_inr DOUBLE PRECISION NOT NULL,
+    image_data TEXT,
+    expected_weight DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    link_views BIGINT NOT NULL DEFAULT 0,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Orders Table
+CREATE TABLE IF NOT EXISTS orders (
+    transaction_id VARCHAR PRIMARY KEY,
+    merchant_id VARCHAR NOT NULL,
+    link_id VARCHAR NOT NULL,
+    buyer_phone VARCHAR NOT NULL,
+    buyer_name VARCHAR NOT NULL DEFAULT '',
+    buyer_email VARCHAR NOT NULL DEFAULT '',
+    shipping_pincode VARCHAR,
+    delivery_address TEXT,
+    price_inr DOUBLE PRECISION NOT NULL,
+    status VARCHAR NOT NULL,
+    vpa VARCHAR NOT NULL DEFAULT '',
+    outbound_weight DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    return_weight DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    proof_data TEXT,
+    proof_received_at TIMESTAMP,
+    settled_at TIMESTAMP,
+    paid_at TIMESTAMP,
+    payu_id VARCHAR NOT NULL DEFAULT '',
+    shipped_at TIMESTAMP,
+    delivered_at TIMESTAMP,
+    estimated_delivery_at TIMESTAMP,
+    shipping_method VARCHAR,
+    is_payment BOOLEAN NOT NULL DEFAULT FALSE,
+    platform_fee_paid BOOLEAN NOT NULL DEFAULT FALSE,
+    platform_fee DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    delivery_fee DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    distance_km DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    risk_score DOUBLE PRECISION DEFAULT 0.0,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Feedback Table
+CREATE TABLE IF NOT EXISTS feedback (
+    id BIGSERIAL PRIMARY KEY,
+    merchant_id VARCHAR,
+    category VARCHAR,
+    message TEXT NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Risk Audit Logs Table
+CREATE TABLE IF NOT EXISTS risk_audit_logs (
+    id BIGSERIAL PRIMARY KEY,
+    transaction_id VARCHAR,
+    merchant_id VARCHAR NOT NULL,
+    event_type VARCHAR NOT NULL,
+    risk_level VARCHAR NOT NULL,
+    details TEXT,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Idempotency Keys Table
+CREATE TABLE IF NOT EXISTS idempotency_keys (
+    key VARCHAR NOT NULL,
+    merchant_id VARCHAR,
+    action_scope VARCHAR NOT NULL DEFAULT 'global',
+    request_hash VARCHAR NOT NULL,
+    response_data TEXT,
+    status VARCHAR NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    expires_at TIMESTAMP,
+    PRIMARY KEY (key, merchant_id, action_scope)
+);
+
+-- Customers Table
+CREATE TABLE IF NOT EXISTS customers (
+    customer_id VARCHAR PRIMARY KEY,
+    phone VARCHAR NOT NULL UNIQUE,
+    name VARCHAR,
+    email VARCHAR,
+    password_hash VARCHAR NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Security Blocks Table
+CREATE TABLE IF NOT EXISTS security_blocks (
+    ip VARCHAR PRIMARY KEY,
+    reason VARCHAR,
+    block_level VARCHAR NOT NULL,
+    expires_at TIMESTAMP NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Login Attempts Table
+CREATE TABLE IF NOT EXISTS login_attempts (
+    id BIGSERIAL PRIMARY KEY,
+    email VARCHAR NOT NULL,
+    ip_address VARCHAR NOT NULL,
+    successful BOOLEAN NOT NULL DEFAULT FALSE,
+    attempted_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Indices
+CREATE INDEX IF NOT EXISTS idx_merchants_email ON merchants(email);
+CREATE INDEX IF NOT EXISTS idx_merchants_slug ON merchants(slug);
+CREATE INDEX IF NOT EXISTS idx_product_links_merchant ON product_links(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_orders_merchant_status ON orders(merchant_id, status);
+CREATE INDEX IF NOT EXISTS idx_orders_link ON orders(link_id);
+CREATE INDEX IF NOT EXISTS idx_orders_phone ON orders(buyer_phone);
+CREATE INDEX IF NOT EXISTS idx_risk_merchant ON risk_audit_logs(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_login_attempts_email_time ON login_attempts(email, attempted_at);
+CREATE INDEX IF NOT EXISTS idx_login_attempts_ip_time ON login_attempts(ip_address, attempted_at);
+CREATE INDEX IF NOT EXISTS idx_idempotency_lookup ON idempotency_keys(key, merchant_id, action_scope);
+CREATE INDEX IF NOT EXISTS idx_orders_pincode_prefix ON orders ((LEFT(shipping_pincode, 3)));
+CREATE INDEX IF NOT EXISTS idx_orders_status_created ON orders(status, created_at);
+CREATE INDEX IF NOT EXISTS idx_orders_merchant_created ON orders(merchant_id, created_at DESC);
+CREATE INDEX IF NOT EXISTS idx_orders_buyer_merchant ON orders(buyer_phone, merchant_id);
+CREATE INDEX IF NOT EXISTS idx_security_blocks_expires ON security_blocks(ip, expires_at);
+CREATE INDEX IF NOT EXISTS idx_customers_phone_unique ON customers(phone);

migrations/0002_api_keys.sql

@@ -0,0 +1,13 @@
+-- 0002_api_keys.sql
+
+CREATE TABLE IF NOT EXISTS api_keys (
+    key_id VARCHAR PRIMARY KEY, -- This is the public part of the key
+    merchant_id VARCHAR NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    name VARCHAR NOT NULL DEFAULT 'Default Key',
+    secret_hash VARCHAR NOT NULL, -- Hashed secret
+    scopes JSONB NOT NULL DEFAULT '["read", "write"]'::jsonb,
+    last_used_at TIMESTAMP,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_api_keys_merchant ON api_keys(merchant_id);

migrations/0003_orders_jsonb_proof.sql

@@ -0,0 +1,13 @@
+-- 0003_orders_jsonb_proof.sql
+
+-- 1. Create a temporary column
+ALTER TABLE orders ADD COLUMN proof_data_new JSONB DEFAULT '[]'::jsonb;
+
+-- 2. Migrate existing data
+UPDATE orders 
+SET proof_data_new = jsonb_build_array(proof_data) 
+WHERE proof_data IS NOT NULL AND proof_data != '';
+
+-- 3. Drop old column and rename new one
+ALTER TABLE orders DROP COLUMN proof_data;
+ALTER TABLE orders RENAME COLUMN proof_data_new TO proof_data;

migrations/0004_webhooks.sql

@@ -0,0 +1,13 @@
+-- 0004_webhooks.sql
+
+CREATE TABLE IF NOT EXISTS merchant_webhooks (
+    webhook_id VARCHAR PRIMARY KEY,
+    merchant_id VARCHAR NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    url TEXT NOT NULL,
+    secret VARCHAR NOT NULL,
+    events JSONB NOT NULL DEFAULT '["order.created", "order.shipped", "order.delivered", "order.disputed", "risk.alert"]'::jsonb,
+    is_active BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_webhooks_merchant ON merchant_webhooks(merchant_id);

migrations/0005_audit_request_id.sql

@@ -0,0 +1,7 @@
+-- 0005_audit_request_id.sql
+
+ALTER TABLE risk_audit_logs ADD COLUMN request_id VARCHAR;
+ALTER TABLE login_attempts ADD COLUMN request_id VARCHAR;
+
+CREATE INDEX IF NOT EXISTS idx_risk_request_id ON risk_audit_logs(request_id);
+CREATE INDEX IF NOT EXISTS idx_login_request_id ON login_attempts(request_id);

migrations/0006_order_risk_flags.sql

@@ -0,0 +1,3 @@
+-- 0006_order_risk_flags.sql
+
+ALTER TABLE orders ADD COLUMN risk_flags JSONB DEFAULT '{}'::jsonb;

migrations/0007_indian_market_hardening.sql

@@ -0,0 +1,11 @@
+-- 0007_indian_market_hardening.sql
+
+ALTER TABLE merchants ADD COLUMN gstin VARCHAR;
+ALTER TABLE merchants ADD COLUMN state_code INT DEFAULT 29; -- Default to Karnataka (29)
+
+ALTER TABLE orders ADD COLUMN cgst DOUBLE PRECISION DEFAULT 0.0;
+ALTER TABLE orders ADD COLUMN sgst DOUBLE PRECISION DEFAULT 0.0;
+ALTER TABLE orders ADD COLUMN igst DOUBLE PRECISION DEFAULT 0.0;
+ALTER TABLE orders ADD COLUMN utr_number VARCHAR;
+
+CREATE INDEX IF NOT EXISTS idx_orders_utr ON orders(utr_number);

migrations/0008_carrier_registry.sql

@@ -0,0 +1,17 @@
+-- 0008_carrier_registry.sql
+
+CREATE TABLE IF NOT EXISTS carrier_registry (
+    carrier_id VARCHAR PRIMARY KEY,
+    name VARCHAR NOT NULL,
+    service_type VARCHAR NOT NULL, -- 'LOCAL', 'REGIONAL', 'NATIONAL'
+    base_rate DOUBLE PRECISION NOT NULL,
+    per_kg_rate DOUBLE PRECISION NOT NULL,
+    is_active BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Seed data for standard Indian carriers
+INSERT INTO carrier_registry (carrier_id, name, service_type, base_rate, per_kg_rate) VALUES
+('shadowfax_local', 'Shadowfax Local', 'LOCAL', 40.0, 10.0),
+('delhivery_regional', 'Delhivery Regional', 'REGIONAL', 65.0, 25.0),
+('bluedart_national', 'BlueDart National', 'NATIONAL', 120.0, 45.0);

migrations/0009_dispute_evidence.sql

@@ -0,0 +1,17 @@
+-- 0009_dispute_evidence.sql
+
+CREATE TABLE IF NOT EXISTS dispute_evidence (
+    evidence_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    transaction_id VARCHAR NOT NULL REFERENCES orders(transaction_id),
+    evidence_url TEXT NOT NULL,
+    uploader_role VARCHAR NOT NULL, -- 'MERCHANT', 'BUYER', 'ADMIN'
+    metadata JSONB DEFAULT '{}'::jsonb,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+ALTER TABLE orders ADD COLUMN delivery_gps_lat DOUBLE PRECISION;
+ALTER TABLE orders ADD COLUMN delivery_gps_lng DOUBLE PRECISION;
+ALTER TABLE orders ADD COLUMN is_geofence_verified BOOLEAN;
+
+-- New status for held settlements
+-- 'DISPUTED_HELD'

migrations/0010_sovereign_intelligence.sql

@@ -0,0 +1,17 @@
+-- 0010_secure_intelligence.sql
+
+CREATE TABLE IF NOT EXISTS pincode_stats (
+    pincode VARCHAR PRIMARY KEY,
+    total_orders BIGINT DEFAULT 0,
+    total_disputes BIGINT DEFAULT 0,
+    avg_delivery_time_hours DOUBLE PRECISION DEFAULT 0.0,
+    volatility_score DOUBLE PRECISION DEFAULT 0.0, -- 0.0 to 1.0
+    last_updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+ALTER TABLE merchants ADD COLUMN reliability_score DOUBLE PRECISION DEFAULT 1.0;
+ALTER TABLE merchants ADD COLUMN network_rank INT;
+
+ALTER TABLE orders ADD COLUMN pincode_volatility_at_checkout DOUBLE PRECISION DEFAULT 0.0;
+
+CREATE INDEX IF NOT EXISTS idx_pincode_volatility ON pincode_stats(volatility_score);

migrations/0011_forensic_chain.sql

@@ -0,0 +1,10 @@
+-- 0011_smart_chain.sql
+
+ALTER TABLE risk_audit_logs ADD COLUMN entry_hash VARCHAR(64);
+ALTER TABLE risk_audit_logs ADD COLUMN previous_hash VARCHAR(64);
+
+ALTER TABLE login_attempts ADD COLUMN entry_hash VARCHAR(64);
+ALTER TABLE login_attempts ADD COLUMN previous_hash VARCHAR(64);
+
+CREATE INDEX IF NOT EXISTS idx_risk_audit_hash ON risk_audit_logs(entry_hash);
+CREATE INDEX IF NOT EXISTS idx_login_attempts_hash ON login_attempts(entry_hash);

migrations/0012_multi_item_orders.sql

@@ -0,0 +1,14 @@
+-- 0012_multi_item_orders.sql
+
+CREATE TABLE IF NOT EXISTS order_items (
+    id BIGSERIAL PRIMARY KEY,
+    transaction_id VARCHAR NOT NULL REFERENCES orders(transaction_id) ON DELETE CASCADE,
+    product_id VARCHAR NOT NULL,
+    product_name VARCHAR NOT NULL,
+    quantity INTEGER NOT NULL DEFAULT 1,
+    price_at_checkout DOUBLE PRECISION NOT NULL,
+    weight_at_checkout DOUBLE PRECISION NOT NULL DEFAULT 0.0,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_order_items_transaction ON order_items(transaction_id);

migrations/0013_merchant_trust_score.sql

@@ -0,0 +1,5 @@
+-- Add trust score to merchants to enable autonomous settlement thresholds
+ALTER TABLE merchants ADD COLUMN trust_score DOUBLE PRECISION DEFAULT 100.0;
+
+-- Index for analytics and scoring performance
+CREATE INDEX idx_merchants_trust_score ON merchants(trust_score);

migrations/0014_product_feedback.sql

@@ -0,0 +1,12 @@
+-- Customer feedback system for product improvement
+CREATE TABLE product_feedback (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    transaction_id VARCHAR(255) NOT NULL REFERENCES orders(transaction_id),
+    product_id UUID NOT NULL,
+    rating INTEGER CHECK (rating >= 1 AND rating <= 5),
+    comment TEXT,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX idx_feedback_product ON product_feedback(product_id);
+CREATE INDEX idx_feedback_transaction ON product_feedback(transaction_id);

migrations/0015_merchant_verification.sql

@@ -0,0 +1,6 @@
+-- Merchant verification tiers for scaling and risk management
+ALTER TABLE merchants ADD COLUMN verification_level VARCHAR(50) DEFAULT 'UNVERIFIED'; -- UNVERIFIED, SILVER, GOLD, PLATINUM
+ALTER TABLE merchants ADD COLUMN max_order_value_inr DOUBLE PRECISION DEFAULT 10000.0;
+
+-- Index for risk-based query optimization
+CREATE INDEX idx_merchants_verification ON merchants(verification_level);

migrations/0016_carrier_registry.sql

@@ -0,0 +1,15 @@
+-- Carrier registry for trusted logistics partners
+CREATE TABLE carriers (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name VARCHAR(255) NOT NULL,
+    api_endpoint VARCHAR(255),
+    trust_level VARCHAR(50) DEFAULT 'STANDARD', -- STANDARD, VERIFIED, STRATEGIC
+    supported_pincodes TEXT[], -- Optional: array of pincodes they specialize in
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+-- Seed some initial data
+INSERT INTO carriers (name, trust_level) VALUES 
+('Secure Express', 'STRATEGIC'),
+('National Logistics Hub', 'VERIFIED'),
+('Local Rapid Relay', 'STANDARD');

migrations/0017_product_inventory.sql

@@ -0,0 +1,3 @@
+-- Basic inventory management for product links
+ALTER TABLE product_links ADD COLUMN inventory_count INTEGER DEFAULT 100;
+ALTER TABLE product_links ADD COLUMN is_unlimited BOOLEAN DEFAULT FALSE;

migrations/0018_merchant_ux_expansion.sql

@@ -0,0 +1,3 @@
+-- Add merchant announcements and product categorization
+ALTER TABLE merchants ADD COLUMN announcement_banner TEXT;
+ALTER TABLE product_links ADD COLUMN category TEXT;

migrations/0019_growth_suite.sql

@@ -0,0 +1,20 @@
+-- Rtix Growth Suite: Coupons and Featured Products
+CREATE TABLE coupons (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    merchant_id TEXT NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    code TEXT NOT NULL,
+    discount_type TEXT NOT NULL, -- 'PERCENTAGE', 'FIXED'
+    discount_value FLOAT8 NOT NULL,
+    min_order_amount FLOAT8 DEFAULT 0.0,
+    max_discount_amount FLOAT8,
+    expiry_date TIMESTAMP,
+    is_active BOOLEAN DEFAULT TRUE,
+    usage_count INT DEFAULT 0,
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    UNIQUE(merchant_id, code)
+);
+
+ALTER TABLE product_links ADD COLUMN is_featured BOOLEAN DEFAULT FALSE;
+
+ALTER TABLE orders ADD COLUMN discount_amount FLOAT8 DEFAULT 0.0;
+ALTER TABLE orders ADD COLUMN coupon_code TEXT;

migrations/0020_social_growth.sql

@@ -0,0 +1,5 @@
+-- Phase 3: Social Growth & Sales Velocity
+ALTER TABLE product_links ADD COLUMN sale_price_inr DECIMAL(12,2);
+ALTER TABLE product_links ADD COLUMN sale_ends_at TIMESTAMP;
+
+ALTER TABLE product_feedback ADD COLUMN is_public BOOLEAN DEFAULT TRUE;

migrations/0021_settlement_ledger.sql

@@ -0,0 +1,18 @@
+-- 0021_settlement_ledger.sql
+
+CREATE TABLE IF NOT EXISTS settlements (
+    settlement_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    transaction_id VARCHAR NOT NULL REFERENCES orders(transaction_id),
+    merchant_id VARCHAR NOT NULL REFERENCES merchants(merchant_id),
+    gross_amount_inr DOUBLE PRECISION NOT NULL,
+    platform_fee_inr DOUBLE PRECISION NOT NULL,
+    delivery_fee_inr DOUBLE PRECISION NOT NULL,
+    tax_amount_inr DOUBLE PRECISION NOT NULL,
+    net_payout_inr DOUBLE PRECISION NOT NULL,
+    utr_number VARCHAR,
+    settled_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    status VARCHAR NOT NULL DEFAULT 'COMPLETED' -- 'COMPLETED', 'FAILED', 'REVERSED'
+);
+
+CREATE INDEX idx_settlements_merchant_id ON settlements(merchant_id);
+CREATE INDEX idx_settlements_transaction_id ON settlements(transaction_id);

migrations/0022_velocity_guard.sql

@@ -0,0 +1,28 @@
+-- 0022_velocity_guard.sql
+
+-- Track transaction activity per fingerprint and IP for velocity monitoring
+CREATE TABLE IF NOT EXISTS velocity_metrics (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    fingerprint VARCHAR,
+    ip_address VARCHAR,
+    merchant_id VARCHAR REFERENCES merchants(merchant_id),
+    activity_count INT DEFAULT 1,
+    last_activity_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    window_start_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    is_blocked BOOLEAN DEFAULT FALSE
+);
+
+CREATE INDEX idx_velocity_fingerprint ON velocity_metrics(fingerprint);
+CREATE INDEX idx_velocity_ip ON velocity_metrics(ip_address);
+CREATE INDEX idx_velocity_merchant ON velocity_metrics(merchant_id);
+
+-- Persistent blacklist for known fraudulent devices
+CREATE TABLE IF NOT EXISTS device_blacklist (
+    fingerprint VARCHAR PRIMARY KEY,
+    reason TEXT,
+    blacklisted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+    severity VARCHAR DEFAULT 'HIGH' -- 'MEDIUM', 'HIGH', 'CRITICAL'
+);
+
+-- Add device_fingerprint to risk_audit_logs if not exists
+ALTER TABLE risk_audit_logs ADD COLUMN IF NOT EXISTS device_fingerprint VARCHAR;

migrations/0023_merchant_plan.sql

@@ -0,0 +1,2 @@
+-- 0023_merchant_plan.sql
+ALTER TABLE merchants ADD COLUMN IF NOT EXISTS plan VARCHAR(50) NOT NULL DEFAULT 'FREE';

migrations/0024_subscriptions.sql

@@ -0,0 +1,55 @@
+-- 0024_subscriptions.sql
+-- Subscription & Recurring Billing Engine for Rtix
+-- Supports SaaS merchants offering recurring payment products.
+
+CREATE TABLE IF NOT EXISTS subscription_plans (
+    id              VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id     VARCHAR(36) NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    name            VARCHAR(255) NOT NULL,
+    description     TEXT,
+    price_inr       NUMERIC(12, 2) NOT NULL CHECK (price_inr > 0),
+    interval_days   INT NOT NULL CHECK (interval_days > 0),  -- e.g. 30 = monthly, 365 = annual
+    trial_days      INT NOT NULL DEFAULT 0,
+    is_active       BOOLEAN NOT NULL DEFAULT TRUE,
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS subscriptions (
+    id                  VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id         VARCHAR(36) NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    plan_id             VARCHAR(36) NOT NULL REFERENCES subscription_plans(id),
+    subscriber_email    VARCHAR(255) NOT NULL,
+    subscriber_phone    VARCHAR(20),
+    subscriber_name     VARCHAR(255) NOT NULL,
+    status              VARCHAR(20) NOT NULL DEFAULT 'TRIAL'
+                            CHECK (status IN ('TRIAL', 'ACTIVE', 'PAST_DUE', 'CANCELLED', 'EXPIRED')),
+    current_period_start TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    current_period_end   TIMESTAMPTZ NOT NULL,
+    cancelled_at        TIMESTAMPTZ,
+    cancel_reason       TEXT,
+    metadata            JSONB DEFAULT '{}',
+    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS subscription_billing_events (
+    id              VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    subscription_id VARCHAR(36) NOT NULL REFERENCES subscriptions(id) ON DELETE CASCADE,
+    merchant_id     VARCHAR(36) NOT NULL,
+    amount_inr      NUMERIC(12, 2) NOT NULL,
+    status          VARCHAR(20) NOT NULL DEFAULT 'PENDING'
+                        CHECK (status IN ('PENDING', 'SUCCESS', 'FAILED', 'REFUNDED')),
+    transaction_id  VARCHAR(36),    -- links to orders.transaction_id on success
+    attempt_count   INT NOT NULL DEFAULT 1,
+    billed_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    settled_at      TIMESTAMPTZ,
+    failure_reason  TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_subscriptions_merchant ON subscriptions(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_subscriptions_status ON subscriptions(status);
+CREATE INDEX IF NOT EXISTS idx_subscriptions_period_end ON subscriptions(current_period_end) WHERE status = 'ACTIVE';
+CREATE INDEX IF NOT EXISTS idx_billing_events_subscription ON subscription_billing_events(subscription_id);
+CREATE INDEX IF NOT EXISTS idx_billing_events_merchant ON subscription_billing_events(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_subscription_plans_merchant ON subscription_plans(merchant_id);

migrations/0025_payouts.sql

@@ -0,0 +1,55 @@
+-- 0025_payouts.sql
+-- Automated Payout Engine for Rtix
+-- Tracks bank payout schedules, disbursements, and reconciliation.
+
+CREATE TABLE IF NOT EXISTS merchant_bank_accounts (
+    id              VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id     VARCHAR(36) NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    account_holder  VARCHAR(255) NOT NULL,
+    account_number  VARCHAR(50) NOT NULL,  -- stored encrypted in practice
+    ifsc_code       VARCHAR(20) NOT NULL,
+    bank_name       VARCHAR(100),
+    is_primary      BOOLEAN NOT NULL DEFAULT TRUE,
+    is_verified     BOOLEAN NOT NULL DEFAULT FALSE,
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE(merchant_id, account_number)
+);
+
+CREATE TABLE IF NOT EXISTS payouts (
+    id              VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id     VARCHAR(36) NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    bank_account_id VARCHAR(36) REFERENCES merchant_bank_accounts(id),
+    amount_inr      NUMERIC(14, 2) NOT NULL CHECK (amount_inr > 0),
+    fee_inr         NUMERIC(10, 2) NOT NULL DEFAULT 0.0,
+    net_inr         NUMERIC(14, 2) GENERATED ALWAYS AS (amount_inr - fee_inr) STORED,
+    status          VARCHAR(20) NOT NULL DEFAULT 'PENDING'
+                        CHECK (status IN ('PENDING', 'PROCESSING', 'SUCCESS', 'FAILED', 'REVERSED')),
+    mode            VARCHAR(10) NOT NULL DEFAULT 'NEFT'
+                        CHECK (mode IN ('NEFT', 'IMPS', 'RTGS', 'UPI')),
+    utr_number      VARCHAR(50),           -- Unique Transaction Reference from bank
+    initiated_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    processed_at    TIMESTAMPTZ,
+    failure_reason  TEXT,
+    order_ids       JSONB DEFAULT '[]',    -- array of settled order IDs in this payout
+    notes           TEXT
+);
+
+CREATE TABLE IF NOT EXISTS notification_log (
+    id              VARCHAR(36) PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id     VARCHAR(36),
+    recipient_email VARCHAR(255) NOT NULL,
+    event_type      VARCHAR(100) NOT NULL, -- e.g. ORDER_PLACED, PAYOUT_SUCCESS, SUB_RENEWAL
+    subject         VARCHAR(500) NOT NULL,
+    status          VARCHAR(20) NOT NULL DEFAULT 'SENT'
+                        CHECK (status IN ('SENT', 'FAILED', 'BOUNCED')),
+    provider_id     VARCHAR(255),          -- external message ID from Resend/SES
+    sent_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    error_message   TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_payouts_merchant ON payouts(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_payouts_status ON payouts(status);
+CREATE INDEX IF NOT EXISTS idx_payouts_initiated ON payouts(initiated_at DESC);
+CREATE INDEX IF NOT EXISTS idx_bank_accounts_merchant ON merchant_bank_accounts(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_notification_log_merchant ON notification_log(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_notification_log_event ON notification_log(event_type);

migrations/0026_ai_engineer.sql

@@ -0,0 +1,23 @@
+CREATE TABLE IF NOT EXISTS ai_engineer_insights (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    status VARCHAR(50) NOT NULL DEFAULT 'PENDING_REVIEW', -- PENDING_REVIEW, APPROVED, REJECTED, IMPLEMENTED
+    issue_summary TEXT NOT NULL,
+    root_cause_analysis TEXT NOT NULL,
+    proposed_solution TEXT NOT NULL,
+    suggested_code_diff TEXT, -- Git diff format
+    metrics_affected JSONB NOT NULL DEFAULT '{}'::jsonb
+);
+
+CREATE TABLE IF NOT EXISTS error_telemetry (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    source VARCHAR(50) NOT NULL, -- FRONTEND, BACKEND, SYSTEM
+    error_level VARCHAR(20) NOT NULL,
+    message TEXT NOT NULL,
+    stack_trace TEXT,
+    user_context JSONB NOT NULL DEFAULT '{}'::jsonb,
+    analyzed BOOLEAN NOT NULL DEFAULT false
+);
+
+CREATE INDEX IF NOT EXISTS idx_error_telemetry_analyzed ON error_telemetry(analyzed);

migrations/0027_ai_engineer_test_diff.sql

@@ -0,0 +1,2 @@
+-- Migration: Add suggested_test_code_diff to ai_engineer_insights
+ALTER TABLE ai_engineer_insights ADD COLUMN IF NOT EXISTS suggested_test_code_diff TEXT;

migrations/0028_ai_engineer_feedback.sql

@@ -0,0 +1,3 @@
+-- Migration: Add pr_url and error_logs to SRE ai_engineer_insights table
+ALTER TABLE ai_engineer_insights ADD COLUMN IF NOT EXISTS pr_url TEXT;
+ALTER TABLE ai_engineer_insights ADD COLUMN IF NOT EXISTS error_logs TEXT;

migrations/0029_secure_upi_mandate_hardening.sql

@@ -0,0 +1,11 @@
+-- 0029_secure_upi_mandate_hardening.sql
+
+-- 1. Create a partial unique index on payu_id to prevent duplicate Razorpay payment IDs (replay attacks)
+-- Ignoring empty strings (which represent unpaid orders)
+CREATE UNIQUE INDEX IF NOT EXISTS idx_orders_payu_id_unique ON orders (payu_id) 
+WHERE payu_id <> '';
+
+-- 2. Create a partial unique index on utr_number to prevent duplicate direct UPI UTR numbers (double spending)
+-- Ignoring NULLs and empty strings (since unpaid orders have no UTR)
+CREATE UNIQUE INDEX IF NOT EXISTS idx_orders_utr_number_unique ON orders (utr_number) 
+WHERE utr_number IS NOT NULL AND utr_number <> '';

migrations/0030_blind_indexing_and_payout_gating.sql

@@ -0,0 +1,8 @@
+-- 0030_blind_indexing_and_payout_gating.sql
+-- Introduce Cryptographic Blind Indexing for orders
+
+-- 1. Add buyer_phone_hash column to allow O(1) indexed searches on encrypted PII
+ALTER TABLE orders ADD COLUMN IF NOT EXISTS buyer_phone_hash VARCHAR;
+
+-- 2. Create the index for sub-millisecond directory lookups
+CREATE INDEX IF NOT EXISTS idx_orders_buyer_phone_hash ON orders(buyer_phone_hash);

migrations/0031_oauth_accounts.sql

@@ -0,0 +1,21 @@
+-- 0031_oauth_accounts.sql
+-- OAuth provider accounts for Google & GitHub sign-in
+-- Links a provider identity to a Rtix merchant account.
+-- One merchant can have multiple OAuth providers linked.
+
+CREATE TABLE IF NOT EXISTS oauth_accounts (
+    id                  VARCHAR(36)  PRIMARY KEY DEFAULT gen_random_uuid()::text,
+    merchant_id         VARCHAR(36)  NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    provider            VARCHAR(20)  NOT NULL CHECK (provider IN ('google', 'github')),
+    provider_user_id    VARCHAR(255) NOT NULL,
+    email               VARCHAR(255) NOT NULL,
+    display_name        VARCHAR(255),
+    avatar_url          TEXT,
+    created_at          TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
+    updated_at          TIMESTAMPTZ  NOT NULL DEFAULT NOW(),
+    -- A given provider user can only be linked to one merchant account
+    UNIQUE (provider, provider_user_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_oauth_merchant   ON oauth_accounts(merchant_id);
+CREATE INDEX IF NOT EXISTS idx_oauth_provider   ON oauth_accounts(provider, provider_user_id);

migrations/0032_add_platform_fee_utr.sql

@@ -0,0 +1,6 @@
+-- 0032_add_platform_fee_utr.sql
+
+ALTER TABLE orders ADD COLUMN IF NOT EXISTS platform_fee_utr VARCHAR;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_orders_platform_fee_utr_unique ON orders (platform_fee_utr)
+WHERE platform_fee_utr IS NOT NULL AND platform_fee_utr <> '';

migrations/0033_postpaid_billing_cycle.sql

@@ -0,0 +1,19 @@
+-- 0033_postpaid_billing_cycle.sql
+
+ALTER TABLE merchants ADD COLUMN IF NOT EXISTS is_frozen BOOLEAN NOT NULL DEFAULT FALSE;
+ALTER TABLE merchants ADD COLUMN IF NOT EXISTS billing_cycle_start TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP;
+
+CREATE TABLE IF NOT EXISTS merchant_invoices (
+    invoice_id VARCHAR PRIMARY KEY,
+    merchant_id VARCHAR NOT NULL REFERENCES merchants(merchant_id) ON DELETE CASCADE,
+    amount_inr DOUBLE PRECISION NOT NULL,
+    order_count INT NOT NULL,
+    status VARCHAR NOT NULL, -- 'UNPAID', 'PAID', 'OVERDUE'
+    billing_period_start TIMESTAMP NOT NULL,
+    billing_period_end TIMESTAMP NOT NULL,
+    due_at TIMESTAMP NOT NULL,
+    paid_at TIMESTAMP,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+
+CREATE INDEX IF NOT EXISTS idx_merchant_invoices_merchant ON merchant_invoices(merchant_id);

src/application/mod.rs

@@ -0,0 +1,2 @@
+pub mod reconciliation;
+pub mod services;

src/application/reconciliation.rs

@@ -0,0 +1,214 @@
+use std::time::Duration;
+
+use futures_util::stream::{self, StreamExt};
+use sqlx::Row;
+use tokio::time::sleep;
+
+use crate::domain::constants::{
+    ORDER_STATUS_DELIVERED_PENDING_APPROVAL, ORDER_STATUS_PAID_PENDING_DELIVERY,
+    ORDER_STATUS_PAYMENT_FAILED, ORDER_STATUS_PENDING_PAYMENT, ORDER_STATUS_SETTLED,
+};
+use crate::interfaces::http::api::{AppState, RealtimeEvent};
+
+pub async fn spawn_reconciliation_worker(state: AppState) {
+    tracing::info!("Reconciliation worker active.");
+
+    let mut interval = tokio::time::interval(Duration::from_secs(300));
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
+    loop {
+        interval.tick().await;
+        metrics::counter!("rtix_reconciliation_cycles_total").increment(1);
+
+        match reconcile_pending_orders(&state).await {
+            Ok(count) if count > 0 => {
+                tracing::info!("Reconciliation cycle resolved {} orders.", count);
+                metrics::counter!("rtix_reconciliation_resolved_total").increment(count as u64);
+            }
+            Ok(_) => {}
+            Err(e) => {
+                tracing::error!("Reconciliation cycle failed: {:?}", e);
+                metrics::counter!("rtix_reconciliation_errors_total").increment(1);
+                // On DB error, wait a bit before next tick
+                sleep(Duration::from_secs(10)).await;
+            }
+        }
+    }
+}
+
+pub async fn reconcile_pending_orders(state: &AppState) -> Result<usize, sqlx::Error> {
+    let mut reconciled_count = 0;
+
+    let stale_pending_orders = sqlx::query(
+        "UPDATE orders SET status = $1 WHERE status = $2 AND created_at < NOW() - INTERVAL '60 minutes' RETURNING transaction_id, merchant_id",
+    )
+    .bind(ORDER_STATUS_PAYMENT_FAILED)
+    .bind(ORDER_STATUS_PENDING_PAYMENT)
+    .fetch_all(&state.pool)
+    .await?;
+
+    if !stale_pending_orders.is_empty() {
+        let stale_count = stale_pending_orders.len();
+
+        let stale_pending_orders_stream = stream::iter(stale_pending_orders)
+            .map(|row| {
+                let state = state.clone();
+                async move {
+                    let txnid: String = row.get("transaction_id");
+                    let merchant_id: String = row.get("merchant_id");
+
+                    if let Ok(mut conn) = state.pool.acquire().await {
+                        crate::domain::audit::log_risk_event(
+                            &mut conn,
+                            Some(&txnid),
+                            &merchant_id,
+                            "PAYMENT_TIMEOUT",
+                            "MEDIUM",
+                            Some("Pending payment expired after 30 minutes without a verified callback."),
+                            None,
+                            None,
+                            None,
+                            Some(&state.tx),
+                        )
+                        .await;
+                    }
+
+                    let _ = state.tx.send(RealtimeEvent::OrderStatusChanged {
+                        transaction_id: txnid,
+                        merchant_id,
+                        new_status: ORDER_STATUS_PAYMENT_FAILED.to_string(),
+                    });
+                }
+            })
+            .buffer_unordered(10);
+
+        stale_pending_orders_stream.collect::<Vec<()>>().await;
+        reconciled_count += stale_count;
+    }
+
+    let aged_shipped_orders = sqlx::query(
+        "UPDATE orders SET status = $1 WHERE status = $2 AND shipped_at IS NOT NULL AND shipped_at < NOW() - INTERVAL '7 days' RETURNING transaction_id, merchant_id",
+    )
+    .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL)
+    .bind(ORDER_STATUS_PAID_PENDING_DELIVERY)
+    .fetch_all(&state.pool)
+    .await?;
+
+    if !aged_shipped_orders.is_empty() {
+        let aged_count = aged_shipped_orders.len();
+
+        let aged_shipped_orders_stream = stream::iter(aged_shipped_orders)
+            .map(|row| {
+                let state = state.clone();
+                async move {
+                    let txnid: String = row.get("transaction_id");
+                    let merchant_id: String = row.get("merchant_id");
+
+                    if let Ok(mut conn) = state.pool.acquire().await {
+                        crate::domain::audit::log_risk_event(
+                            &mut conn,
+                            Some(&txnid),
+                            &merchant_id,
+                            "AUTO_DELIVERY_CONFIRMATION",
+                            "LOW",
+                            Some("Order auto-transitioned after a 7-day shipped window."),
+                            None,
+                            None,
+                            None,
+                            Some(&state.tx),
+                        )
+                        .await;
+                    }
+
+                    let _ = state.tx.send(RealtimeEvent::OrderStatusChanged {
+                        transaction_id: txnid,
+                        merchant_id,
+                        new_status: ORDER_STATUS_DELIVERED_PENDING_APPROVAL.to_string(),
+                    });
+                }
+            })
+            .buffer_unordered(10);
+
+        aged_shipped_orders_stream.collect::<Vec<()>>().await;
+        reconciled_count += aged_count;
+    }
+
+    // 3. Autonomous Settlement for Aged Delivered Orders (48-hour window)
+    let aged_delivered_orders = sqlx::query(
+        "UPDATE orders SET status = $1, settled_at = CURRENT_TIMESTAMP WHERE status = $2 AND delivered_at IS NOT NULL AND delivered_at < NOW() - INTERVAL '48 hours' RETURNING transaction_id, merchant_id, price_inr",
+    )
+    .bind(crate::domain::constants::ORDER_STATUS_SETTLED)
+    .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL)
+    .fetch_all(&state.pool)
+    .await?;
+
+    if !aged_delivered_orders.is_empty() {
+        let aged_delivered_count = aged_delivered_orders.len();
+
+        let aged_delivered_orders_stream = stream::iter(aged_delivered_orders)
+            .map(|row| {
+                let state = state.clone();
+                async move {
+                    let txnid: String = row.get("transaction_id");
+                    let merchant_id: String = row.get("merchant_id");
+                    let price: f64 = row.get("price_inr");
+
+                    if let Ok(mut conn) = state.pool.acquire().await {
+                        crate::domain::audit::log_risk_event(
+                            &mut conn,
+                            Some(&txnid),
+                            &merchant_id,
+                            "AUTONOMOUS_SETTLEMENT",
+                            "LOW",
+                            Some(
+                                "Order auto-settled after 48-hour verification window closed without disputes.",
+                            ),
+                            None,
+                            None,
+                            None,
+                            Some(&state.tx),
+                        )
+                        .await;
+
+                        // Dynamic Trust Scoring: Reward successful autonomous settlement
+                        // Reuse the acquired connection `conn` here instead of checkout from pool
+                        let _ = sqlx::query("UPDATE merchants SET trust_score = LEAST(100.0, trust_score + 0.1) WHERE merchant_id = $1")
+                            .bind(&merchant_id)
+                            .execute(&mut *conn)
+                            .await;
+                    }
+
+                    let _ = state.tx.send(RealtimeEvent::OrderStatusChanged {
+                        transaction_id: txnid,
+                        merchant_id: merchant_id.clone(),
+                        new_status: ORDER_STATUS_SETTLED.to_string(),
+                    });
+
+                    metrics::counter!("rtix_settlement_volume_inr_total").increment(price as u64);
+                }
+            })
+            .buffer_unordered(10);
+
+        aged_delivered_orders_stream.collect::<Vec<()>>().await;
+        reconciled_count += aged_delivered_count;
+    }
+
+    // 4. Handle Stale Unpaid Orders (Expiration)
+    let expired_count = sqlx::query(
+        "UPDATE orders SET status = $1 WHERE status = $2 AND created_at < NOW() - INTERVAL '2 hours'"
+    )
+    .bind(crate::domain::constants::ORDER_STATUS_PAYMENT_FAILED)
+    .bind(crate::domain::constants::ORDER_STATUS_PENDING_PAYMENT)
+    .execute(&state.pool)
+    .await?
+    .rows_affected();
+
+    if expired_count > 0 {
+        tracing::info!(
+            "Reconciliation: Expired {} stale unpaid orders.",
+            expired_count
+        );
+    }
+
+    Ok(reconciled_count)
+}

src/application/services/arbitration.rs

@@ -0,0 +1,81 @@
+use crate::domain::error::AppResult;
+use crate::infrastructure::db::DbPool;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
+pub struct DisputeEvidence {
+    pub evidence_id: Uuid,
+    pub transaction_id: String,
+    pub evidence_url: String,
+    pub uploader_role: String,
+    pub metadata: serde_json::Value,
+}
+
+pub struct ArbitrationService {
+    pool: DbPool,
+}
+
+impl ArbitrationService {
+    pub fn new(pool: DbPool) -> Self {
+        Self { pool }
+    }
+
+    pub async fn upload_evidence(
+        &self,
+        transaction_id: &str,
+        evidence_url: &str,
+        role: &str,
+    ) -> AppResult<Uuid> {
+        let evidence_id = Uuid::new_v4();
+        sqlx::query(
+            "INSERT INTO dispute_evidence (evidence_id, transaction_id, evidence_url, uploader_role) VALUES ($1, $2, $3, $4)"
+        )
+        .bind(evidence_id)
+        .bind(transaction_id)
+        .bind(evidence_url)
+        .bind(role)
+        .execute(&self.pool)
+        .await?;
+
+        Ok(evidence_id)
+    }
+
+    pub async fn get_evidence_for_transaction(
+        &self,
+        transaction_id: &str,
+    ) -> AppResult<Vec<DisputeEvidence>> {
+        let evidence = sqlx::query_as::<_, DisputeEvidence>(
+            r#"SELECT evidence_id, transaction_id, evidence_url, uploader_role, metadata FROM dispute_evidence WHERE transaction_id = $1"#
+        )
+        .bind(transaction_id)
+        .fetch_all(&self.pool)
+        .await?;
+
+        Ok(evidence)
+    }
+
+    pub async fn resolve_dispute(
+        &self,
+        transaction_id: &str,
+        resolution: &str, // 'SETTLED', 'REVERSED'
+    ) -> AppResult<()> {
+        let status = match resolution {
+            "SETTLED" => crate::domain::constants::ORDER_STATUS_SETTLED,
+            "REVERSED" => "REVERSED", // New status
+            _ => {
+                return Err(crate::domain::error::AppError::BadRequest(
+                    "Invalid resolution".into(),
+                ))
+            }
+        };
+
+        sqlx::query("UPDATE orders SET status = $1 WHERE transaction_id = $2")
+            .bind(status)
+            .bind(transaction_id)
+            .execute(&self.pool)
+            .await?;
+
+        Ok(())
+    }
+}

src/application/services/auth.rs

@@ -0,0 +1,366 @@
+use crate::domain::error::{AppError, AppResult};
+use crate::domain::models::Merchant;
+use crate::infrastructure::repositories::MerchantRepository;
+use argon2::{
+    password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
+    Argon2,
+};
+use async_trait::async_trait;
+use dashmap::DashMap;
+use once_cell::sync::Lazy;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use uuid::Uuid;
+
+// ─── Session Cache ────────────────────────────────────────────────────────────
+
+struct CachedSession {
+    version: i64,
+    expires_at: Instant,
+}
+
+static SESSION_CACHE: Lazy<DashMap<String, CachedSession>> = Lazy::new(DashMap::new);
+
+// ─── Login Attempt Tracker — Brute-Force Protection ──────────────────────────
+//
+// Keyed by lowercase email address. Tracks consecutive failed attempts within
+// a rolling 15-minute window. After MAX_FAILURES attempts the account is locked
+// until the window expires. A successful login resets the counter.
+//
+// This is intentionally in-memory (not DB) so it:
+//   a) adds zero latency to the happy path
+//   b) cannot be bypassed by hitting a different DB replica
+//   c) auto-recovers on restart (acceptable for in-memory rate limiting)
+//
+// For persistent cross-replica locking, wire this to Redis or a Postgres
+// advisory lock keyed on the email hash.
+
+const MAX_FAILURES: u32 = 5;
+const LOCKOUT_WINDOW: Duration = Duration::from_secs(15 * 60); // 15 minutes
+
+struct LoginAttempt {
+    failures: u32,
+    first_failure: Instant,
+    locked_until: Option<Instant>,
+}
+
+static LOGIN_ATTEMPTS: Lazy<DashMap<String, LoginAttempt>> = Lazy::new(DashMap::new);
+
+/// Check if an email is currently locked out. Returns `Err(AppError::RateLimited)`
+/// if the account should be refused, `Ok(())` if the attempt may proceed.
+fn check_lockout(email: &str) -> AppResult<()> {
+    let key = email.to_lowercase();
+    if let Some(attempt) = LOGIN_ATTEMPTS.get(&key) {
+        // If a hard lockout timestamp is set and still in the future, refuse.
+        if let Some(locked_until) = attempt.locked_until {
+            if Instant::now() < locked_until {
+                tracing::warn!(
+                    email = %email,
+                    "Login blocked: account locked out after {} failed attempts",
+                    attempt.failures
+                );
+                return Err(AppError::RateLimited);
+            }
+        }
+        // Rolling window: if the first failure was > 15 minutes ago, the window
+        // has expired and the counter will be reset on the next record_failure call.
+    }
+    Ok(())
+}
+
+/// Record a failed login. Increments the counter; locks the account after
+/// MAX_FAILURES attempts within LOCKOUT_WINDOW.
+fn record_failure(email: &str) {
+    let key = email.to_lowercase();
+    let now = Instant::now();
+
+    let mut entry = LOGIN_ATTEMPTS.entry(key).or_insert_with(|| LoginAttempt {
+        failures: 0,
+        first_failure: now,
+        locked_until: None,
+    });
+
+    // Reset window if the previous failure was outside the rolling window
+    if now.duration_since(entry.first_failure) > LOCKOUT_WINDOW {
+        entry.failures = 0;
+        entry.first_failure = now;
+        entry.locked_until = None;
+    }
+
+    entry.failures += 1;
+
+    if entry.failures >= MAX_FAILURES {
+        entry.locked_until = Some(now + LOCKOUT_WINDOW);
+        tracing::warn!(
+            email = %email,
+            failures = entry.failures,
+            "Login lockout triggered: too many failed attempts"
+        );
+    }
+}
+
+/// Reset the login attempt counter on successful authentication.
+fn record_success(email: &str) {
+    LOGIN_ATTEMPTS.remove(&email.to_lowercase());
+}
+
+// ─── Auth Service Trait ───────────────────────────────────────────────────────
+
+#[async_trait]
+pub trait AuthService: Send + Sync {
+    async fn register(
+        &self,
+        email: &str,
+        password: &str,
+        brand_name: &str,
+        slug: Option<&str>,
+        social_url: Option<&str>,
+        upi_id: Option<&str>,
+    ) -> AppResult<(Merchant, String, String)>;
+    async fn login(&self, email: &str, password: &str) -> AppResult<(Merchant, String)>;
+    async fn reset_password(
+        &self,
+        email: &str,
+        recovery_key: &str,
+        new_password: &str,
+    ) -> AppResult<()>;
+    async fn refresh_token(&self, merchant_id: &str) -> AppResult<String>;
+    async fn verify_session(&self, merchant_id: &str, version: i64) -> AppResult<()>;
+}
+
+pub struct RtixAuthService {
+    merchant_repo: Arc<dyn MerchantRepository>,
+    jwt_secret: Vec<u8>,
+}
+
+impl RtixAuthService {
+    pub fn new(merchant_repo: Arc<dyn MerchantRepository>, jwt_secret: Vec<u8>) -> Self {
+        Self {
+            merchant_repo,
+            jwt_secret,
+        }
+    }
+
+    fn hash_password(&self, password: &str) -> AppResult<String> {
+        let mut rng = rand::thread_rng();
+        let salt = SaltString::generate(&mut rng);
+        let argon2 = Argon2::default();
+        argon2
+            .hash_password(password.as_bytes(), &salt)
+            .map(|h| h.to_string())
+            .map_err(|e| AppError::Internal(format!("Hashing failed: {}", e)))
+    }
+
+    fn verify_password(&self, password: &str, hash: &str) -> AppResult<()> {
+        let parsed_hash = PasswordHash::new(hash)
+            .map_err(|_| AppError::Internal("Invalid hash stored".to_string()))?;
+        Argon2::default()
+            .verify_password(password.as_bytes(), &parsed_hash)
+            .map_err(|_| AppError::Auth("Invalid credentials".to_string()))
+    }
+
+    fn issue_token(&self, merchant: &crate::domain::models::Merchant) -> AppResult<String> {
+        let claims = crate::interfaces::http::routes::auth::Claims {
+            sub: merchant.merchant_id.clone(),
+            email: merchant.email.clone(),
+            brand_name: merchant.brand_name.clone(),
+            slug: merchant.slug.clone(),
+            role: Some(merchant.role.clone()),
+            version: merchant.session_version,
+            exp: crate::core::session::access_token_expiry(),
+        };
+
+        jsonwebtoken::encode(
+            &jsonwebtoken::Header::default(),
+            &claims,
+            &jsonwebtoken::EncodingKey::from_secret(&self.jwt_secret),
+        )
+        .map_err(|e| AppError::Internal(format!("Token issuance failed: {}", e)))
+    }
+}
+
+#[async_trait]
+impl AuthService for RtixAuthService {
+    async fn register(
+        &self,
+        email: &str,
+        password: &str,
+        brand_name: &str,
+        slug: Option<&str>,
+        social_url: Option<&str>,
+        upi_id: Option<&str>,
+    ) -> AppResult<(Merchant, String, String)> {
+        let existing: Option<Merchant> = self.merchant_repo.find_by_email(email).await?;
+        if existing.is_some() {
+            return Err(AppError::BadRequest("Email already exists".to_string()));
+        }
+
+        let merchant_id = Uuid::new_v4().to_string();
+        let password_hash = self.hash_password(password)?;
+        let recovery_key = format!(
+            "VTX-{}",
+            &Uuid::new_v4().to_string().replace('-', "")[..16].to_uppercase()
+        );
+        let recovery_key_hash = self.hash_password(&recovery_key)?;
+
+        let mut final_slug = slug.unwrap_or(brand_name).to_lowercase().replace(' ', "-");
+
+        // Pillar IV: Identity Resilience - Automatic collision resolution
+        let mut retry_count = 0;
+        while self
+            .merchant_repo
+            .find_by_slug(&final_slug)
+            .await?
+            .is_some()
+        {
+            retry_count += 1;
+            if retry_count > 5 {
+                return Err(AppError::Internal(
+                    "Could not generate a unique slug".to_string(),
+                ));
+            }
+            let suffix = &Uuid::new_v4().to_string()[..4].to_lowercase();
+            final_slug = format!(
+                "{}-{}",
+                slug.unwrap_or(brand_name).to_lowercase().replace(' ', "-"),
+                suffix
+            );
+        }
+
+        let merchant = Merchant {
+            merchant_id: merchant_id.clone(),
+            email: email.to_string(),
+            password_hash,
+            brand_name: brand_name.to_string(),
+            slug: final_slug,
+            social_url: social_url.map(ToString::to_string),
+            upi_id: upi_id.map(ToString::to_string),
+            business_address: None,
+            recovery_key: Some(recovery_key_hash),
+            session_version: 1,
+            delivery_rate_per_km: 10.0,
+            delivery_base_fee: 20.0,
+            logistics_config: serde_json::json!({
+                "complexity_bias": 1.0,
+                "weight_coefficient": 0.02,
+                "distance_coefficient": 1.0
+            }),
+            base_pincode: "560001".to_string(),
+            auto_settle_threshold: 50.0,
+            trust_score: 100.0,
+            verification_level: "UNVERIFIED".to_string(),
+            max_order_value_inr: 10000.0,
+            created_at: None,
+            state_code: Some(29),
+            gstin: None,
+            announcement_banner: None,
+            plan: "FREE".to_string(),
+            role: "MERCHANT".to_string(),
+            is_frozen: false,
+            billing_cycle_start: None,
+        };
+
+        self.merchant_repo.create(&merchant).await?;
+
+        let token = self.issue_token(&merchant)?;
+
+        Ok((merchant, token, recovery_key))
+    }
+
+    /// Authenticate a merchant with brute-force protection.
+    ///
+    /// # Brute-Force Guard
+    /// - After **5 consecutive failed attempts** within a 15-minute window the
+    ///   email is locked and every further attempt returns HTTP 429 immediately
+    ///   (before any DB query or Argon2 verification, preventing timing oracles).
+    /// - A successful login resets the attempt counter.
+    /// - The lockout is in-memory and resets on server restart (acceptable for
+    ///   single-replica deployments; use Redis for multi-replica hardening).
+    async fn login(&self, email: &str, password: &str) -> AppResult<(Merchant, String)> {
+        // ── Brute-force check BEFORE any DB work ──────────────────────────────
+        check_lockout(email)?;
+
+        let merchant: Option<Merchant> = self.merchant_repo.find_by_email(email).await?;
+        let merchant = merchant.ok_or_else(|| {
+            // Record failure even on unknown email to prevent user enumeration
+            // via timing differences (non-existent vs. wrong-password paths).
+            record_failure(email);
+            AppError::Auth("Invalid email or password".to_string())
+        })?;
+
+        // ── Argon2 password verification ──────────────────────────────────────
+        if let Err(e) = self.verify_password(password, &merchant.password_hash) {
+            record_failure(email);
+            tracing::warn!(
+                email = %email,
+                "Failed login attempt recorded"
+            );
+            return Err(e);
+        }
+
+        // ── Success: clear attempt counter ────────────────────────────────────
+        record_success(email);
+
+        let token = self.issue_token(&merchant)?;
+        Ok((merchant, token))
+    }
+
+    async fn reset_password(
+        &self,
+        email: &str,
+        recovery_key: &str,
+        new_password: &str,
+    ) -> AppResult<()> {
+        let merchant: Option<Merchant> = self.merchant_repo.find_by_email(email).await?;
+        let merchant = merchant.ok_or(AppError::NotFound("Merchant not found".to_string()))?;
+
+        if let Some(hash) = &merchant.recovery_key {
+            self.verify_password(recovery_key, hash)?;
+        } else {
+            return Err(AppError::Auth("No recovery key set".to_string()));
+        }
+
+        let new_hash = self.hash_password(new_password)?;
+        self.merchant_repo.update_password(email, &new_hash).await?;
+
+        Ok(())
+    }
+
+    async fn refresh_token(&self, merchant_id: &str) -> AppResult<String> {
+        let merchant = self.merchant_repo.find_by_id(merchant_id).await?;
+        let merchant = merchant.ok_or(AppError::NotFound("Merchant not found".to_string()))?;
+
+        let token = self.issue_token(&merchant)?;
+        Ok(token)
+    }
+
+    async fn verify_session(&self, merchant_id: &str, version: i64) -> AppResult<()> {
+        let now = Instant::now();
+        if let Some(cached) = SESSION_CACHE.get(merchant_id) {
+            if cached.version == version && cached.expires_at > now {
+                return Ok(());
+            }
+        }
+
+        let merchant = self.merchant_repo.find_by_id(merchant_id).await?;
+        let merchant = merchant.ok_or(AppError::Auth("Session invalid".to_string()))?;
+
+        if merchant.session_version != version {
+            return Err(AppError::Auth("Session version mismatch".to_string()));
+        }
+
+        if !SESSION_CACHE.contains_key(merchant_id) && SESSION_CACHE.len() > 10_000 {
+            SESSION_CACHE.retain(|_, v| v.expires_at > now);
+        }
+
+        SESSION_CACHE.insert(
+            merchant_id.to_string(),
+            CachedSession {
+                version,
+                expires_at: now + Duration::from_secs(60),
+            },
+        );
+
+        Ok(())
+    }
+}

src/application/services/background.rs

@@ -0,0 +1,299 @@
+use crate::application::services::MerchantService;
+use crate::domain::constants::*;
+use crate::infrastructure::db::DbPool;
+use futures_util::stream::{self, StreamExt};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
+
+pub struct ProtocolSentinel {
+    merchant_service: Arc<dyn MerchantService>,
+    pool: DbPool,
+}
+
+impl ProtocolSentinel {
+    pub fn new(merchant_service: Arc<dyn MerchantService>, pool: DbPool) -> Self {
+        Self {
+            merchant_service,
+            pool,
+        }
+    }
+
+    pub async fn run(&self) {
+        tracing::info!("Protocol Sentinel Active: Secure Health Guard Initialized.");
+        loop {
+            // 1. Process Auto-Settlements (Liquidity Guard)
+            if let Err(e) = self.process_auto_settlements().await {
+                tracing::error!("Sentinel Error [Auto-Settlement]: {:?}", e);
+            }
+
+            // 2. Cleanup Stale Intents (Resource Guard)
+            if let Err(e) = self.cleanup_stale_intents().await {
+                tracing::error!("Sentinel Error [Stale-Cleanup]: {:?}", e);
+            }
+
+            // 3. Finalize Aged Deliveries (Custodial Guard)
+            if let Err(e) = self.enforce_custodial_deadlines().await {
+                tracing::error!("Sentinel Error [Custodial-Enforcement]: {:?}", e);
+            }
+
+            // 4. Escalate Stale Forensic Holds (Arbitration Guard)
+            if let Err(e) = self.escalate_stale_holds().await {
+                tracing::error!("Sentinel Error [Hold-Escalation]: {:?}", e);
+            }
+
+            // 5. Generate Monthly Billing (Postpaid Billing Guard)
+            if let Err(e) = self.generate_monthly_billing().await {
+                tracing::error!("Sentinel Error [Monthly-Billing]: {:?}", e);
+            }
+
+            // 6. Freeze Overdue Merchants (Billing Enforcement Guard)
+            if let Err(e) = self.freeze_overdue_merchants().await {
+                tracing::error!("Sentinel Error [Billing-Enforcement]: {:?}", e);
+            }
+
+            // ─── Jitter Sleep ──────────────────────────────────────────────────
+            // Add ±300s random jitter to the 1-hour base interval.
+            // This prevents the thundering herd problem on cold starts / restarts
+            // where every replica would otherwise fire all tasks in perfect sync,
+            // creating a coordinated spike against the database every 3600 seconds.
+            let jitter_secs = rand::random::<u64>() % 300;
+            let sleep_duration = Duration::from_secs(3600 + jitter_secs);
+            tracing::debug!(
+                "Protocol Sentinel: sleeping {}s until next pulse (base 3600s + {}s jitter)",
+                sleep_duration.as_secs(),
+                jitter_secs
+            );
+            sleep(sleep_duration).await;
+        }
+    }
+
+    async fn escalate_stale_holds(&self) -> Result<(), sqlx::Error> {
+        // Escalate DISPUTED_HELD orders after 48 hours to DISPUTED_IN_REVIEW
+        let result = sqlx::query(
+            "UPDATE orders SET status = $1 WHERE status = $2 AND created_at < CURRENT_TIMESTAMP - INTERVAL '48 hours'"
+        )
+        .bind(ORDER_STATUS_DISPUTED)
+        .bind(ORDER_STATUS_DISPUTED_HELD)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() > 0 {
+            tracing::info!(
+                "Sentinel: Escalated {} stale forensic holds to full review.",
+                result.rows_affected()
+            );
+        }
+        Ok(())
+    }
+
+    async fn process_auto_settlements(&self) -> Result<(), sqlx::Error> {
+        // ── FOR UPDATE SKIP LOCKED ─────────────────────────────────────────────
+        // SKIP LOCKED ensures each replica claims a disjoint set of rows.
+        let eligible_orders = sqlx::query(
+            r#"
+            SELECT o.transaction_id, o.merchant_id, m.auto_settle_threshold, o.risk_score
+            FROM orders o
+            JOIN merchants m ON o.merchant_id = m.merchant_id
+            WHERE o.status = $1 
+              AND o.delivered_at < CURRENT_TIMESTAMP - INTERVAL '24 hours'
+              AND o.risk_score <= m.auto_settle_threshold
+            FOR UPDATE SKIP LOCKED
+            "#,
+        )
+        .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let merchant_service = self.merchant_service.clone();
+        let _results = stream::iter(eligible_orders)
+            .map(|order| {
+                let ms = merchant_service.clone();
+                async move {
+                    use sqlx::Row;
+                    let tid: String = order.get("transaction_id");
+                    let mid: String = order.get("merchant_id");
+
+                    let _ = ms.approve_settlement(&mid, &tid, None, None, None).await;
+                }
+            })
+            .buffer_unordered(10)
+            .collect::<Vec<()>>()
+            .await;
+
+        Ok(())
+    }
+
+    async fn cleanup_stale_intents(&self) -> Result<(), sqlx::Error> {
+        // Expire orders stuck in PENDING_PAYMENT for > 6 hours
+        let result = sqlx::query(
+            "UPDATE orders SET status = 'EXPIRED_VOID' WHERE status = $1 AND created_at < CURRENT_TIMESTAMP - INTERVAL '6 hours'"
+        )
+        .bind(ORDER_STATUS_PENDING_PAYMENT)
+        .execute(&self.pool)
+        .await?;
+
+        if result.rows_affected() > 0 {
+            tracing::info!(
+                "Sentinel: Purged {} stale payment intents.",
+                result.rows_affected()
+            );
+        }
+        Ok(())
+    }
+
+    async fn enforce_custodial_deadlines(&self) -> Result<(), sqlx::Error> {
+        // ── FOR UPDATE SKIP LOCKED ─────────────────────────────────────────────
+        let eligible_orders = sqlx::query(
+            "SELECT transaction_id, merchant_id FROM orders WHERE status = $1 AND delivered_at < CURRENT_TIMESTAMP - INTERVAL '48 hours' FOR UPDATE SKIP LOCKED"
+        )
+        .bind(ORDER_STATUS_DELIVERED_PENDING_APPROVAL)
+        .fetch_all(&self.pool)
+        .await?;
+
+        let merchant_service = self.merchant_service.clone();
+        let _results = stream::iter(eligible_orders)
+            .map(|order| {
+                let ms = merchant_service.clone();
+                async move {
+                    use sqlx::Row;
+                    let tid: String = order.get("transaction_id");
+                    let mid: String = order.get("merchant_id");
+
+                    tracing::info!(
+                        "Sentinel: Enforcing custodial deadline for {}. Finalizing liquidity.",
+                        tid
+                    );
+                    let _ = ms.approve_settlement(&mid, &tid, None, None, None).await;
+                }
+            })
+            .buffer_unordered(10)
+            .collect::<Vec<()>>()
+            .await;
+
+        Ok(())
+    }
+
+    pub async fn generate_monthly_billing(&self) -> Result<(), sqlx::Error> {
+        // Query merchants whose billing_cycle_start has passed 30 days
+        let merchants = sqlx::query(
+            "SELECT merchant_id, billing_cycle_start FROM merchants WHERE billing_cycle_start <= CURRENT_TIMESTAMP - INTERVAL '30 days'"
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        for row in merchants {
+            use sqlx::Row;
+            let merchant_id: String = row.get("merchant_id");
+            let billing_cycle_start: chrono::NaiveDateTime = row.get("billing_cycle_start");
+
+            let mut tx = self.pool.begin().await?;
+
+            // Count successful orders placed within this billing cycle
+            // Statuses like PENDING_PAYMENT, PAYMENT_FAILED, EXPIRED_VOID are excluded.
+            let order_count: i32 = sqlx::query_scalar(
+                "SELECT COUNT(*)::INT FROM orders \
+                 WHERE merchant_id = $1 \
+                   AND created_at >= $2 \
+                   AND created_at < $2 + INTERVAL '30 days' \
+                   AND status NOT IN ('PENDING_PAYMENT', 'PAYMENT_FAILED', 'EXPIRED_VOID')"
+            )
+            .bind(&merchant_id)
+            .bind(billing_cycle_start)
+            .fetch_one(&mut *tx)
+            .await?;
+
+            let amount_inr = order_count as f64 * 2.0;
+            let invoice_id = format!("INV-{}", uuid::Uuid::new_v4().to_string()[..8].to_uppercase());
+
+            // Generate invoice
+            sqlx::query(
+                "INSERT INTO merchant_invoices (invoice_id, merchant_id, amount_inr, order_count, status, billing_period_start, billing_period_end, due_at) \
+                 VALUES ($1, $2, $3, $4, 'UNPAID', $5, $5 + INTERVAL '30 days', CURRENT_TIMESTAMP + INTERVAL '7 days')"
+            )
+            .bind(&invoice_id)
+            .bind(&merchant_id)
+            .bind(amount_inr)
+            .bind(order_count)
+            .bind(billing_cycle_start)
+            .execute(&mut *tx)
+            .await?;
+
+            // Update merchant's billing_cycle_start
+            sqlx::query(
+                "UPDATE merchants SET billing_cycle_start = billing_cycle_start + INTERVAL '30 days' WHERE merchant_id = $1"
+            )
+            .bind(&merchant_id)
+            .execute(&mut *tx)
+            .await?;
+
+            tx.commit().await?;
+
+            tracing::info!(
+                merchant_id = %merchant_id,
+                invoice_id = %invoice_id,
+                amount = amount_inr,
+                "Sentinel: Generated monthly postpaid invoice."
+            );
+        }
+
+        Ok(())
+    }
+
+    pub async fn freeze_overdue_merchants(&self) -> Result<(), sqlx::Error> {
+        // Find merchants with unpaid/overdue invoices past due_at
+        let overdue_merchants = sqlx::query(
+            "SELECT DISTINCT merchant_id FROM merchant_invoices WHERE status = 'UNPAID' AND due_at < CURRENT_TIMESTAMP"
+        )
+        .fetch_all(&self.pool)
+        .await?;
+
+        for row in overdue_merchants {
+            use sqlx::Row;
+            let merchant_id: String = row.get("merchant_id");
+
+            let mut tx = self.pool.begin().await?;
+
+            // Freeze merchant account
+            sqlx::query("UPDATE merchants SET is_frozen = TRUE WHERE merchant_id = $1")
+                .bind(&merchant_id)
+                .execute(&mut *tx)
+                .await?;
+
+            // Mark invoice(s) as OVERDUE
+            sqlx::query("UPDATE merchant_invoices SET status = 'OVERDUE' WHERE merchant_id = $1 AND status = 'UNPAID' AND due_at < CURRENT_TIMESTAMP")
+                .bind(&merchant_id)
+                .execute(&mut *tx)
+                .await?;
+
+            tx.commit().await?;
+
+            tracing::warn!(
+                merchant_id = %merchant_id,
+                "Sentinel: Merchant account frozen due to overdue invoice."
+            );
+        }
+
+        // Auto-unfreeze merchants who have no outstanding unpaid/overdue invoices past due date
+        // (This acts as a self-healing sweep)
+        let unfrozen = sqlx::query(
+            "UPDATE merchants SET is_frozen = FALSE \
+             WHERE is_frozen = TRUE \
+               AND merchant_id NOT IN ( \
+                   SELECT DISTINCT merchant_id FROM merchant_invoices \
+                   WHERE status IN ('UNPAID', 'OVERDUE') AND due_at < CURRENT_TIMESTAMP \
+               )"
+        )
+        .execute(&self.pool)
+        .await?;
+
+        if unfrozen.rows_affected() > 0 {
+            tracing::info!(
+                "Sentinel: Auto-unfroze {} merchants with settled invoices.",
+                unfrozen.rows_affected()
+            );
+        }
+
+        Ok(())
+    }
+}

src/application/services/billing.rs

@@ -0,0 +1,125 @@
+use crate::domain::models::{Merchant, OrderRecord};
+
+pub struct BillingService;
+
+impl BillingService {
+    pub fn generate_invoice_html(order: &OrderRecord, merchant: &Merchant) -> String {
+        let subtotal = order.price_inr;
+        let total_tax = order.cgst + order.sgst + order.igst;
+        let grand_total = subtotal + total_tax + order.delivery_fee;
+
+        format!(
+            r#"<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>Invoice - {tx_id}</title>
+    <style>
+        body {{ font-family: 'Inter', sans-serif; color: #111; margin: 0; padding: 40px; line-height: 1.5; }}
+        .invoice-box {{ max-width: 800px; margin: auto; border: 1px solid #eee; padding: 30px; border-radius: 8px; }}
+        .header {{ display: flex; justify-content: space-between; margin-bottom: 40px; border-bottom: 2px solid #000; padding-bottom: 20px; }}
+        .brand {{ font-size: 24px; font-weight: 800; text-transform: uppercase; letter-spacing: 2px; }}
+        .meta {{ text-align: right; }}
+        .details {{ display: grid; grid-template-columns: 1fr 1fr; gap: 40px; margin-bottom: 40px; }}
+        .section-title {{ font-size: 10px; font-weight: 800; color: #888; text-transform: uppercase; letter-spacing: 1px; margin-bottom: 8px; }}
+        table {{ width: 100%; border-collapse: collapse; margin-bottom: 40px; }}
+        th {{ text-align: left; background: #fafafa; padding: 12px; font-size: 12px; text-transform: uppercase; border-bottom: 1px solid #eee; }}
+        td {{ padding: 12px; font-size: 14px; border-bottom: 1px solid #eee; }}
+        .totals {{ margin-left: auto; width: 300px; }}
+        .total-row {{ display: flex; justify-content: space-between; padding: 8px 0; }}
+        .grand-total {{ font-size: 18px; font-weight: 800; border-top: 2px solid #000; margin-top: 10px; padding-top: 10px; }}
+        .footer {{ margin-top: 60px; text-align: center; color: #888; font-size: 12px; }}
+    </style>
+</head>
+<body>
+    <div class="invoice-box">
+        <div class="header">
+            <div class="brand">{brand_name}</div>
+            <div class="meta">
+                <div style="font-weight: 700;">INVOICE</div>
+                <div style="font-size: 12px; color: #666;">#{tx_id}</div>
+                <div style="font-size: 12px; color: #666;">Date: {date}</div>
+            </div>
+        </div>
+
+        <div class="details">
+            <div>
+                <div class="section-title">Sold By</div>
+                <div style="font-weight: 600;">{brand_name}</div>
+                <div style="font-size: 13px; color: #444; max-width: 250px;">{merchant_addr}</div>
+            </div>
+            <div>
+                <div class="section-title">Ship To</div>
+                <div style="font-weight: 600;">{buyer_name}</div>
+                <div style="font-size: 13px; color: #444;">{buyer_email}</div>
+                <div style="font-size: 13px; color: #444; max-width: 250px;">{buyer_addr}</div>
+            </div>
+        </div>
+
+        <table>
+            <thead>
+                <tr>
+                    <th>Item Description</th>
+                    <th style="text-align: right;">Amount</th>
+                </tr>
+            </thead>
+            <tbody>
+                <tr>
+                    <td>Product Purchase ({link_id})</td>
+                    <td style="text-align: right;">₹{subtotal:.2}</td>
+                </tr>
+            </tbody>
+        </table>
+
+        <div class="totals">
+            <div class="total-row">
+                <span>Subtotal</span>
+                <span>₹{subtotal:.2}</span>
+            </div>
+            <div class="total-row">
+                <span>Tax (GST)</span>
+                <span>₹{total_tax:.2}</span>
+            </div>
+            <div class="total-row">
+                <span>Shipping</span>
+                <span>₹{shipping:.2}</span>
+            </div>
+            {discount_row}
+            <div class="total-row grand-total">
+                <span>Grand Total</span>
+                <span>₹{total:.2}</span>
+            </div>
+        </div>
+
+        <div class="footer">
+            Generated by Rtix Sovereign Settlement Protocol. Secure Transaction Verified.
+        </div>
+    </div>
+</body>
+</html>"#,
+            brand_name = merchant.brand_name,
+            tx_id = order.transaction_id,
+            date = order
+                .created_at
+                .map(|t| t.format("%d %b %Y").to_string())
+                .unwrap_or_else(|| "N/A".to_string()),
+            merchant_addr = merchant.business_address.as_deref().unwrap_or("N/A"),
+            buyer_name = order.buyer_name,
+            buyer_email = order.buyer_email,
+            buyer_addr = order.delivery_address.as_deref().unwrap_or("N/A"),
+            link_id = order.link_id,
+            subtotal = subtotal,
+            total_tax = total_tax,
+            shipping = order.delivery_fee,
+            total = grand_total,
+            discount_row = if order.discount_amount > 0.0 {
+                format!(
+                    r#"<div class="total-row" style="color: #00E59B;"><span>Discount</span><span>-₹{:.2}</span></div>"#,
+                    order.discount_amount
+                )
+            } else {
+                String::new()
+            }
+        )
+    }
+}

src/application/services/checkout.rs

@@ -0,0 +1,233 @@
+use crate::domain::error::{AppError, AppResult};
+use crate::domain::models::{OrderRecord, ProductLink};
+use crate::infrastructure::db::DbPool;
+use crate::infrastructure::repositories::{OrderRepository, ProductRepository};
+use async_trait::async_trait;
+use std::sync::Arc;
+
+#[async_trait]
+pub trait CheckoutService: Send + Sync {
+    async fn get_checkout_view(&self, link_id: &str) -> AppResult<ProductLink>;
+    #[allow(clippy::too_many_arguments)]
+    async fn execute_checkout(
+        &self,
+        link_id: &str,
+        buyer_phone: &str,
+        buyer_name: &str,
+        buyer_email: &str,
+        shipping_pincode: &str,
+        delivery_address: &str,
+        coupon_code: Option<String>,
+        request_id: Option<String>,
+        client_ip: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+        device_fingerprint: Option<String>,
+    ) -> AppResult<OrderRecord>;
+    #[allow(clippy::too_many_arguments)]
+    async fn execute_cart_checkout(
+        &self,
+        items: Vec<(String, u32)>, // (link_id, quantity)
+        buyer_phone: &str,
+        buyer_name: &str,
+        buyer_email: &str,
+        shipping_pincode: &str,
+        delivery_address: &str,
+        coupon_code: Option<String>,
+        request_id: Option<String>,
+        client_ip: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+        device_fingerprint: Option<String>,
+    ) -> AppResult<OrderRecord>;
+    async fn submit_delivery_proof(
+        &self,
+        transaction_id: &str,
+        proof_data: &str,
+        proof_token: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+    ) -> AppResult<()>;
+    async fn estimate_delivery(
+        &self,
+        link_id: &str,
+        pincode: &str,
+        _buyer_phone: Option<String>,
+    ) -> AppResult<(f64, f64)>;
+}
+
+pub struct RtixCheckoutService {
+    product_repo: Arc<dyn ProductRepository>,
+    merchant_repo: Arc<dyn crate::infrastructure::repositories::MerchantRepository>,
+    order_repo: Arc<dyn OrderRepository>,
+    assets: Arc<dyn crate::infrastructure::storage::assets::AssetProvider>,
+    tx: tokio::sync::broadcast::Sender<crate::interfaces::http::api::RealtimeEvent>,
+    pool: DbPool,
+}
+
+impl RtixCheckoutService {
+    pub fn new(
+        product_repo: Arc<dyn ProductRepository>,
+        merchant_repo: Arc<dyn crate::infrastructure::repositories::MerchantRepository>,
+        order_repo: Arc<dyn OrderRepository>,
+        assets: Arc<dyn crate::infrastructure::storage::assets::AssetProvider>,
+        tx: tokio::sync::broadcast::Sender<crate::interfaces::http::api::RealtimeEvent>,
+        pool: DbPool,
+    ) -> Self {
+        Self {
+            product_repo,
+            merchant_repo,
+            order_repo,
+            assets,
+            tx,
+            pool,
+        }
+    }
+}
+
+#[async_trait]
+impl CheckoutService for RtixCheckoutService {
+    async fn get_checkout_view(&self, link_id: &str) -> AppResult<ProductLink> {
+        let product = self.product_repo.find_by_id(link_id).await?;
+        let mut product =
+            product.ok_or_else(|| AppError::NotFound("Product link not found".to_string()))?;
+
+        let _ = self.product_repo.increment_views(link_id).await;
+        product.image_data = crate::core::utils::hydrate_file_to_base64(product.image_data).await;
+
+        Ok(product)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    async fn execute_checkout(
+        &self,
+        link_id: &str,
+        buyer_phone: &str,
+        buyer_name: &str,
+        buyer_email: &str,
+        shipping_pincode: &str,
+        delivery_address: &str,
+        coupon_code: Option<String>,
+        request_id: Option<String>,
+        client_ip: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+        device_fingerprint: Option<String>,
+    ) -> AppResult<OrderRecord> {
+        crate::application::services::checkout_impl::execute_checkout_helper(
+            &self.product_repo,
+            &self.merchant_repo,
+            &self.order_repo,
+            &self.pool,
+            &self.tx,
+            link_id,
+            buyer_phone,
+            buyer_name,
+            buyer_email,
+            shipping_pincode,
+            delivery_address,
+            coupon_code,
+            request_id,
+            client_ip,
+            lat,
+            lng,
+            device_fingerprint,
+        )
+        .await
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    async fn execute_cart_checkout(
+        &self,
+        items: Vec<(String, u32)>,
+        buyer_phone: &str,
+        buyer_name: &str,
+        buyer_email: &str,
+        shipping_pincode: &str,
+        delivery_address: &str,
+        coupon_code: Option<String>,
+        request_id: Option<String>,
+        client_ip: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+        device_fingerprint: Option<String>,
+    ) -> AppResult<OrderRecord> {
+        crate::application::services::checkout_impl::execute_cart_checkout_helper(
+            &self.product_repo,
+            &self.merchant_repo,
+            &self.order_repo,
+            &self.pool,
+            &self.tx,
+            items,
+            buyer_phone,
+            buyer_name,
+            buyer_email,
+            shipping_pincode,
+            delivery_address,
+            coupon_code,
+            request_id,
+            client_ip,
+            lat,
+            lng,
+            device_fingerprint,
+        )
+        .await
+    }
+
+    async fn submit_delivery_proof(
+        &self,
+        transaction_id: &str,
+        proof_data: &str,
+        proof_token: &str,
+        lat: Option<f64>,
+        lng: Option<f64>,
+    ) -> AppResult<()> {
+        crate::application::services::checkout_impl::execute_submit_delivery_proof_helper(
+            &self.order_repo,
+            &self.assets,
+            &self.tx,
+            transaction_id,
+            proof_data,
+            proof_token,
+            lat,
+            lng,
+        )
+        .await
+    }
+
+    async fn estimate_delivery(
+        &self,
+        link_id: &str,
+        pincode: &str,
+        _buyer_phone: Option<String>,
+    ) -> AppResult<(f64, f64)> {
+        let product = self
+            .product_repo
+            .find_by_id(link_id)
+            .await?
+            .ok_or_else(|| AppError::NotFound("Product link not found".to_string()))?;
+
+        let merchant = self
+            .merchant_repo
+            .find_by_id(&product.merchant_id)
+            .await?
+            .ok_or_else(|| AppError::NotFound("Merchant not found".to_string()))?;
+
+        let distance_km =
+            crate::domain::distance::estimate_distance_km(&merchant.base_pincode, pincode);
+
+        let pricing_features = crate::application::services::pricing::PricingFeatures {
+            distance_km,
+            user_rate_per_km: merchant.delivery_rate_per_km,
+            product_weight: product.expected_weight,
+            base_charge: merchant.delivery_base_fee,
+            config: serde_json::from_value(merchant.logistics_config.clone()).unwrap_or_default(),
+        };
+
+        let fee = crate::application::services::pricing::PricingEngine::estimate_delivery_fee(
+            pricing_features,
+        );
+
+        Ok((fee, distance_km))
+    }
+}

src/application/services/checkout_impl/mod.rs

@@ -0,0 +1,873 @@
+use crate::domain::error::{AppError, AppResult};
+use crate::domain::models::OrderRecord;
+use crate::infrastructure::db::DbPool;
+use crate::infrastructure::repositories::{MerchantRepository, OrderRepository, ProductRepository};
+use crate::infrastructure::storage::assets::AssetProvider;
+use crate::interfaces::http::api::RealtimeEvent;
+use std::sync::Arc;
+use tokio::sync::broadcast::Sender;
+use uuid::Uuid;
+
+#[allow(clippy::too_many_arguments)]
+pub async fn execute_checkout_helper(
+    product_repo: &Arc<dyn ProductRepository>,
+    merchant_repo: &Arc<dyn MerchantRepository>,
+    order_repo: &Arc<dyn OrderRepository>,
+    pool: &DbPool,
+    tx_sender: &Sender<RealtimeEvent>,
+    link_id: &str,
+    buyer_phone: &str,
+    buyer_name: &str,
+    buyer_email: &str,
+    shipping_pincode: &str,
+    delivery_address: &str,
+    coupon_code: Option<String>,
+    request_id: Option<String>,
+    client_ip: &str,
+    lat: Option<f64>,
+    lng: Option<f64>,
+    device_fingerprint: Option<String>,
+) -> AppResult<OrderRecord> {
+    let product = product_repo.find_by_id(link_id).await?;
+    let mut product =
+        product.ok_or_else(|| AppError::NotFound("Product link not found".to_string()))?;
+
+    let _ = product_repo.increment_views(link_id).await;
+    product.image_data = crate::core::utils::hydrate_file_to_base64(product.image_data).await;
+
+    let merchant = merchant_repo
+        .find_by_id(&product.merchant_id)
+        .await?
+        .ok_or_else(|| AppError::NotFound("Merchant not found".to_string()))?;
+
+    if merchant.is_frozen {
+        return Err(AppError::Forbidden("Merchant account is frozen due to unpaid outstanding invoices.".to_string()));
+    }
+
+    // 1. Secure Logistics Circuit Breaker
+    // Check if the pincode has high smart volatility (recent violations)
+    let volatility_count = sqlx::query_scalar::<_, i64>(
+        "SELECT COUNT(*) FROM risk_audit_logs WHERE details LIKE $1 AND created_at > NOW() - INTERVAL '1 hour'"
+    )
+    .bind(format!("%{}%", shipping_pincode))
+    .fetch_one(order_repo.find_pool())
+    .await
+    .unwrap_or(0);
+
+    if volatility_count > 5 {
+        return Err(AppError::Forbidden(format!(
+            "Logistics Circuit Breaker Active for zone {}. High volatility detected in recent smart audit cycles.",
+            shipping_pincode
+        )));
+    }
+
+    // 2. Institutional Velocity Guard (Anti-Abuse)
+    let intelligence =
+        crate::application::services::intelligence::IntelligenceService::new(pool.clone());
+    let velocity_risk = intelligence
+        .evaluate_velocity_risk(
+            device_fingerprint.as_deref(),
+            Some(client_ip),
+            &product.merchant_id,
+        )
+        .await?;
+
+    if velocity_risk >= 90.0 {
+        // Log Critical Security Event
+        let _ = sqlx::query(
+            "INSERT INTO risk_audit_logs (merchant_id, event_type, risk_level, details, device_fingerprint) VALUES ($1, $2, $3, $4, $5)"
+        )
+        .bind(&product.merchant_id)
+        .bind("VELOCITY_BLOCK")
+        .bind("CRITICAL")
+        .bind(format!("Transaction blocked due to high velocity risk ({}). Fingerprint: {:?}, IP: {}", velocity_risk, device_fingerprint, client_ip))
+        .bind(device_fingerprint.as_deref())
+        .execute(pool)
+        .await;
+
+        return Err(AppError::Forbidden(
+            "Security Protocol Active: High-frequency transaction activity detected from this device/network. Access restricted for protocol safety.".to_string()
+        ));
+    }
+
+    let distance_km =
+        crate::domain::distance::estimate_distance_km(&merchant.base_pincode, shipping_pincode);
+
+    let pricing_features = crate::application::services::pricing::PricingFeatures {
+        distance_km,
+        user_rate_per_km: merchant.delivery_rate_per_km,
+        product_weight: product.expected_weight,
+        base_charge: merchant.delivery_base_fee,
+        config: serde_json::from_value(merchant.logistics_config.clone()).unwrap_or_default(),
+    };
+    let delivery_fee = crate::application::services::pricing::PricingEngine::estimate_delivery_fee(
+        pricing_features,
+    );
+
+    // 3. Precision Geofence Check
+    let mut geofence_verified = None;
+    if let (Some(l_lat), Some(l_lng)) = (lat, lng) {
+        let intelligence =
+            crate::application::services::intelligence::IntelligenceService::new(pool.clone());
+        if !intelligence
+            .verify_geofence_with_precision(shipping_pincode, l_lat, l_lng)
+            .await?
+        {
+            return Err(AppError::Forbidden(format!(
+                "Geofence Verification Failed: Your current GPS coordinates do not match the shipping pincode {}. Forensic integrity active.",
+                shipping_pincode
+            )));
+        }
+        geofence_verified = Some(true);
+    }
+
+    let order_count = order_repo
+        .count_by_buyer(&product.merchant_id, buyer_phone)
+        .await?;
+
+    let transaction_id = Uuid::new_v4().to_string();
+
+    // 2. Merchant Transaction Limits
+    if merchant.plan == "FREE" && product.price_inr > 10000.0 {
+        return Err(AppError::Forbidden(
+            "Transaction value exceeds the ₹10,000 limit for merchants on the FREE plan. Upgrade to PRO to accept higher-value payments without limitations.".to_string()
+        ));
+    }
+
+    if product.price_inr > merchant.max_order_value_inr {
+        return Err(AppError::Forbidden(format!(
+            "Transaction value ₹{} exceeds the current limit for this merchant (₹{}). Increase merchant verification level to lift this restriction.",
+            product.price_inr, merchant.max_order_value_inr
+        )));
+    }
+
+    // Start Transaction for Atomicity
+    let mut tx = pool.begin().await.map_err(AppError::Database)?;
+
+    // 3. Inventory Enforcement
+    if !product.is_unlimited {
+        let rows_affected = sqlx::query(
+            "UPDATE product_links SET inventory_count = inventory_count - 1 WHERE link_id = $1 AND inventory_count > 0",
+        )
+        .bind(&product.link_id)
+        .execute(&mut *tx)
+        .await
+        .map_err(AppError::Database)?
+        .rows_affected();
+
+        if rows_affected == 0 {
+            return Err(AppError::BadRequest(
+                "Product is currently out of stock.".to_string(),
+            ));
+        }
+    }
+
+    let current_price =
+        if let (Some(sale_price), Some(ends_at)) = (product.sale_price_inr, product.sale_ends_at) {
+            if ends_at > chrono::Utc::now().naive_utc() {
+                sale_price
+            } else {
+                product.price_inr
+            }
+        } else {
+            product.price_inr
+        };
+
+    let platform_fee = crate::application::services::pricing::PricingEngine::calculate_platform_fee(
+        current_price,
+        merchant.trust_score,
+    );
+
+    // Calculate preliminary risk score
+    let mut calculated_risk = 0.0;
+    calculated_risk += (volatility_count as f64 * 5.0).min(30.0);
+    calculated_risk += velocity_risk * 0.4; // Incorporate velocity guard signal
+    if geofence_verified == Some(true) {
+        calculated_risk *= 0.8; // Lower risk if GPS verified
+    }
+
+    let mut order = OrderRecord {
+        transaction_id: transaction_id.clone(),
+        merchant_id: product.merchant_id.clone(),
+        link_id: link_id.to_string(),
+        buyer_phone: buyer_phone.to_string(),
+        buyer_phone_hash: None, // populated by encrypt_pii() at persist time
+        buyer_name: buyer_name.to_string(),
+        buyer_email: buyer_email.to_string(),
+        shipping_pincode: Some(shipping_pincode.to_string()),
+        delivery_address: Some(delivery_address.to_string()),
+        price_inr: current_price,
+        status: crate::domain::constants::ORDER_STATUS_PENDING_PAYMENT.to_string(),
+        vpa: None,
+        payu_id: String::new(),
+        outbound_weight: product.expected_weight,
+        return_weight: 0.0,
+        proof_data: None,
+        settled_at: None,
+        shipped_at: None,
+        delivered_at: None,
+        shipping_method: None,
+        estimated_delivery_at: None,
+        is_payment: false,
+        platform_fee_paid: false,
+        platform_fee,
+        delivery_fee,
+        distance_km,
+        risk_score: calculated_risk,
+        risk_flags: None,
+        cgst: 0.0,
+        sgst: 0.0,
+        igst: 0.0,
+        utr_number: None,
+        platform_fee_utr: None,
+        delivery_gps_lat: None,
+        delivery_gps_lng: None,
+        is_geofence_verified: geofence_verified,
+        pincode_volatility_at_checkout: 0.0,
+        discount_amount: 0.0,
+        coupon_code: None,
+        checkout_gps_lat: lat,
+        checkout_gps_lng: lng,
+        device_fingerprint: device_fingerprint.clone(),
+        paid_at: None,
+        proof_received_at: None,
+        created_at: None,
+        brand_name: None,
+    };
+
+    if let Some(ref code) = coupon_code {
+        let coupon = sqlx::query_as::<_, crate::domain::models::Coupon>(
+            "SELECT * FROM coupons WHERE merchant_id = $1 AND code = $2 AND is_active = TRUE",
+        )
+        .bind(&product.merchant_id)
+        .bind(code.to_uppercase())
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        if let Some(c) = coupon {
+            if c.is_valid(order.price_inr) {
+                let discount = c.calculate_discount(order.price_inr);
+                order.discount_amount = discount;
+                order.price_inr -= discount;
+                order.coupon_code = Some(code.clone());
+
+                let _ =
+                    sqlx::query("UPDATE coupons SET usage_count = usage_count + 1 WHERE id = $1")
+                        .bind(c.id)
+                        .execute(&mut *tx)
+                        .await;
+            }
+        }
+    }
+
+    let gst_breakdown = crate::application::services::india_tax::IndiaTaxService::calculate_gst(
+        order.price_inr,
+        merchant.state_code.unwrap_or(29),
+        order.shipping_pincode.as_deref().unwrap_or_default(),
+        0.18, // 18% standard rate
+    );
+    order.cgst = gst_breakdown.cgst;
+    order.sgst = gst_breakdown.sgst;
+    order.igst = gst_breakdown.igst;
+
+    let volatility =
+        crate::application::services::intelligence::IntelligenceService::new(pool.clone())
+            .get_pincode_volatility(order.shipping_pincode.as_deref().unwrap_or_default())
+            .await
+            .unwrap_or(0.0);
+    order.pincode_volatility_at_checkout = volatility;
+
+    if volatility > 0.5 {
+        let _ = tx_sender.send(
+            crate::interfaces::http::api::RealtimeEvent::NetworkVolatilityAlert {
+                pincode: order.shipping_pincode.clone().unwrap_or_default(),
+                volatility_score: volatility,
+                message: format!(
+                    "High logistics volatility detected for pincode {}.",
+                    order.shipping_pincode.clone().unwrap_or_default()
+                ),
+            },
+        );
+    }
+
+    let (risk_score, risk_flags) =
+        crate::application::services::risk::RiskEngine::calculate_risk_score(
+            &order,
+            order_count,
+            volatility,
+        );
+    order.risk_score = risk_score;
+    order.risk_flags = Some(risk_flags);
+
+    if order.risk_score >= 80.0 {
+        // Log Critical/High Security Event outside the transaction so it's persisted even when rolled back
+        if let Ok(mut conn) = pool.acquire().await {
+            crate::domain::audit::log_risk_event(
+                &mut *conn,
+                Some(&transaction_id),
+                &product.merchant_id,
+                "HIGH_RISK_BLOCK",
+                "CRITICAL",
+                Some(&format!(
+                    "Transaction blocked due to high risk score ({:.1}) during checkout for link {}. Flags: {:?}",
+                    order.risk_score, link_id, order.risk_flags
+                )),
+                Some(order.risk_score),
+                request_id.as_deref(),
+                device_fingerprint.as_deref(),
+                Some(tx_sender),
+            )
+            .await;
+        }
+
+        if order.risk_score > 90.0 {
+            crate::interfaces::http::middleware::block_ip_persistently(
+                pool,
+                client_ip,
+                &format!("Automated Defense: High Risk Score ({:.1}) detected during checkout for link {}", order.risk_score, link_id),
+                Some(tx_sender)
+            ).await;
+        }
+
+        return Err(AppError::Forbidden(format!(
+            "Security restriction: High risk profile detected (Score: {:.1}). This transaction has been blocked to prevent potential fraud.",
+            order.risk_score
+        )));
+    } else if order.risk_score > 60.0 {
+        crate::domain::audit::log_risk_event(
+            &mut tx,
+            Some(&transaction_id),
+            &product.merchant_id,
+            "HIGH_RISK_ORDER",
+            "HIGH",
+            Some(&format!(
+                "Order {} flagged with risk score {}",
+                transaction_id, order.risk_score
+            )),
+            Some(order.risk_score),
+            request_id.as_deref(),
+            device_fingerprint.as_deref(),
+            Some(tx_sender),
+        )
+        .await;
+    }
+
+    // Persist Order using transaction
+    order.created_at = Some(chrono::Utc::now().naive_utc());
+    crate::domain::models::OrderRecord::create_with_tx(&mut tx, &order).await?;
+
+    tx.commit().await.map_err(AppError::Database)?;
+
+    Ok(order)
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn execute_submit_delivery_proof_helper(
+    order_repo: &Arc<dyn OrderRepository>,
+    assets: &Arc<dyn AssetProvider>,
+    tx_sender: &Sender<RealtimeEvent>,
+    transaction_id: &str,
+    proof_data: &str,
+    proof_token: &str,
+    lat: Option<f64>,
+    lng: Option<f64>,
+) -> AppResult<()> {
+    let transaction_id = crate::domain::validation::sanitize_filename(transaction_id);
+
+    if crate::core::session::verify_proof_token(proof_token, &transaction_id).is_err() {
+        return Err(AppError::Forbidden(
+            "Invalid proof authorization token".to_string(),
+        ));
+    }
+
+    let order = order_repo.find_by_id(&transaction_id).await?;
+    let order = order.ok_or_else(|| AppError::NotFound("Order not found".to_string()))?;
+
+    if order.status != crate::domain::constants::ORDER_STATUS_PAID_PENDING_DELIVERY {
+        return Err(AppError::BadRequest(
+            "Order is not in a state to accept delivery proof".to_string(),
+        ));
+    }
+
+    let mut tx = order_repo
+        .find_pool()
+        .begin()
+        .await
+        .map_err(AppError::Database)?;
+
+    let is_video = proof_data.contains("video") && proof_data.contains("mp4");
+    let is_png = proof_data.contains("image/png");
+    let file_extension = if is_video {
+        "mp4"
+    } else if is_png {
+        "png"
+    } else {
+        "jpg"
+    };
+    let filename = format!("proof_{}.{}", Uuid::new_v4(), file_extension);
+
+    let bytes = crate::domain::validation::validate_base64_payload(proof_data, 10 * 1024 * 1024)
+        .map_err(|e| AppError::BadRequest(e.message))?;
+
+    // Institutional Enforcement: Require Video for High-Value Orders (> ₹5,000)
+    if order.price_inr > 5000.0 && !is_video {
+        return Err(AppError::BadRequest(
+            "High-value order detected. Smart video proof is mandatory for this transaction."
+                .to_string(),
+        ));
+    }
+
+    if !is_video {
+        let allowed_headers: [Vec<u8>; 2] = [
+            vec![0xFF, 0xD8, 0xFF],
+            vec![0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A],
+        ];
+        if !allowed_headers
+            .iter()
+            .any(|header| bytes.starts_with(header))
+        {
+            return Err(AppError::BadRequest("Invalid image format".to_string()));
+        }
+    }
+
+    let merchant_plan: String =
+        sqlx::query_scalar("SELECT plan FROM merchants WHERE merchant_id = $1")
+            .bind(&order.merchant_id)
+            .fetch_one(&mut *tx)
+            .await
+            .map_err(AppError::Database)?;
+
+    let lat_val = lat.unwrap_or(0.0);
+    let lng_val = lng.unwrap_or(0.0);
+    let mut zk_proof = crate::core::crypto::CryptoService::generate_zk_telemetry_proof(
+        &transaction_id,
+        &bytes,
+        lat_val,
+        lng_val,
+    );
+
+    if merchant_plan == "PRO" {
+        let asset_path = assets
+            .store_asset(&filename, &bytes)
+            .await
+            .map_err(AppError::Internal)?;
+
+        if let Some(obj) = zk_proof.as_object_mut() {
+            obj.insert("is_zero_storage".to_string(), serde_json::json!(false));
+            obj.insert("asset_path".to_string(), serde_json::json!(asset_path));
+        }
+    }
+
+    let mut is_geofence_verified = None;
+    if let (Some(l_lat), Some(l_lng), Some(pincode)) = (lat, lng, &order.shipping_pincode) {
+        let (p_lat, p_lng) = crate::domain::geofence::GeofenceService::get_coordinates(pincode)
+            .unwrap_or((12.9716, 77.5946)); // Default to Bangalore Central
+
+        let distance = crate::domain::geofence::GeofenceService::calculate_distance_km(
+            l_lat, l_lng, p_lat, p_lng,
+        );
+        is_geofence_verified = Some(distance < 5.0); // 5km tolerance
+
+        if !is_geofence_verified.unwrap_or(false) {
+            crate::domain::audit::log_risk_event(
+                &mut tx,
+                Some(&transaction_id),
+                &order.merchant_id,
+                "GEOFENCE_VIOLATION",
+                "MEDIUM",
+                Some(&format!(
+                    "Delivery proof submitted from {} km away from shipping pincode {}.",
+                    distance.round(),
+                    pincode
+                )),
+                Some(distance),
+                None,
+                order.device_fingerprint.as_deref(),
+                Some(tx_sender),
+            )
+            .await;
+
+            // Trust Score Penalty for logistics deviations
+            let _ = sqlx::query("UPDATE merchants SET trust_score = GREATEST(0.0, trust_score - 2.0) WHERE merchant_id = $1")
+                .bind(&order.merchant_id)
+                .execute(&mut *tx)
+                .await;
+        }
+    }
+
+    sqlx::query(
+        "UPDATE orders SET proof_data = proof_data || $1::jsonb, status = $2, delivered_at = CURRENT_TIMESTAMP, delivery_gps_lat = $3, delivery_gps_lng = $4, is_geofence_verified = $5 WHERE transaction_id = $6 AND status = $7",
+    )
+    .bind(serde_json::json!([zk_proof]))
+    .bind(crate::domain::constants::ORDER_STATUS_DELIVERED_PENDING_APPROVAL)
+    .bind(lat)
+    .bind(lng)
+    .bind(is_geofence_verified)
+    .bind(&transaction_id)
+    .bind(crate::domain::constants::ORDER_STATUS_PAID_PENDING_DELIVERY)
+    .execute(&mut *tx)
+    .await
+    .map_err(AppError::Database)?;
+
+    tx.commit().await.map_err(AppError::Database)?;
+
+    let _ = tx_sender.send(
+        crate::interfaces::http::api::RealtimeEvent::OrderStatusChanged {
+            transaction_id: transaction_id.to_string(),
+            merchant_id: order.merchant_id,
+            new_status: crate::domain::constants::ORDER_STATUS_DELIVERED_PENDING_APPROVAL
+                .to_string(),
+        },
+    );
+
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+pub async fn execute_cart_checkout_helper(
+    product_repo: &Arc<dyn crate::infrastructure::repositories::ProductRepository>,
+    merchant_repo: &Arc<dyn crate::infrastructure::repositories::MerchantRepository>,
+    _order_repo: &Arc<dyn crate::infrastructure::repositories::OrderRepository>,
+    pool: &crate::infrastructure::db::DbPool,
+    tx_sender: &tokio::sync::broadcast::Sender<crate::interfaces::http::api::RealtimeEvent>,
+    items: Vec<(String, u32)>,
+    buyer_phone: &str,
+    buyer_name: &str,
+    buyer_email: &str,
+    shipping_pincode: &str,
+    delivery_address: &str,
+    coupon_code: Option<String>,
+    _request_id: Option<String>,
+    client_ip: &str,
+    lat: Option<f64>,
+    lng: Option<f64>,
+    device_fingerprint: Option<String>,
+) -> AppResult<crate::domain::models::OrderRecord> {
+    let transaction_id = format!(
+        "TX_{}",
+        uuid::Uuid::new_v4().to_string()[..8].to_uppercase()
+    );
+
+    let mut total_price = 0.0;
+    let mut total_weight = 0.0;
+    let mut first_merchant_id = String::new();
+    let mut product_details = Vec::new();
+
+    for (link_id, qty) in &items {
+        let product = product_repo
+            .find_by_id(link_id)
+            .await?
+            .ok_or_else(|| AppError::NotFound(format!("Product {} not found", link_id)))?;
+
+        if first_merchant_id.is_empty() {
+            first_merchant_id = product.merchant_id.clone();
+        } else if first_merchant_id != product.merchant_id {
+            return Err(AppError::BadRequest(
+                "Cross-merchant checkout not allowed in single cart".into(),
+            ));
+        }
+
+        let current_price = if let (Some(sale_price), Some(ends_at)) =
+            (product.sale_price_inr, product.sale_ends_at)
+        {
+            if ends_at > chrono::Utc::now().naive_utc() {
+                sale_price
+            } else {
+                product.price_inr
+            }
+        } else {
+            product.price_inr
+        };
+
+        total_price += current_price * (*qty as f64);
+        total_weight += product.expected_weight * (*qty as f64);
+
+        // Inventory Enforcement for Cart (Preliminary Check)
+        if !product.is_unlimited && product.inventory_count < *qty as i32 {
+            return Err(AppError::BadRequest(format!(
+                "Product '{}' is low on stock ({} available).",
+                product.product_name, product.inventory_count
+            )));
+        }
+
+        product_details.push((product, *qty));
+    }
+
+    if first_merchant_id.is_empty() {
+        return Err(AppError::BadRequest("Cart is empty".into()));
+    }
+
+    let merchant = merchant_repo
+        .find_by_id(&first_merchant_id)
+        .await?
+        .ok_or_else(|| AppError::NotFound("Merchant not found".into()))?;
+
+    if merchant.is_frozen {
+        return Err(AppError::Forbidden("Merchant account is frozen due to unpaid outstanding invoices.".to_string()));
+    }
+
+    // 2. Institutional Velocity Guard (Anti-Abuse)
+    let intelligence =
+        crate::application::services::intelligence::IntelligenceService::new(pool.clone());
+    let velocity_risk = intelligence
+        .evaluate_velocity_risk(
+            device_fingerprint.as_deref(),
+            Some(client_ip),
+            &first_merchant_id,
+        )
+        .await?;
+
+    if velocity_risk >= 90.0 {
+        return Err(AppError::Forbidden(
+            "Security Protocol Active: High-frequency transaction activity detected from this device/network. Access restricted for protocol safety.".to_string()
+        ));
+    }
+
+    let distance_km =
+        crate::domain::distance::estimate_distance_km(&merchant.base_pincode, shipping_pincode);
+
+    let pricing_features = crate::application::services::pricing::PricingFeatures {
+        distance_km,
+        user_rate_per_km: merchant.delivery_rate_per_km,
+        product_weight: total_weight,
+        base_charge: merchant.delivery_base_fee,
+        config: serde_json::from_value(merchant.logistics_config.clone()).unwrap_or_default(),
+    };
+
+    let delivery_fee = crate::application::services::pricing::PricingEngine::estimate_delivery_fee(
+        pricing_features,
+    );
+
+    // Precision Geofence Check
+    let mut geofence_verified = None;
+    if let (Some(l_lat), Some(l_lng)) = (lat, lng) {
+        let intelligence =
+            crate::application::services::intelligence::IntelligenceService::new(pool.clone());
+        if !intelligence
+            .verify_geofence_with_precision(shipping_pincode, l_lat, l_lng)
+            .await?
+        {
+            return Err(AppError::Forbidden(format!(
+                "Geofence Verification Failed: Your current GPS coordinates do not match the shipping pincode {}. Forensic integrity active.",
+                shipping_pincode
+            )));
+        }
+        geofence_verified = Some(true);
+    }
+    let platform_fee = crate::application::services::pricing::PricingEngine::calculate_platform_fee(
+        total_price,
+        merchant.trust_score,
+    );
+
+    let gst_breakdown = crate::application::services::india_tax::IndiaTaxService::calculate_gst(
+        total_price,
+        merchant.state_code.unwrap_or(29),
+        shipping_pincode,
+        0.18,
+    );
+
+    let volatility =
+        crate::application::services::intelligence::IntelligenceService::new(pool.clone())
+            .get_pincode_volatility(shipping_pincode)
+            .await
+            .unwrap_or(0.0);
+
+    // Calculate preliminary risk score
+    let mut calculated_risk = 0.0;
+    calculated_risk += (volatility * 50.0).min(30.0);
+    calculated_risk += velocity_risk * 0.4;
+    if geofence_verified == Some(true) {
+        calculated_risk *= 0.8;
+    }
+
+    let mut order = crate::domain::models::OrderRecord {
+        transaction_id: transaction_id.clone(),
+        merchant_id: first_merchant_id.clone(),
+        link_id: "CART_TRANSACTION".into(),
+        buyer_phone: buyer_phone.to_string(),
+        buyer_phone_hash: None, // populated by encrypt_pii() at persist time
+        buyer_name: buyer_name.to_string(),
+        buyer_email: buyer_email.to_string(),
+        shipping_pincode: Some(shipping_pincode.to_string()),
+        delivery_address: Some(delivery_address.to_string()),
+        price_inr: total_price,
+        status: crate::domain::constants::ORDER_STATUS_PENDING_PAYMENT.to_string(),
+        vpa: Some(String::new()),
+        outbound_weight: total_weight,
+        return_weight: 0.0,
+        proof_data: Some(serde_json::json!([])),
+        settled_at: None,
+        shipped_at: None,
+        delivered_at: None,
+        shipping_method: None,
+        estimated_delivery_at: None,
+        payu_id: String::new(),
+        is_payment: false,
+        platform_fee_paid: false,
+        platform_fee,
+        delivery_fee,
+        distance_km,
+        risk_score: calculated_risk,
+        risk_flags: None,
+        cgst: gst_breakdown.cgst,
+        sgst: gst_breakdown.sgst,
+        igst: gst_breakdown.igst,
+        utr_number: None,
+        platform_fee_utr: None,
+        delivery_gps_lat: None,
+        delivery_gps_lng: None,
+        is_geofence_verified: geofence_verified,
+        pincode_volatility_at_checkout: volatility,
+        discount_amount: 0.0,
+        coupon_code: None,
+        checkout_gps_lat: lat,
+        checkout_gps_lng: lng,
+        device_fingerprint: device_fingerprint.clone(),
+        paid_at: None,
+        proof_received_at: None,
+        created_at: None,
+        brand_name: None,
+    };
+
+    let mut tx = pool.begin().await.map_err(AppError::Database)?;
+
+    // Enforce cart inventory atomically within the transaction
+    for (p, qty) in &product_details {
+        if !p.is_unlimited {
+            let rows_affected = sqlx::query(
+                "UPDATE product_links SET inventory_count = inventory_count - $1 WHERE link_id = $2 AND inventory_count >= $3",
+            )
+            .bind(*qty as i32)
+            .bind(&p.link_id)
+            .bind(*qty as i32)
+            .execute(&mut *tx)
+            .await
+            .map_err(AppError::Database)?
+            .rows_affected();
+
+            if rows_affected == 0 {
+                return Err(AppError::BadRequest(format!(
+                    "Product '{}' went out of stock or has insufficient quantity.",
+                    p.product_name
+                )));
+            }
+        }
+    }
+
+    if let Some(ref code) = coupon_code {
+        let coupon = sqlx::query_as::<_, crate::domain::models::Coupon>(
+            "SELECT * FROM coupons WHERE merchant_id = $1 AND code = $2 AND is_active = TRUE",
+        )
+        .bind(&first_merchant_id)
+        .bind(code.to_uppercase())
+        .fetch_optional(&mut *tx)
+        .await?;
+
+        if let Some(c) = coupon {
+            if c.is_valid(order.price_inr) {
+                let discount = c.calculate_discount(order.price_inr);
+                order.discount_amount = discount;
+                order.price_inr -= discount;
+                order.coupon_code = Some(code.clone());
+
+                let _ =
+                    sqlx::query("UPDATE coupons SET usage_count = usage_count + 1 WHERE id = $1")
+                        .bind(c.id)
+                        .execute(&mut *tx)
+                        .await;
+            }
+        }
+    }
+
+    let (risk_score, risk_flags) =
+        crate::application::services::risk::RiskEngine::calculate_risk_score(&order, 0, volatility);
+    order.risk_score = risk_score;
+    order.risk_flags = Some(risk_flags);
+
+    if order.risk_score >= 80.0 {
+        // Log Critical/High Security Event outside the transaction so it's persisted even when rolled back
+        if let Ok(mut conn) = pool.acquire().await {
+            crate::domain::audit::log_risk_event(
+                &mut *conn,
+                Some(&transaction_id),
+                &first_merchant_id,
+                "HIGH_RISK_BLOCK",
+                "CRITICAL",
+                Some(&format!(
+                    "Cart transaction blocked due to high risk score ({:.1}) during checkout. Flags: {:?}",
+                    order.risk_score, order.risk_flags
+                )),
+                Some(order.risk_score),
+                None,
+                device_fingerprint.as_deref(),
+                Some(tx_sender),
+            )
+            .await;
+        }
+
+        if order.risk_score > 90.0 {
+            crate::interfaces::http::middleware::block_ip_persistently(
+                pool,
+                client_ip,
+                &format!(
+                    "Automated Defense: High Risk Score ({:.1}) detected during cart checkout",
+                    order.risk_score
+                ),
+                Some(tx_sender),
+            )
+            .await;
+        }
+
+        return Err(AppError::Forbidden(format!(
+            "Security restriction: High risk profile detected (Score: {:.1}). This transaction has been blocked to prevent potential fraud.",
+            order.risk_score
+        )));
+    } else if order.risk_score > 60.0 {
+        crate::domain::audit::log_risk_event(
+            &mut tx,
+            Some(&transaction_id),
+            &first_merchant_id,
+            "HIGH_RISK_ORDER",
+            "HIGH",
+            Some(&format!(
+                "Cart order {} flagged with risk score {}",
+                transaction_id, order.risk_score
+            )),
+            Some(order.risk_score),
+            None,
+            device_fingerprint.as_deref(),
+            Some(tx_sender),
+        )
+        .await;
+    }
+
+    // Persist Order
+    order.created_at = Some(chrono::Utc::now().naive_utc());
+    crate::domain::models::OrderRecord::create_with_tx(&mut tx, &order).await?;
+
+    // Persist Order Items
+    for (p, qty) in product_details {
+        sqlx::query(
+            "INSERT INTO order_items (transaction_id, product_id, product_name, quantity, price_at_checkout, weight_at_checkout) VALUES ($1, $2, $3, $4, $5, $6)"
+        )
+        .bind(&transaction_id)
+        .bind(&p.link_id)
+        .bind(&p.product_name)
+        .bind(qty as i32)
+        .bind(p.price_inr)
+        .bind(p.expected_weight)
+        .execute(&mut *tx)
+        .await
+        .map_err(AppError::Database)?;
+    }
+    tx.commit().await.map_err(AppError::Database)?;
+
+    let _ = tx_sender.send(crate::interfaces::http::api::RealtimeEvent::NewOrder {
+        transaction_id: transaction_id.clone(),
+        merchant_id: order.merchant_id.clone(),
+        amount: total_price,
+        buyer_phone: buyer_phone.to_string(),
+    });
+
+    Ok(order)
+}

src/application/services/customer.rs

@@ -0,0 +1,189 @@
+use crate::domain::error::{AppError, AppResult};
+use crate::domain::models::{Customer, CustomerProfile, OrderRecord};
+use crate::infrastructure::db::DbPool;
+use argon2::{
+    password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
+    Argon2,
+};
+use async_trait::async_trait;
+use jsonwebtoken::{encode, EncodingKey, Header};
+use serde::{Deserialize, Serialize};
+#[allow(unused_imports)]
+use std::sync::Arc;
+use uuid::Uuid;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct CustomerClaims {
+    pub sub: String, // customer_id
+    pub phone: String,
+    pub role: String,
+    pub exp: usize,
+}
+
+#[async_trait]
+pub trait CustomerService: Send + Sync {
+    async fn signup(
+        &self,
+        phone: &str,
+        password: &str,
+        name: Option<&str>,
+        email: Option<&str>,
+    ) -> AppResult<String>;
+    async fn login(&self, phone: &str, password: &str) -> AppResult<String>;
+    async fn get_orders(
+        &self,
+        customer_id: &str,
+        merchant_id: Option<&str>,
+    ) -> AppResult<Vec<OrderRecord>>;
+    async fn get_profile(&self, customer_id: &str) -> AppResult<Option<CustomerProfile>>;
+}
+
+pub struct RtixCustomerService {
+    pool: DbPool,
+    jwt_secret: Vec<u8>,
+}
+
+impl RtixCustomerService {
+    pub fn new(pool: DbPool, jwt_secret: Vec<u8>) -> Self {
+        Self { pool, jwt_secret }
+    }
+
+    fn hash_password(&self, password: &str) -> AppResult<String> {
+        let salt = SaltString::generate(&mut rand::thread_rng());
+        Argon2::default()
+            .hash_password(password.as_bytes(), &salt)
+            .map(|h| h.to_string())
+            .map_err(|e| AppError::Internal(format!("Hashing failed: {}", e)))
+    }
+
+    fn verify_password(&self, password: &str, hash: &str) -> AppResult<()> {
+        let parsed_hash = PasswordHash::new(hash)
+            .map_err(|_| AppError::Internal("Invalid hash stored".to_string()))?;
+        Argon2::default()
+            .verify_password(password.as_bytes(), &parsed_hash)
+            .map_err(|_| AppError::Auth("Invalid credentials".to_string()))
+    }
+}
+
+#[async_trait]
+impl CustomerService for RtixCustomerService {
+    async fn signup(
+        &self,
+        phone: &str,
+        password: &str,
+        name: Option<&str>,
+        email: Option<&str>,
+    ) -> AppResult<String> {
+        let password_hash = self.hash_password(password)?;
+        let customer_id = Uuid::new_v4().to_string();
+
+        let res = sqlx::query(
+            "INSERT INTO customers (customer_id, phone, name, email, password_hash) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (phone) DO NOTHING"
+        )
+        .bind(&customer_id)
+        .bind(phone)
+        .bind(name)
+        .bind(email)
+        .bind(password_hash)
+        .execute(&self.pool)
+        .await?;
+
+        if res.rows_affected() == 0 {
+            return Err(AppError::Conflict(
+                "An account with this phone number already exists".into(),
+            ));
+        }
+
+        Ok(customer_id)
+    }
+
+    async fn login(&self, phone: &str, password: &str) -> AppResult<String> {
+        let customer = sqlx::query_as::<_, Customer>("SELECT * FROM customers WHERE phone = $1 OR email = $1")
+            .bind(phone)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| AppError::Auth("Invalid credentials".into()))?;
+
+        self.verify_password(password, &customer.password_hash)?;
+
+        let exp = (chrono::Utc::now() + chrono::Duration::days(7)).timestamp() as usize;
+        let claims = CustomerClaims {
+            sub: customer.customer_id.clone(),
+            phone: customer.phone.clone(),
+            role: "customer".into(),
+            exp,
+        };
+
+        let token = encode(
+            &Header::default(),
+            &claims,
+            &EncodingKey::from_secret(&self.jwt_secret),
+        )
+        .map_err(|_| AppError::Internal("Token generation failed".into()))?;
+
+        Ok(token)
+    }
+
+    async fn get_orders(
+        &self,
+        customer_id: &str,
+        merchant_id: Option<&str>,
+    ) -> AppResult<Vec<OrderRecord>> {
+        use crate::core::crypto::CryptoService;
+        use sqlx::Row;
+
+        // 1. Retrieve the customer's plaintext phone number from the auth table
+        let row = sqlx::query("SELECT phone FROM customers WHERE customer_id = $1")
+            .bind(customer_id)
+            .fetch_optional(&self.pool)
+            .await?
+            .ok_or_else(|| AppError::NotFound("Customer not found".into()))?;
+        let target_phone: String = row.get("phone");
+
+        // 2. Compute a deterministic HMAC-SHA256 blind index of the phone number.
+        //    This allows an O(1) indexed DB lookup without exposing PII in the query.
+        let phone_hash = CryptoService::deterministic_hash(&target_phone);
+
+        // 3. Run an indexed query — hit buyer_phone_hash directly, no full-table scan.
+        let raw_orders: Vec<OrderRecord> = if let Some(m_id) = merchant_id {
+            sqlx::query_as::<_, OrderRecord>(
+                "SELECT o.*, m.brand_name FROM orders o LEFT JOIN merchants m ON o.merchant_id = m.merchant_id WHERE o.buyer_phone_hash = $1 AND o.merchant_id = $2 ORDER BY o.created_at DESC",
+            )
+            .bind(&phone_hash)
+            .bind(m_id)
+            .fetch_all(&self.pool)
+            .await?
+        } else {
+            sqlx::query_as::<_, OrderRecord>(
+                "SELECT o.*, m.brand_name FROM orders o LEFT JOIN merchants m ON o.merchant_id = m.merchant_id WHERE o.buyer_phone_hash = $1 ORDER BY o.created_at DESC",
+            )
+            .bind(&phone_hash)
+            .fetch_all(&self.pool)
+            .await?
+        };
+
+        // 4. Decrypt only the matched rows (typically a very small set per customer)
+        let matched_orders = raw_orders
+            .into_iter()
+            .map(|mut order| {
+                order.decrypt_pii();
+                if order.vpa.as_deref() == Some("") {
+                    order.vpa = None;
+                }
+                order
+            })
+            .collect();
+
+        Ok(matched_orders)
+    }
+
+    async fn get_profile(&self, customer_id: &str) -> AppResult<Option<CustomerProfile>> {
+        let profile = sqlx::query_as::<_, CustomerProfile>(
+            "SELECT customer_id, phone, email, name FROM customers WHERE customer_id = $1",
+        )
+        .bind(customer_id)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(profile)
+    }
+}