# Streaming Provider Resolver Pattern

## Context

When building a streaming app backend with Modular Clean Architecture, the provider resolver layer bridges the application use case and external debrid provider APIs (Real-Debrid, AllDebrid, etc.).

## Module Structure

```
app/modules/streaming/
├── domain/
│   ├── value_objects.py     # ResolveSource, StreamResolveResult, StreamFile, StreamQuality, ResolveSourceType
│   ├── entities.py          # StreamResolveRequest
│   └── __init__.py
├── application/
│   ├── interfaces.py        # DebridAccountLookupPort, ProviderStreamResolver (Protocol)
│   ├── exceptions.py        # StreamResolveError + subclasses (public_message pattern)
│   ├── use_cases.py         # StreamResolveUseCase (account selection + resolve orchestration)
│   └── __init__.py
├── infrastructure/
│   ├── provider_resolvers.py  # RealDebridResolver implementation
│   ├── account_lookup.py      # DebridAccountLookup adapter (wraps debrid repo)
│   └── __init__.py
└── presentation/
    ├── schemas.py             # Pydantic v2: ResolveRequest, StreamResultResponse, ResolveResponse
    ├── dependencies.py        # Auth, use case wiring, resolver lifecycle management
    └── routers.py             # POST /api/v1/streams/resolve
```

### Cross-module infrastructure adapter

The streaming module needs to look up debrid account credentials and verify ownership. Rather than importing the debrid module's repository directly into the streaming use case, create an **infrastructure adapter** in the streaming module:

```python
# infrastructure/account_lookup.py
class DebridAccountLookup(DebridAccountLookupPort):
    """Wraps the debrid module's repository to implement the streaming port."""

    def __init__(self, session: AsyncSession) -> None:
        self._repo = SQLAlchemyDebridAccountRepository(session)

    async def get_api_key(self, account_id: int) -> str | None:
        return await self._repo.get_api_key(account_id)

    async def get_account_provider(
        self, account_id: int, user_id: int
    ) -> str | None:
        account = await self._repo.get_by_id(account_id)
        if account is None or account.user_id != user_id:
            return None
        return account.provider
```

This keeps the streaming module self-contained — the presentation layer wires the adapter, but the application layer only depends on the port protocol.

### Resolver lifecycle management (parallel registries)

Stream resolvers follow the same app-lifespan pattern as debrid provider clients. Create parallel registries that coexist without importing each other:

```python
# presentation/dependencies.py
def _build_stream_resolvers() -> dict[str, ProviderStreamResolver]:
    return {
        "real_debrid": RealDebridResolver(),
        "all_debrid": AllDebridResolver(),
        "premiumize": PremiumizeResolver(),
        "torbox": TorBoxResolver(),
    }

STREAM_RESOLVERS: dict[str, ProviderStreamResolver] = {}

def ensure_stream_resolvers() -> dict[str, ProviderStreamResolver]:
    if not STREAM_RESOLVERS:
        STREAM_RESOLVERS.update(_build_stream_resolvers())
    return STREAM_RESOLVERS

async def close_stream_resolvers(resolvers=None) -> None:
    registry = resolvers if resolvers is not None else STREAM_RESOLVERS
    for name, resolver in registry.items():
        if hasattr(resolver, "close"):
            await resolver.close()
    if resolvers is None:
        STREAM_RESOLVERS.clear()
```

In `main.py`, both registries are initialized in the same lifespan:

```python
_app.state.provider_clients = _build_provider_clients()
_app.state.stream_resolvers = _build_stream_resolvers()
# ... shutdown:
await close_provider_clients(_app.state.provider_clients)
await close_stream_resolvers(_app.state.stream_resolvers)
```

This keeps the two systems independent — adding a resolver doesn't touch debrid code and vice versa.

### Frozen dataclass mutation with dataclasses.replace

`StreamResolveResult` is a frozen dataclass — the resolver creates instances with `account_id=0` since it doesn't know the real ID. The use case fixes this after resolution using `dataclasses.replace()`:

```python
from dataclasses import replace

results = await resolver.resolve(request.source, api_key)
return [replace(r, account_id=account_id) for r in results]
```

This is cleaner than `object.__setattr__` hacks — it creates new instances with the corrected field while preserving all other values.

### Source auto-detection (from_string)

For API input, `ResolveSource` has a `from_string` classmethod that auto-detects the source type:

