# Streaming Source Pipeline Architecture (Kotlin/Android TV, 2026)

Condensed from AIOStreams (Node.js), Comet (Python/FastAPI), and MediaFusion (Rust) — applied to a Hilt-injected Kotlin/Compose Android TV app.

## The Pipeline (Chain of Responsibility)

```
User selects content
  │
  ├─ 1. Fetcher — parallel HTTP to all StreamProviders
  │     ProviderRegistry.fetchAllStreams(type, id)
  │     Each provider: GET {baseUrl}/stream/{type}/{id}.json
  │     → List<Stream> with infoHash, name, title
  │
  ├─ 2. Parser — extract metadata from stream titles
  │     StreamFactExtractor: resolution, quality, codec, HDR, audio, language
  │     → ParsedStream with structured attributes
  │
  ├─ 3. Deduplicator — remove duplicates
  │     StreamDeduplicator: union-find by infoHash, URL, smart hash
  │     → deduplicated List<Stream>
  │
  ├─ 4. Enricher — bridge P2P infoHash → debrid cache  ← THE CRITICAL STAGE
  │     DebridEnricher.enrich(streams)
  │     Fans out to all active DebridApis: checkCachePerProvider(hashes)
  │     Marks isDebridCached=true, sets debridProvider, adjusts healthScore
  │     → enriched List<Stream> with cache metadata
  │
  ├─ 5. Filterer — user-defined rules
  │     StreamFilterEngine.apply(streams, RegexFilterRule[])
  │     Exclude/include/require/boost by resolution, quality, codec, size, release group
  │     → filtered List<Stream>
  │
  ├─ 6. Ranker — multi-criteria sort
  │     SourceRanker.rankStreams(streams, addonHealth, preferences)
  │     Cached-first partition, then resolution > quality > size > seeders
  │     → List<RankedStream>
  │
  └─ 7. UI — SourceListState with cache badges
       HeroState.hasStreams, SourceListState.streams, CacheBadge
```

## Key Interfaces

### StreamProvider
```kotlin
interface StreamProvider {
    val metadata: ProviderMetadata
    suspend fun getStreams(type: String, id: String): Result<List<Stream>>
    suspend fun isAvailable(): Boolean
}

data class ProviderMetadata(
    val id: String,           // "torrentio", "comet", "mediafusion"
    val name: String,         // display name
    val baseUrl: String,      // manifest base URL
    val supportedServices: List<DebridProvider>,
    val supportedResources: List<String>,
    val priority: Int,        // higher = preferred
)
```

### DebridEnricher
Bridges the gap between stream fetching (returns infoHashes) and debrid resolution (returns cache status). This is the stage that was MISSING — causing "source lists not appearing."

```kotlin
@Singleton
class DebridEnricher @Inject constructor(
    private val debridService: DebridService,
) {
    suspend fun enrich(streams: List<Stream>): List<Stream> {
        // 1. Partition: P2P (infoHash, no url) vs Resolved (has url)
        // 2. Extract unique infoHashes
        // 3. Fan-out: DebridService.checkCachePerProvider(hashes)
        // 4. Map results back to streams: isDebridCached, debridProvider, healthScore
        // 5. Return enriched list
    }
}
```

### ProviderRegistry
Aggregates all providers, fan-out fetches:
```kotlin
@Singleton
class ProviderRegistry @Inject constructor(
    torrentioProvider: dagger.Lazy<TorrentioProvider>,
    noTorrentProvider: dagger.Lazy<NoTorrentProvider>,
    // ... comet, mediafusion stubs
) {
    suspend fun fetchAllStreams(type: String, id: String): Result<List<Stream>> {
        // Query each enabled provider
        // Concatenate results
        // Return combined list (dedup happens downstream)
    }
}
```

## Debrid Integration

### Cache Checking (Batch)
```kotlin
// DebridService public API added for batch enrichment:
suspend fun checkCachePerProvider(
    hashes: List<String>
): Map<DebridProvider, Map<String, InstantAvailability>>
```

