Designing Spotify: Music Streaming at Scale

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

┌─────────────────────────────────────────────────────────────────────┐
│                    DOWNLOAD MANAGEMENT ARCHITECTURE                  │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    USER ACTIONS                               │   │
│  │  • Toggle "Download" on playlist                              │   │
│  │  • "Download" button on album                                 │   │
│  │  • Settings: "Download over WiFi only"                        │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│                               ▼                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                   DOWNLOAD INTENT MANAGER                     │   │
│  │  • Tracks which playlists/albums are marked for download      │   │
│  │  • Resolves to individual track list                          │   │
│  │  • Watches for playlist changes (new tracks added)            │   │
│  │  • Manages download quotas (10,000 track limit)               │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│                               ▼                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    DOWNLOAD QUEUE                             │   │
│  │  • Priority queue of tracks to download                       │   │
│  │  • Priority: Recently played > Current playlist > Older       │   │
│  │  • Handles pause/resume, cancellation                         │   │
│  │  • Deduplication (track in multiple playlists)                │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│          ┌────────────────────┴────────────────────┐                │
│          ▼                                         ▼                │
│  ┌─────────────────────┐               ┌────────────────────────┐   │
│  │ NETWORK MONITOR     │               │ DOWNLOAD WORKERS       │   │
│  │                     │               │ (Parallel downloaders) │   │
│  │ • Connection type   │               │                        │   │
│  │ • WiFi/Cellular     │◄──────────────│ • Concurrent downloads │   │
│  │ • Bandwidth estimate│               │ • Retry with backoff   │   │
│  │ • Low-power mode    │               │ • Progress tracking    │   │
│  └─────────────────────┘               └────────────────────────┘   │
│                                                    │                 │
│                                                    ▼                 │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    DRM PROCESSOR                              │   │
│  │  • Encrypts downloaded content with device-specific key       │   │
│  │  • Stores license information                                 │   │
│  │  • Validates playback rights                                  │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                │                                     │
│                                ▼                                     │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                    OFFLINE STORAGE                            │   │
│  │  • Encrypted audio files                                      │   │
│  │  • Metadata cache                                             │   │
│  │  • License store                                              │   │
│  │  • Album art cache                                            │   │
│  └──────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

class DownloadIntentManager:
    """
    Manages what content user wants downloaded.
    
    Works at the "container" level (playlists, albums) not individual tracks.
    Automatically resolves to tracks and keeps in sync.
    """
    
    def __init__(self, offline_store, track_limit=10000):
        self.offline_store = offline_store
        self.track_limit = track_limit
        self.download_intents = {}  # container_id -> DownloadIntent
    
    async def mark_for_download(
        self,
        container_type: str,  # 'playlist' or 'album'
        container_id: str,
        quality: DownloadQuality
    ) -> DownloadResult:
        """
        Mark a playlist or album for offline download.
        """
        # Check quotas
        current_count = await self.get_downloaded_track_count()
        container_tracks = await self.get_container_tracks(container_type, container_id)
        new_tracks = [t for t in container_tracks if not self.is_downloaded(t)]
        
        if current_count + len(new_tracks) > self.track_limit:
            raise DownloadQuotaExceeded(
                f"Would exceed {self.track_limit} track limit. "
                f"Currently: {current_count}, Adding: {len(new_tracks)}"
            )
        
        # Create download intent
        intent = DownloadIntent(
            container_type=container_type,
            container_id=container_id,
            quality=quality,
            created_at=datetime.utcnow(),
            track_ids=container_tracks
        )
        
        # Store intent
        self.download_intents[container_id] = intent
        await self._persist_intents()
        
        # Queue tracks for download
        for track_id in new_tracks:
            await self.download_queue.enqueue(
                track_id,
                quality=quality,
                priority=self._calculate_priority(track_id, container_id)
            )
        
        return DownloadResult(
            queued_tracks=len(new_tracks),
            already_downloaded=len(container_tracks) - len(new_tracks)
        )
    
    async def handle_playlist_updated(self, playlist_id: str):
        """
        Called when a downloaded playlist changes (tracks added/removed).
        
        Auto-queues new tracks for download.
        """
        if playlist_id not in self.download_intents:
            return  # Not marked for download
        
        intent = self.download_intents[playlist_id]
        current_tracks = await self.get_container_tracks('playlist', playlist_id)
        
        # Find new tracks
        old_tracks = set(intent.track_ids)
        new_tracks = [t for t in current_tracks if t not in old_tracks]
        
        # Find removed tracks
        removed_tracks = [t for t in old_tracks if t not in current_tracks]
        
        # Queue new tracks
        for track_id in new_tracks:
            await self.download_queue.enqueue(
                track_id,
                quality=intent.quality,
                priority=DownloadPriority.NORMAL
            )
        
        # Optionally clean up removed tracks (if not in other playlists)
        for track_id in removed_tracks:
            if not await self._is_in_other_downloaded_content(track_id):
                await self.offline_store.remove_track(track_id)
        
        # Update intent
        intent.track_ids = current_tracks
        await self._persist_intents()
 