```python
@classmethod
def from_string(cls, source: str) -> "ResolveSource":
    if source.startswith("magnet:"):
        return cls(source_type=ResolveSourceType.MAGNET, value=source)
    if source.startswith("http://") or source.startswith("https://"):
        return cls(source_type=ResolveSourceType.DIRECT_URL, value=source)
    if _HEX_PATTERN.match(source):
        return cls(source_type=ResolveSourceType.INFO_HASH, value=source)
    if _BASE32_PATTERN.match(source):
        return cls(source_type=ResolveSourceType.INFO_HASH, value=source)
    raise ValueError(f"Unknown source type: {source}")
```

**Pitfall:** Torrent URLs start with http/https and are auto-detected as `DIRECT_URL`. This is acceptable since both types flow through the same debrid API path, but worth noting if distinction ever matters.

### Presentation layer patterns

The resolve endpoint uses **semantic HTTP status codes** rather than a 200 envelope:

| HTTP Status | Meaning | Response Body |
|---|---|---|
| 200 | Success | `{"streams": [...]}` |
| 404 | No account / no streams | `{"error": "..."}` |
| 422 | Invalid or unsupported source | `{"error": "..."}` |
| 502 | Provider unavailable | `{"error": "..."}` |
| 500 | Unexpected error | `{"error": "..."}` |

Response schemas:

```python
class ResolveSuccessResponse(BaseModel):
    streams: list[StreamResultResponse]

class ErrorResponse(BaseModel):
    error: str
```

This gives mobile clients clear HTTP-level signals to surface appropriate UI (error dialog vs no-results message vs retry). The 502 Bad Gateway code is particularly important for provider outages — clients can show a "service temporarily unavailable" message distinct from "no content found."

### Integration testing pattern

Full-stack integration tests follow the debrid pattern exactly:

1. **Module-level crypto patches** — `encrypt_api_key` / `decrypt_api_key` as identity
2. **SQLite in-memory DB** — `create_async_engine("sqlite+aiosqlite:///:memory:")`
3. **ASGITransport** — `AsyncClient(transport=ASGITransport(app=app), base_url="http://test")`
4. **`create_app()`** — uses lifespan which initializes resolvers
5. **Swap `_client` on resolvers** — not the registry dict, only the HTTP attribute

```python
from app.modules.streaming.presentation.dependencies import ensure_stream_resolvers
stream_resolvers = ensure_stream_resolvers()
mock_transport = httpx.MockTransport(_resolve_handler)
for name, resolver in stream_resolvers.items():
    if hasattr(resolver, "_client"):
        resolver._client = httpx.AsyncClient(transport=mock_transport, timeout=15.0)
```

**Multi-resolver mock transport:** The `_resolve_handler` must handle ALL provider endpoints. Route by URL patterns (e.g. `"alldebrid" in url_str`, `"premiumize" in url_str`, `"torbox" in url_str`):

```python
def _resolve_handler(request: httpx.Request) -> httpx.Response:
    url_str = str(request.url)
    if "unrestrict/link" in url_str:
        return httpx.Response(200, json={...})  # Real-Debrid direct or DDL unrestrict
    if "addMagnet" in url_str:
        return httpx.Response(201, json={"id": 12345, "uri": "..."})  # RD addMagnet
    if "selectFiles" in url_str:
        return httpx.Response(204)  # RD selectFiles
    if "torrents/info" in url_str:
        return httpx.Response(200, json={"id": 12345, "filename": "...", "links": [...], ...})  # RD info
    if "alldebrid" in url_str and "link/unrestrict" in url_str:
        return httpx.Response(200, json={"status": "success", "data": {...}})  # AllDebrid
    if "premiumize" in url_str and "torrent/create" in url_str:
        return httpx.Response(200, json={"status": "success", "id": "pm_..."})  # Premiumize
    if "torbox" in url_str and "createtorrent" in url_str:
        return httpx.Response(200, json={"status": "success", "data": {...}})  # TorBox
    return httpx.Response(404, json={"error": "Not found"})
```

Test categories to cover:
- Auth required (401 without token)
- No debrid account (graceful error, not 500)
- Direct URL success path (full round-trip through resolver)
- Magnet success path
- Other user's account denied (cross-user ownership check)
- Invalid source (Pydantic 422 for empty, domain validation error for malformed)
- API key not leaked in any error response

