Provider Routing Priorities: How 0fee Chooses the Best Path

pythonasync def get_provider_for_unified_method(
    payment_method: str,
    app_id: str,
    environment: str = "sandbox",
) -> tuple:
    """Select the best provider for a payment method.

    Returns (provider_instance, provider_method_code) tuple.
    """
    # Get all routing entries for this method, ordered by priority
    routing_entries = await db.fetch_all(
        """
        SELECT pr.provider_id, pr.provider_method_code, pr.priority
        FROM provider_routing pr
        JOIN payin_methods pm ON pr.payin_method_id = pm.id
        WHERE pm.code = :method
        AND pr.is_active = 1
        ORDER BY pr.priority ASC
        """,
        {"method": payment_method},
    )

    for entry in routing_entries:
        provider_id = entry["provider_id"]
        method_code = entry["provider_method_code"]

        # Check if app has credentials for this provider
        if await _has_credentials(app_id, provider_id, environment):
            # Instantiate the provider with app's credentials
            provider = await _get_provider_instance(
                provider_id, app_id, environment
            )
            return (provider, method_code)

    raise NoProviderConfiguredError(
        f"No provider configured for {payment_method}. "
        "Add provider credentials in your dashboard."
    )

sql-- Payment methods (117 entries)
CREATE TABLE payin_methods (
    id TEXT PRIMARY KEY,
    code TEXT UNIQUE NOT NULL,      -- e.g., PAYIN_ORANGE_CI
    name TEXT NOT NULL,              -- e.g., Orange Money (Ivory Coast)
    country TEXT NOT NULL,           -- e.g., CI
    currency TEXT NOT NULL,          -- e.g., XOF
    operator TEXT,                   -- e.g., Orange
    type TEXT NOT NULL,              -- e.g., mobile_money, card, wallet
    is_active INTEGER DEFAULT 1,
    created_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Provider routing (many-to-many with priority)
CREATE TABLE provider_routing (
    id TEXT PRIMARY KEY,
    provider_id TEXT NOT NULL,           -- e.g., paiementpro
    payin_method_id TEXT NOT NULL,       -- FK to payin_methods
    provider_method_code TEXT NOT NULL,  -- e.g., OMCIV2
    priority INTEGER NOT NULL DEFAULT 1,
    is_active INTEGER DEFAULT 1,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    FOREIGN KEY (payin_method_id) REFERENCES payin_methods(id),
    UNIQUE(provider_id, payin_method_id)
);

pythonasync def _has_credentials(
    app_id: str, provider_id: str, environment: str
) -> bool:
    """Check if app has credentials configured for a provider."""
    if provider_id == "test":
        return True  # Test provider needs no credentials

    credential = await db.fetch_one(
        """
        SELECT id FROM app_provider_credentials
        WHERE app_id = :app_id
        AND provider_id = :provider_id
        AND environment = :environment
        AND is_active = 1
        """,
        {
            "app_id": app_id,
            "provider_id": provider_id,
            "environment": environment,
        },
    )

    return credential is not None

pythonasync def route_payment_with_fallback(
    payment_method: str,
    app_id: str,
    environment: str,
    data: dict,
) -> dict:
    """Route payment with automatic fallback to next provider."""
    routing_entries = await get_routing_entries(payment_method)
    last_error = None

    for entry in routing_entries:
        provider_id = entry["provider_id"]

        if not await _has_credentials(app_id, provider_id, environment):
            continue

        try:
            provider = await _get_provider_instance(
                provider_id, app_id, environment
            )
            result = await provider.initiate_payment(data)

            if result.get("status") != "failed":
                # Store which provider was used
                data["actual_provider"] = provider_id
                return result

            last_error = result.get("error", "Provider returned failure")

        except Exception as e:
            last_error = str(e)
            continue  # Try next provider

    raise AllProvidersFailedError(
        f"All providers failed for {payment_method}. "
        f"Last error: {last_error}"
    )

python# Session 001: Hardcoded routing
PROVIDER_MAP = {
    "PAYIN_ORANGE_CI": [
        ("paiementpro", "OMCIV2"),
        ("hub2", "ORANGE_MONEY"),
    ],
    "PAYIN_CARD": [
        ("stripe", "card"),
    ],
}

pythonasync def seed_routing_table():
    """Seed the provider routing table with default priorities."""

    entries = [
        # Orange Money Ivory Coast
        ("paiementpro", "PAYIN_ORANGE_CI", "OMCIV2", 1),
        ("pawapay", "PAYIN_ORANGE_CI", "ORANGE_CIV", 2),
        ("hub2", "PAYIN_ORANGE_CI", "ORANGE_MONEY", 3),
        ("bui", "PAYIN_ORANGE_CI", "orange_ci", 4),

        # MTN Ivory Coast
        ("pawapay", "PAYIN_MTN_CI", "MTN_CIV", 1),
        ("hub2", "PAYIN_MTN_CI", "MTN_MOMO", 2),
        ("bui", "PAYIN_MTN_CI", "mtn_ci", 3),

        # Wave Ivory Coast
        ("paiementpro", "PAYIN_WAVE_CI", "WAVECI", 1),
        ("hub2", "PAYIN_WAVE_CI", "WAVE", 2),
        ("bui", "PAYIN_WAVE_CI", "wave_ci", 3),

        # ... 100+ more entries

        # Global
        ("stripe", "PAYIN_CARD", "card", 1),
        ("paypal", "PAYIN_PAYPAL", "paypal", 1),

        # Test provider (always last)
        ("test", "PAYIN_ORANGE_CI", "TEST_ORANGE_CI", 100),
        ("test", "PAYIN_CARD", "TEST_CARD", 100),
        # ... test entries for all methods
    ]

    for provider_id, method_code, provider_method_code, priority in entries:
        method = await get_payin_method_by_code(method_code)
        if method:
            await upsert_routing_entry(
                provider_id=provider_id,
                payin_method_id=method["id"],
                provider_method_code=provider_method_code,
                priority=priority,
            )

GET  /v1/apps/{app_id}/routes          -- List routing rules for app
POST /v1/apps/{app_id}/routes          -- Create custom routing rule
DELETE /v1/apps/{app_id}/routes/{id}   -- Delete routing rule

GET  /v1/payin-methods/{code}          -- Get method with providers
GET  /v1/providers/{id}/methods        -- Get provider's methods

json{
    "code": "PAYIN_ORANGE_CI",
    "name": "Orange Money (Ivory Coast)",
    "country": "CI",
    "currency": "XOF",
    "providers": [
        {"provider_id": "paiementpro", "priority": 1, "code": "OMCIV2"},
        {"provider_id": "pawapay", "priority": 2, "code": "ORANGE_CIV"},
        {"provider_id": "hub2", "priority": 3, "code": "ORANGE_MONEY"},
        {"provider_id": "bui", "priority": 4, "code": "orange_ci"},
        {"provider_id": "test", "priority": 100, "code": "TEST_ORANGE_CI"}
    ]
}

pythondef parse_payment_method(method: str) -> dict:
    """Parse unified payment method.

    PAYIN_ORANGE_CI -> {operator: "ORANGE", country: "CI"}
    PAYIN_MPESA_KE -> {operator: "MPESA", country: "KE"}
    PAYIN_CARD -> {operator: "CARD", country: None}
    """
    parts = method.replace("PAYIN_", "").rsplit("_", 1)

    if len(parts) == 2:
        return {"operator": parts[0], "country": parts[1]}
    else:
        return {"operator": parts[0], "country": None}