class DownloadQueue:
    """
    Priority queue for track downloads with concurrency control.
    """
    
    def __init__(self, max_concurrent=3):
        self.queue = PriorityQueue()
        self.in_progress = set()
        self.max_concurrent = max_concurrent
        self.paused = False
    
    async def enqueue(
        self,
        track_id: str,
        quality: DownloadQuality,
        priority: int
    ):
        """Add track to download queue."""
        if track_id in self.in_progress:
            return  # Already downloading
        
        if await self.is_downloaded(track_id, quality):
            return  # Already have it
        
        item = DownloadItem(
            track_id=track_id,
            quality=quality,
            priority=priority,
            queued_at=datetime.utcnow()
        )
        
        await self.queue.put((priority, item))
        await self._maybe_start_downloads()
    
    async def _download_worker(self, item: DownloadItem):
        """Worker that handles single download."""
        self.in_progress.add(item.track_id)
        
        try:
            # Get download URL from server
            download_info = await self.api.get_download_url(
                item.track_id,
                item.quality
            )
            
            # Download with progress tracking
            async for chunk in self.http_client.stream(download_info.url):
                if self.paused:
                    # Save partial download state
                    await self._save_partial(item.track_id, chunk)
                    return
                
                await self.offline_store.write_chunk(item.track_id, chunk)
                await self._update_progress(item.track_id, chunk.size)
            
            # Finalize download (verify, decrypt, store)
            await self.drm_processor.finalize_download(
                item.track_id,
                download_info.license
            )
            
            await self._emit_download_complete(item.track_id)
            
        except DownloadError as e:
            await self._handle_download_error(item, e)
        finally:
            self.in_progress.discard(item.track_id)
            await self._maybe_start_downloads()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