---

## Key Implementation Details

### ResolveSource Value Object

```python
@dataclass(frozen=True, slots=True)
class ResolveSource:
    source_type: ResolveSourceType  # magnet, torrent_url, direct_url, info_hash
    value: str                       # <-- field is "value", NOT "source_value"
```

**Pitfall:** The field is `value`, not `source_value`. This is a common mistake when delegating — subagents and manual implementations often use `source_value` which causes `TypeError` at runtime.

### Exception Pattern

Follow the debrid module's `public_message` + `__init__` pattern:

```python
class StreamResolveError(Exception):
    public_message = "Stream resolution failed"
    def __init__(self, message: str | None = None):
        super().__init__(message or self.public_message)
```

Each subclass defines its own `public_message`. The base class handles the `__init__` logic.

### Provider Resolver Implementation

All resolvers share these module-level helpers in `provider_resolvers.py`:

```python
async def _request_json(client, method, url, *, retry_once=True, **kwargs) -> Mapping[str, Any]:
    """Shared by all resolvers — retries, HTTP→exception translation, JSON parsing."""

async def _request_no_content(client, method, url, *, retry_once=True, **kwargs) -> None:
    """Like _request_json but for endpoints returning 204 No Content (e.g., selectFiles).
    Raises ProviderStreamUnavailableError for non-2xx responses. No JSON parsing."""

def _stream_url(value: Any) -> str:
    """Validate provider-returned URL is absolute HTTP(S), not empty/script/file: etc."""
    if not isinstance(value, str) or not value.strip():
        raise ProviderStreamUnavailableError()
    ...

def _size_bytes(value: Any) -> int:
    """Coerce to non-negative int, rejecting NaN/Inf."""
    ...

def _is_safe_stream_url(value: str) -> bool:
    """Check scheme, netloc, and forbid userinfo."""
    ...
```

Each resolver class follows this template:

```python
class SomeDebridResolver(ProviderStreamResolver):
    provider: str = "some_debrid"
    BASE_URL: str = "https://api.someprovider.com/v1"

    def __init__(self, http_client: httpx.AsyncClient | None = None):
        self._owns_client = http_client is None
        self._client = http_client or httpx.AsyncClient(
            timeout=_DEFAULT_TIMEOUT,
            limits=_DEFAULT_LIMITS,
        )

    async def close(self) -> None:
        if self._owns_client:
            await self._client.aclose()

    async def resolve(self, source: ResolveSource, api_key: str) -> list[StreamResolveResult]:
        if source.source_type == ResolveSourceType.DIRECT_URL:
            return await self._resolve_direct(source, params_or_headers)
        if source.source_type in (MAGNET, INFO_HASH):
            return await self._resolve_torrent(source, params_or_headers)
        raise UnsupportedStreamSourceError()
```

**Pitfall — shared `_request_json` extraction:** When the first resolver has `_request_json` as an instance method, adding a second resolver forces extraction to module level. Extract it proactively before adding the second resolver to avoid back-patching three call sites.

### HTTP Configuration (mirrors debrid module)

```python
_DEFAULT_TIMEOUT = httpx.Timeout(connect=3.0, read=10.0, write=5.0, pool=2.0)
_DEFAULT_LIMITS = httpx.Limits(max_connections=100, max_keepalive_connections=20)
_RETRYABLE_STATUS_CODES = {502, 503, 504}
```

### Error Translation

All HTTP errors → `ProviderStreamUnavailableError`. Never leak API keys, URLs, or raw provider responses.

```python
async def _request_json(self, method, url, *, retry_once=True, **kwargs):
    # Retry logic for 502/503/504
    # 401/403 → ProviderStreamUnavailableError
    # 429 → ProviderStreamUnavailableError
    # 500+ → ProviderStreamUnavailableError
    # 400+ → ProviderStreamUnavailableError
    # ValueError (JSON parse) → ProviderStreamUnavailableError
```

### Quality Inference

Infer `StreamQuality` from filename patterns:

```python
_QUALITY_PATTERNS = [
    ("4k", StreamQuality.UHD),
    ("2160p", StreamQuality.UHD),
    ("1080p", StreamQuality.FHD),
    ("720p", StreamQuality.HD),
    ("480p", StreamQuality.SD),
    ("360p", StreamQuality.SD),
    ("240p", StreamQuality.SD),
    ("audio", StreamQuality.UNKNOWN),
]
```