### Lazy Resolution (Playback Time)
In `PlaybackOrchestrator.prepare()`:
```kotlin
val url = when {
    !stream.debridUrl.isNullOrBlank() -> stream.debridUrl
    !stream.url.isNullOrBlank() -> stream.url
    stream.infoHash != null && stream.isDebridCached && stream.debridProvider != null -> {
        // Lazy resolve: infoHash → debrid API → playable URL
        debridService.resolveStream(stream, DebridProvider.valueOf(stream.debridProvider!!))
    }
    else -> null  // skip
}
```

## Stream Model (Enriched Fields)

```kotlin
data class Stream(
    // From addon JSON:
    val url: String? = null,
    val infoHash: String? = null,
    val title: String? = null,
    val name: String? = null,
    
    // Enriched by pipeline:
    val addonName: String? = null,      // set by StremioAddonClient
    val addonId: String? = null,        // set by StremioAddonClient
    val isDebridCached: Boolean = false, // set by DebridEnricher
    val debridProvider: String? = null,  // set by DebridEnricher
    val debridUrl: String? = null,       // set at playback time
    val resolvedUrl: String? = null,     // set at playback time
    val healthScore: Float = 1.0f,       // set by DebridEnricher
)
```

## Provider Implementations

### TorrentioProvider (Primary)
- Constructs URL from ALL debrid accounts: `realdebrid=TOK1|alldebrid=TOK2`
- Fetches via Stremio protocol
- Enriches through DebridEnricher

### NoTorrentProvider (Fallback)
- Free HTTP streams, no debrid required
- Lower priority (5 vs Torrentio's 10)

### CometProvider / MediaFusionProvider (Stubs)
- Self-hosted addons, user-configurable URL
- Stubs return empty — full implementation needs user server URL

## Debug Logging Pattern

Every pipeline stage logs:
```
[AddonRepo] getStreamsForContent: type=movie id=tt0111161
[AddonRepo] Provider registry: 45 streams
[DebridEnricher] Enriching 30 P2P streams (15 already resolved)
[DebridEnricher] 18/30 P2P cached, 15 resolved, 2 providers checked
[AddonRepo] Enriched: cached=18/45
[AddonRepo] Filter: 45 → 42 streams (3 excluded by regex)
```

For individual addon HTTP requests:
```
[AddonClient] [torrentio] Request: https://torrentio.strem.fun/.../stream/movie/tt0111161.json
[AddonClient] [torrentio] Response: 50 streams, urlField=25, hashField=25
```

## Integration Pattern (AddonRepository)

Two-tier stream fetching:
```kotlin
suspend fun getStreamsForContent(type: String, id: String): Result<List<Stream>> {
    // Tier 1: Modular ProviderRegistry (Torrentio + NoTorrent)
    val providerResult = providerRegistry.fetchAllStreams(type, id)
    val providerStreams = providerResult.getOrElse { emptyList() }
    if (providerStreams.isNotEmpty()) {
        val enriched = debridEnricher.get().enrich(providerStreams)
        return Result.success(applyFilters(enriched, type))
    }
    
    // Tier 2: Legacy addon system (fallback)
    val streamAddons = getStreamAddons()
    val rawResult = client.getStreams(streamAddons, type, id)
    return rawResult.map { streams ->
        val enriched = debridEnricher.get().enrich(streams)
        applyFilters(enriched, type)
    }
}
```

## Performance Targets

- Streams visible within 3s on cold start (parallel fetch)
- Streams visible within 0.5s on warm cache hit
- Debrid resolution < 5s (single provider)
- Multi-provider fan-out < 8s (all enabled providers)

## Security Rules

- 0 force-unwraps — all nullable fields null-safe
- 0 secrets in logs — `SensitiveLogSanitizer` on all URLs
- Narrow catch blocks: `IOException`, `HttpException`, `SerializationException` only
- `CancellationException` always rethrown
- Debrid budget enforced before every API call
- Tokenized unrestricted URLs in memory only, never persisted