class DRMManager:
    """
    Manages digital rights for offline content.
    
    Uses a hierarchy of keys:
    - Master License Key (MLK): From server, protected by device binding
    - Content Encryption Key (CEK): Unique per track, encrypted by MLK
    - Device Key: Unique to device, protects key storage
    """
    
    def __init__(self, secure_storage):
        self.secure_storage = secure_storage  # Platform secure enclave
        self.license_cache = {}
    
    def initialize_device(self) -> DeviceBinding:
        """
        Initialize DRM on first app install.
        
        Generates device-specific keys and registers with server.
        """
        # Generate device key pair in secure enclave
        # This key never leaves the secure hardware
        device_key_id = self.secure_storage.generate_key_pair()
        
        # Get device fingerprint (hardware IDs, etc.)
        device_fingerprint = self._compute_device_fingerprint()
        
        # Register with license server
        binding = self.license_server.register_device(
            device_key_id=device_key_id,
            device_fingerprint=device_fingerprint,
            device_public_key=self.secure_storage.get_public_key(device_key_id)
        )
        
        # Store device binding
        self.secure_storage.store('device_binding', binding)
        
        return binding
    
    async def acquire_license(self, track_id: str) -> ContentLicense:
        """
        Acquire license to play/download a track.
        
        Returns encrypted CEK that can only be decrypted
        with this device's key.
        """
        device_binding = self.secure_storage.get('device_binding')
        
        # Request license from server
        license_response = await self.license_server.request_license(
            track_id=track_id,
            device_id=device_binding.device_id,
            license_type='offline'  # vs 'streaming'
        )
        
        # Verify license signature
        if not self._verify_license_signature(license_response):
            raise LicenseVerificationError("Invalid license signature")
        
        # Store license
        license = ContentLicense(
            track_id=track_id,
            encrypted_cek=license_response.encrypted_cek,
            expiry=license_response.expiry,
            policy=license_response.policy
        )
        
        await self._store_license(track_id, license)
        return license
    
    async def encrypt_content(
        self,
        track_id: str,
        raw_audio: bytes,
        license: ContentLicense
    ) -> bytes:
        """
        Encrypt audio content for offline storage.
        
        Uses AES-128-CTR for streaming encryption.
        """
        # Decrypt CEK using device key (happens in secure enclave)
        cek = self.secure_storage.decrypt(
            license.encrypted_cek,
            key_id=self.device_key_id
        )
        
        # Generate random IV
        iv = os.urandom(16)
        
        # Encrypt audio with CEK
        cipher = AES.new(cek, AES.MODE_CTR, nonce=iv[:8])
        encrypted_audio = cipher.encrypt(raw_audio)
        
        # Combine IV + encrypted data
        return iv + encrypted_audio
    
    def decrypt_for_playback(
        self,
        track_id: str,
        encrypted_audio: bytes,
        start_pos: int = 0
    ) -> Generator[bytes, None, None]:
        """
        Decrypt audio during playback.
        
        Yields decrypted chunks for real-time playback.
        This happens in a streaming manner to support seeking.
        """
        # Verify license is still valid
        license = self.license_cache.get(track_id)
        if not license:
            license = self._load_license(track_id)
        
        if license.expiry < datetime.utcnow():
            raise LicenseExpiredError(f"License for {track_id} expired")
        
        # Decrypt CEK in secure enclave
        cek = self.secure_storage.decrypt(
            license.encrypted_cek,
            key_id=self.device_key_id
        )
        
        # Extract IV from start of file
        iv = encrypted_audio[:16]
        
        # Initialize cipher at correct position for seeking
        cipher = AES.new(cek, AES.MODE_CTR, nonce=iv[:8])
        if start_pos > 0:
            # CTR mode allows random access by computing keystream
            cipher._counter = self._compute_counter(iv, start_pos)
        
        # Stream decrypted chunks
        chunk_size = 16384  # 16KB chunks
        offset = 16 + start_pos  # Skip IV, start at position
        
        while offset < len(encrypted_audio):
            chunk = encrypted_audio[offset:offset + chunk_size]
            yield cipher.decrypt(chunk)
            offset += chunk_size
    
    async def check_offline_validity(self) -> OfflineStatus:
        """
        Check if offline content is still valid.
        
        Called on app launch to verify licenses.
        """
        device_binding = self.secure_storage.get('device_binding')
        
        # Check time since last online check
        last_check = device_binding.last_license_check
        days_offline = (datetime.utcnow() - last_check).days
        
        if days_offline > 30:
            return OfflineStatus(
                valid=False,
                reason='offline_too_long',
                message='Please go online to refresh your licenses'
            )
        
        # Check individual license expiries
        expired_tracks = []
        for track_id, license in self._get_all_licenses():
            if license.expiry < datetime.utcnow():
                expired_tracks.append(track_id)
        
        if expired_tracks:
            # Remove expired content
            for track_id in expired_tracks:
                await self.offline_store.remove_track(track_id)
        
        return OfflineStatus(
            valid=True,
            days_remaining=30 - days_offline,
            expired_tracks=len(expired_tracks)
        )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