### MIME Type Inference

```python
_MIME_MAP = {
    ".mp4": "video/mp4",
    ".mkv": "video/x-matroska",
    # ... etc
}

def _infer_mime_type(filename: str) -> str | None:
    if not filename or "." not in filename:  # Guard against "unknown"
        return None
    _, ext = filename.rsplit(".", 1)
    return _MIME_MAP.get(f".{ext.lower()}")
```

**Pitfall:** Filenames like "unknown" (no extension) crash `rsplit`. Always check for `.` before splitting.

### Multi-Provider Support

When adding multiple provider resolvers, extract shared HTTP helpers to module-level functions so all resolvers can use them:

```python
async def _request_json(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    *,
    retry_once: bool = True,
    **kwargs: Any,
) -> Mapping[str, Any]:
    """Module-level helper shared by all resolvers."""
```

### Provider Auth Methods

| Provider | Auth Mechanism | How to Pass |
|----------|---------------|-------------|
| Real-Debrid | Bearer token | `headers={"Authorization": f"Bearer {api_key}"}` |
| AllDebrid | Query parameter | `params={"apikey": api_key}` |
| Premiumize | Query parameter | `params={"apikey": api_key}` |
| TorBox | Query parameter | `params={"apikey": api_key}` |

**Key insight:** Design each resolver to accept `api_key: str` in `resolve()`. The resolver decides how to pass it (header vs query param). Never leak auth in exceptions/logs.

### Provider Capability Matrix

| Provider | Direct URL | Magnet | Info-Hash | Torrent URL |
|----------|-----------|--------|-----------|-------------|
| Real-Debrid | ✅ | ✅ | ✅ | ❌ |
| AllDebrid | ✅ | ✅ | ✅ | ❌ |
| Premiumize | ❌ | ✅ | ✅ | ❌ |
| TorBox | ❌ | ✅ | ✅ | ❌ |

### AllDebrid API Flows

**Base URL:** `https://api.alldebrid.com/v5`  
**Auth:** `?apikey={key}` on every request  
**Response envelope:** `{"status": "success"|"error", "data": {...}, "error": {"code": "...", "message": "..."}}`

**Direct URL Flow:**
```
POST /link/unrestrict?apikey={key}
data: {"link": "https://..."}
→ {"status": "success", "data": {"filename": "...", "size": ..., "link": "https://..."}}
```

**Magnet Flow:**
```
Step 1: POST /magnet/upload?apikey={key}
        data: {"magnets": "magnet:?xt=urn:btih:..."}
        → {"status": "success", "data": {"magnets": [{"id": ..., "hash": "..."}]}}

Step 2: GET /magnet/status?apikey={key}&id={id}
        → {"status": "success", "data": {"magnets": [{"id": ..., "links": [{"link": "...", "filename": "...", "size": ...}]}]}}
```

**Error handling:** Check `data["status"] != "success"` before accessing `data["data"]`. The `status` field exists even on HTTP 200 responses.

### Premiumize API Flows

**Base URL:** `https://www.premiumize.me/api/v1`  
**Auth:** `?apikey={key}` on every request  
**Response envelope:** `{"status": "success"|"error", "content": [...]|{...}, "id": ...}`

**Magnet Flow (torrent-only, no direct URL support):**
```
Step 1: POST /torrent/create?apikey={key}
        data: {"src": "magnet:?xt=urn:btih:..."}
        → {"status": "success", "id": "pm_12345"}

Step 2: GET /torrent/browse?apikey={key}&id={id}
        → {"status": "success", "content": [{"id": "f1", "name": "file.mkv", "size": 1500000000, "stream_url": "https://..."}]}
```

**Fallback:** If `stream_url` is not present in browse content, fetch per-file details:
```
GET /item/details?apikey={key}&id={file_id}
→ {"status": "success", "content": {"stream_url": "https://..."}}
```

### TorBox API Flows

**Base URL:** `https://api.torbox.app/api`  
**Auth:** `?apikey={key}` on every request  
**Response envelope:** `{"status": "success"|"error", "data": [...]|{...}}`