class OfflineStorageManager:
    """
    Manages storage for offline content.
    """
    
    def __init__(self, storage_path: str):
        self.storage_path = storage_path
        self.database = OfflineDatabase(storage_path)
    
    async def check_storage(self) -> StorageStatus:
        """
        Check available storage and offline usage.
        """
        # Get device storage info
        total, used, free = shutil.disk_usage(self.storage_path)
        
        # Get offline content usage
        offline_usage = await self._calculate_offline_usage()
        
        return StorageStatus(
            device_total_gb=total / (1024**3),
            device_free_gb=free / (1024**3),
            offline_usage_gb=offline_usage / (1024**3),
            track_count=await self.database.count_tracks(),
            artwork_usage_gb=await self._get_artwork_size() / (1024**3)
        )
    
    async def estimate_download_size(
        self,
        track_ids: List[str],
        quality: DownloadQuality
    ) -> DownloadSizeEstimate:
        """
        Estimate storage needed for download.
        
        Used to warn user before large downloads.
        """
        # Get total duration of tracks
        total_duration_ms = 0
        for track_id in track_ids:
            metadata = await self.metadata_service.get(track_id)
            total_duration_ms += metadata.duration_ms
        
        # Calculate size based on quality
        bytes_per_second = self._get_bytes_per_second(quality)
        estimated_size = (total_duration_ms / 1000) * bytes_per_second
        
        # Add overhead for metadata, artwork, etc.
        estimated_size *= 1.05
        
        # Check if enough space
        status = await self.check_storage()
        has_space = estimated_size < (status.device_free_gb * 1024**3)
        
        return DownloadSizeEstimate(
            size_bytes=int(estimated_size),
            size_readable=self._format_size(estimated_size),
            duration_minutes=total_duration_ms / 60000,
            has_sufficient_space=has_space
        )
    
    async def smart_cleanup(
        self,
        target_free_mb: int = 500
    ) -> CleanupResult:
        """
        Free up storage by removing least valuable offline content.
        
        Priority for removal (lowest first):
        1. Tracks not played in 90+ days
        2. Tracks not in any downloaded playlist
        3. Tracks from podcasts (usually listen-once)
        4. Tracks played longest ago
        """
        status = await self.check_storage()
        free_mb = status.device_free_gb * 1024
        
        if free_mb >= target_free_mb:
            return CleanupResult(freed_mb=0, removed_tracks=0)
        
        needed_mb = target_free_mb - free_mb
        freed_mb = 0
        removed_tracks = []
        
        # Get candidates sorted by cleanup priority
        candidates = await self._get_cleanup_candidates()
        
        for track_id, size_mb, priority in candidates:
            if freed_mb >= needed_mb:
                break
            
            # Check if track is in a downloaded playlist
            if await self._is_in_downloaded_playlist(track_id):
                continue  # Don't remove, user explicitly wants it
            
            await self.remove_track(track_id)
            freed_mb += size_mb
            removed_tracks.append(track_id)
        
        return CleanupResult(
            freed_mb=freed_mb,
            removed_tracks=len(removed_tracks),
            tracks=removed_tracks
        )
    
    async def move_to_external_storage(self):
        """
        Move offline content to SD card (Android only).
        """
        if not self._has_external_storage():
            raise NoExternalStorageError()
        
        external_path = self._get_external_storage_path()
        
        # Move encrypted files
        for track in await self.database.get_all_tracks():
            source = f"{self.storage_path}/audio/{track.id}.enc"
            dest = f"{external_path}/audio/{track.id}.enc"
            shutil.move(source, dest)
        
        # Update database with new paths
        await self.database.update_storage_location(external_path)
        
        self.storage_path = external_path
    
    def _get_cleanup_candidates(self) -> List[Tuple[str, int, int]]:
        """
        Get tracks ranked by cleanup priority.
        
        Returns: [(track_id, size_mb, priority_score), ...]
        Lower priority score = remove first.
        """
        tracks = self.database.query('''
            SELECT 
                track_id,
                size_bytes / 1024 / 1024 as size_mb,
                CASE 
                    WHEN last_played_at IS NULL THEN 0
                    WHEN julianday('now') - julianday(last_played_at) > 90 THEN 1
                    WHEN julianday('now') - julianday(last_played_at) > 30 THEN 2
                    ELSE 3
                END + 
                CASE WHEN is_podcast THEN 0 ELSE 2 END +
                CASE WHEN play_count > 5 THEN 2 ELSE 0 END
                as priority_score
            FROM offline_tracks
            ORDER BY priority_score ASC, last_played_at ASC
        ''')
        
        return [(r['track_id'], r['size_mb'], r['priority_score']) for r in tracks]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170

class UnifiedPlaybackManager:
    """
    Unified playback manager that seamlessly handles
    online streaming and offline playback.
    """
    
    def __init__(
        self,
        streaming_player: StreamingPlayer,
        offline_store: OfflineStore,
        drm_manager: DRMManager
    ):
        self.streaming_player = streaming_player
        self.offline_store = offline_store
        self.drm_manager = drm_manager
        self.network_monitor = NetworkMonitor()
    
    async def play_track(
        self,
        track_id: str,
        position_ms: int = 0
    ):
        """
        Play track from best available source.
        
        Priority:
        1. Offline (if available and valid) - fastest
        2. Streaming (if online)
        3. Cached streaming segments (partial offline)
        """
        # Check offline availability
        offline_status = await self.offline_store.get_track_status(track_id)
        
        if offline_status.available:
            # Verify license before playing
            try:
                await self.drm_manager.verify_playback_rights(track_id)
                return await self._play_offline(track_id, position_ms)
            except LicenseExpiredError:
                # Fall through to streaming if license expired
                pass
        
        # Check network for streaming
        if self.network_monitor.is_online():
            return await self._play_streaming(track_id, position_ms)
        
        # Completely offline with no local content
        raise PlaybackError(
            f"Track {track_id} not available offline and no network connection"
        )
    
    async def _play_offline(self, track_id: str, position_ms: int):
        """
        Play from offline storage.
        """
        # Load encrypted audio from storage
        encrypted_audio = await self.offline_store.read_track(track_id)
        
        # Create decrypting audio source
        audio_source = DecryptingAudioSource(
            encrypted_audio=encrypted_audio,
            drm_manager=self.drm_manager,
            track_id=track_id,
            start_position=self._ms_to_bytes(position_ms)
        )
        
        # Play through audio output
        await self.audio_output.play(audio_source)
        
        # Update local play stats
        await self.local_stats.record_play(track_id, offline=True)
    
    async def _play_streaming(self, track_id: str, position_ms: int):
        """
        Stream from CDN with optional opportunistic caching.
        """
        # Check if we should opportunistically download
        should_cache = await self._should_opportunistic_cache(track_id)
        
        if should_cache:
            # Stream and cache simultaneously
            audio_source = CachingStreamSource(
                track_id=track_id,
                streaming_player=self.streaming_player,
                cache_writer=self.offline_store
            )
        else:
            audio_source = self.streaming_player.get_source(track_id)
        
        await self.audio_output.play(audio_source, position_ms)
    
    async def _should_opportunistic_cache(self, track_id: str) -> bool:
        """
        Decide if we should cache this track while streaming.
        
        Cache if:
        - User has storage space
        - Track is in a playlist marked for download
        - On WiFi (don't use cellular for caching)
        - User has Premium
        """
        if not self.user_service.is_premium():
            return False
        
        if not self.network_monitor.is_wifi():
            return False
        
        storage = await self.offline_store.check_storage()
        if storage.device_free_gb < 1:
            return False
        
        # Check if track should be downloaded
        return await self.download_intent_manager.wants_track_offline(track_id)
 
class DecryptingAudioSource:
    """
    Audio source that decrypts on-the-fly during playback.
    Supports seeking by calculating correct keystream position.
    """
    
    def __init__(
        self,
        encrypted_audio: bytes,
        drm_manager: DRMManager,
        track_id: str,
        start_position: int = 0
    ):
        self.encrypted_audio = encrypted_audio
        self.drm_manager = drm_manager
        self.track_id = track_id
        self.position = start_position
        
        # Initialize decryption cipher
        self.decrypt_generator = self.drm_manager.decrypt_for_playback(
            track_id,
            encrypted_audio,
            start_pos=start_position
        )
    
    def read(self, num_bytes: int) -> bytes:
        """Read and decrypt next bytes."""
        if not hasattr(self, '_buffer'):
            self._buffer = b''
        
        while len(self._buffer) < num_bytes:
            try:
                chunk = next(self.decrypt_generator)
                self._buffer += chunk
            except StopIteration:
                break  # End of file
        
        result = self._buffer[:num_bytes]
        self._buffer = self._buffer[num_bytes:]
        self.position += len(result)
        
        return result
    
    def seek(self, position: int):
        """
        Seek to position in audio.
        
        Reinitializes decryption at new position.
        """
        self.position = position
        self.decrypt_generator = self.drm_manager.decrypt_for_playback(
            self.track_id,
            self.encrypted_audio,
            start_pos=position
        )
        self._buffer = b''

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