**Magnet Flow (torrent-only, no direct URL support):**
```
Step 1: POST /torrents/createtorrent?apikey={key}
        data: {"magnet": "magnet:?xt=urn:btih:..."}
        → {"status": "success", "data": {"torrent_id": "tb_..."}}

Step 2: GET /torrents/mylist?apikey={key}
        → {"status": "success", "data": [{"id": "...", "files": [{"url": "https://...", "name": "...", "size": ...}]}]}
```

**Field name quirk:** TorBox uses `url` (not `link` or `stream_url`) as the primary stream URL field. Support both: `file_item.get("url") or file_item.get("stream_url")`.

### Real-Debrid API Endpoints Used

| Endpoint | Method | Purpose |
|----------|--------|---------|
| `/torrents/addMagnet` | POST | Add magnet link (form body: `magnet=`) |
| `/torrents/selectFiles/{id}` | POST | Select files to process (`files=all`) |
| `/torrents/info/{id}` | GET | Get torrent files/links/status (numeric ID) |
| `/unrestrict/link` | POST | Unrestrict a hoster/download link for playback |
| `/torrents/instantAvailability/{hash}` | GET | Check if cached |

**⚠️ Common mistakes (historical):**
- `POST /torrents/add/{hash}` does NOT exist — always use `addMagnet`
- `GET /torrents/stream/{hash}/{file}` does NOT exist — always use `unrestrict/link`
- Info endpoint uses numeric torrent ID, NOT info-hash
- `selectFiles` is mandatory or links won't be available

### Testing Strategy

Use `httpx.MockTransport` for HTTP mocking. Write hand-written stubs for ports (not `MagicMock`). Test:
- Success paths (magnet resolve, direct URL resolve)
- Error paths (timeout, 401, 429, 500)
- API key not leaked in exceptions
- Quality/MIME inference edge cases
- `_extract_hash` static method

## Account Selection Priority

The `StreamResolveUseCase` supports three account selection modes:

1. **Explicit `account_id`:** Verify ownership via `get_account_provider(account_id, user_id)`, then fetch API key.
2. **Provider-only (`provider` without `account_id`):** List user accounts via `list_by_user(user_id)`, find matching provider, fetch API key.
3. **Auto-select (neither specified):** List user accounts, pick first available, fetch API key.

```python
if account_id is not None:
    # Priority 1: explicit ID
    ...
else:
    accounts = await self.account_lookup.list_by_user(user_id)
    if provider_name is not None:
        # Priority 2: match by provider
        matching = [a for a in accounts if a.provider == provider_name]
        if not matching:
            raise NoDebridAccountError()
        account_id = matching[0].account_id
    else:
        # Priority 3: first available
        if not accounts:
            raise NoDebridAccountError()
        account_id = accounts[0].account_id
        provider_name = accounts[0].provider
    api_key = await self.account_lookup.get_api_key(account_id)
```

**Port interface:**

```python
class DebridAccountLookupPort(Protocol):
    async def get_api_key(self, account_id: int) -> str | None: ...
    async def get_account_provider(self, account_id: int, user_id: int) -> str | None: ...
    async def list_by_user(self, user_id: int) -> list[AccountInfo]: ...

class AccountInfo:
    def __init__(self, account_id: int, provider: str) -> None: ...
```

This enables full auto-selection without needing a separate list-accounts endpoint in the presentation layer.

## Real-Debrid Torrent Flow

1. Extract info-hash from magnet: `magnet:?xt=urn:btih:{hash}`
2. `POST /torrents/addMagnet` with `data={"magnet": magnet_url}` → returns `{"id": 12345, ...}`
3. `POST /torrents/selectFiles/{id}` with `data={"files": "all"}` → returns 204
4. `GET /torrents/info/{id}` → returns `files[]`, `links[]` (download URLs)
5. For each link in `links[]`: `POST /unrestrict/link` with `data={"link": ddl_url}` → returns playable `link`
6. Map to `StreamResolveResult` objects

**Pitfalls:**
- The torrent ID from step 2 (numeric) is what steps 3-4 need — NOT the info-hash
- The `selectFiles` step is mandatory; RD won't process the torrent without it
- There is no `/torrents/stream/{hash}/{file}` endpoint — unrestrict download links instead
- The unrestrict response uses `filesize` (not `size`) for byte count

## Real-Debrid Direct URL Flow

1. `POST /unrestrict/link` with `data={"link": direct_url}`
2. Parse response: `filename`, `filesize`, `link` (stream URL)
3. Map to single `StreamResolveResult`