class LicenseSyncManager:
    """
    Manages license renewal and synchronization.
    """
    
    def __init__(self, drm_manager, license_server):
        self.drm_manager = drm_manager
        self.license_server = license_server
        self.renewal_in_progress = set()
    
    async def on_app_foreground(self):
        """
        Called when app comes to foreground.
        Opportunistically sync licenses if online.
        """
        if not self.network_monitor.is_online():
            return
        
        # Check for licenses needing renewal
        expiring = await self._get_expiring_licenses(days=7)
        
        if expiring:
            await self.batch_renew_licenses([l.track_id for l in expiring])
        
        # Update last online timestamp
        await self._update_online_timestamp()
    
    async def batch_renew_licenses(
        self,
        track_ids: List[str],
        batch_size: int = 50
    ):
        """
        Renew licenses for multiple tracks in batches.
        """
        for batch in self._chunks(track_ids, batch_size):
            try:
                # Request batch renewal from server
                renewals = await self.license_server.renew_batch(
                    track_ids=batch,
                    device_id=self.drm_manager.device_id
                )
                
                # Update local licenses
                for track_id, new_license in renewals.items():
                    if new_license.status == 'renewed':
                        await self.drm_manager.update_license(
                            track_id,
                            new_license
                        )
                    elif new_license.status == 'revoked':
                        # Content no longer available
                        await self.offline_store.remove_track(track_id)
                        await self._notify_user_removal(track_id, new_license.reason)
                
            except NetworkError:
                # Will retry on next online opportunity
                logging.warning(f"License renewal failed, will retry: {batch}")
    
    async def check_offline_validity(self) -> OfflineValidityResult:
        """
        Comprehensive check of offline content validity.
        Called on app startup.
        """
        results = OfflineValidityResult()
        
        # Check subscription status (cached)
        subscription = await self._get_cached_subscription()
        if subscription.status != 'active':
            results.subscription_valid = False
            results.offline_available = False
            return results
        
        # Check time since last online
        last_online = await self._get_last_online_timestamp()
        offline_days = (datetime.utcnow() - last_online).days
        
        if offline_days > 30:
            results.offline_available = False
            results.reason = 'offline_too_long'
            results.days_offline = offline_days
            return results
        
        results.offline_available = True
        results.days_remaining = 30 - offline_days
        
        # Check individual license states
        licenses = await self.drm_manager.get_all_licenses()
        for license in licenses:
            if license.expiry < datetime.utcnow():
                results.expired_tracks.append(license.track_id)
            elif license.expiry < datetime.utcnow() + timedelta(days=3):
                results.expiring_soon.append(license.track_id)
        
        return results
    
    async def handle_subscription_change(self, new_status: str):
        """
        Handle subscription status changes.
        
        Called when user subscription changes (cancel, downgrade, etc.)
        """
        if new_status == 'cancelled' or new_status == 'free':
            # User lost Premium - revoke all offline content
            await self._revoke_all_offline_content()
            await self._notify_user_offline_revoked()
        
        elif new_status == 'lapsed':
            # Subscription payment failed - grace period
            # Content still playable but warn user
            await self._notify_payment_issue()
 
class OfflineSyncScheduler:
    """
    Schedules background sync tasks for offline content.
    """
    
    def schedule_license_check(self):
        """
        Schedule periodic license validity checks.
        
        Uses background task APIs (WorkManager on Android,
        BackgroundTasks on iOS) for reliable execution.
        """
        # Check at least daily when charging and on WiFi
        self.background_scheduler.schedule(
            task_id='license_check',
            interval_hours=24,
            constraints=BackgroundConstraints(
                requires_wifi=True,
                requires_charging=True,
                run_on_idle=True
            ),
            callback=self.license_sync_manager.on_background_check
        )
    
    def schedule_content_update(self):
        """
        Schedule sync of downloaded playlist updates.
        
        If user downloaded a playlist, and new tracks were added,
        download those tracks too.
        """
        self.background_scheduler.schedule(
            task_id='playlist_sync',
            interval_hours=6,
            constraints=BackgroundConstraints(
                requires_wifi=True,
                requires_storage=True
            ),
            callback=self.download_intent_manager.sync_downloaded_playlists
        )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

class SmartDownloadManager:
    """
    Intelligently manages automatic downloads based on
    user behavior and predictions.
    """
    
    def __init__(self):
        self.auto_download_enabled = True  # User setting
        self.max_auto_download_mb = 1000  # 1GB limit for auto-downloads
    
    async def on_wifi_connected(self):
        """
        Called when device connects to WiFi.
        Triggers smart download evaluation.
        """
        if not self.auto_download_enabled:
            return
        
        if not await self._has_storage_space():
            return
        
        # Check what to auto-download
        candidates = await self._get_auto_download_candidates()
        
        # Prioritize and queue
        prioritized = self._prioritize_candidates(candidates)
        
        for track_id in prioritized[:50]:  # Limit batch size
            await self.download_queue.enqueue(
                track_id,
                quality=self._get_auto_download_quality(),
                priority=DownloadPriority.LOW  # Lower than explicit downloads
            )
    
    async def _get_auto_download_candidates(self) -> List[str]:
        """
        Identify tracks that should be auto-downloaded.
        """
        candidates = []
        
        # 1. Personalized playlists (DW, Daily Mix, etc.)
        personalized = await self._get_personalized_playlists()
        for playlist in personalized:
            if playlist.auto_download_eligible:
                tracks = await self.playlist_service.get_tracks(playlist.id)
                for track in tracks:
                    if not await self.offline_store.is_downloaded(track.id):
                        candidates.append((track.id, 'personalized', playlist.priority))
        
        # 2. Frequently played but not downloaded
        frequent = await self._get_frequently_played(days=14)
        for track_id, play_count in frequent:
            if not await self.offline_store.is_downloaded(track_id):
                candidates.append((track_id, 'frequent', play_count))
        
        # 3. Predicted next listens based on patterns
        predicted = await self._predict_next_listens()
        for track_id, probability in predicted:
            if probability > 0.7 and not await self.offline_store.is_downloaded(track_id):
                candidates.append((track_id, 'predicted', probability))
        
        return candidates
    
    async def _predict_next_listens(self) -> List[Tuple[str, float]]:
        """
        Predict what user is likely to listen to next.
        
        Uses:
        - Time of day patterns
        - Day of week patterns
        - Recent listening context
        - Current queue
        """
        predictions = []
        
        now = datetime.now()
        hour = now.hour
        weekday = now.weekday()
        
        # Get tracks typically played at this time
        time_patterns = await self.listening_patterns.get_for_time(
            hour=hour,
            weekday=weekday
        )
        
        for track_id, frequency in time_patterns:
            probability = min(1.0, frequency / 10)  # Normalize
            predictions.append((track_id, probability))
        
        # Get tracks from similar sessions
        current_context = await self._get_current_listening_context()
        similar_sessions = await self.session_analyzer.find_similar(current_context)
        
        for session in similar_sessions[:5]:
            for track_id in session.upcoming_tracks:
                probability = session.similarity * 0.8
                predictions.append((track_id, probability))
        
        # Deduplicate and aggregate probabilities
        aggregated = self._aggregate_predictions(predictions)
        
        return sorted(aggregated, key=lambda x: x[1], reverse=True)
    
    async def refresh_personalized_playlists(self):
        """
        Refresh auto-downloaded personalized playlists.
        
        Called weekly (Monday) for Discover Weekly,
        daily for Daily Mixes.
        """
        for playlist in await self._get_auto_download_playlists():
            current_tracks = await self.playlist_service.get_tracks(playlist.id)
            downloaded_tracks = await self.offline_store.get_downloaded_for_playlist(playlist.id)
            
            # Find new tracks
            new_tracks = [t for t in current_tracks if t.id not in downloaded_tracks]
            
            # Find removed tracks
            removed_tracks = [t for t in downloaded_tracks if t not in current_tracks]
            
            # Download new
            for track in new_tracks:
                await self.download_queue.enqueue(
                    track.id,
                    quality=playlist.quality,
                    priority=DownloadPriority.NORMAL
                )
            
            # Clean up removed (if not in other content)
            for track_id in removed_tracks:
                if not await self._is_in_other_offline_content(track_id):
                    await self.offline_store.remove_track(track_id)