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

# Example: Rights structure for a single track
track:
  title: "Example Hit Song"
  isrc: "USRC12345678"  # International Standard Recording Code
  
  # Composition (Publishing) Rights
  composition:
    title: "Example Hit Song"
    iswc: "T-123.456.789-0"  # International Standard Musical Work Code
    writers:
      - name: "John Songwriter"
        share: 50%
        publisher: "Sony/ATV Music Publishing"
        territories: ["worldwide"]
      - name: "Jane Composer"
        share: 50%
        publisher: "Warner Chappell"
        territories: ["worldwide except EU"]
        # Different publisher for EU...
        eu_publisher: "Universal Music Publishing"
    
  # Sound Recording (Master) Rights
  master:
    label: "Atlantic Records"
    distributor: "Warner Music Group"
    territories:
      - region: "North America"
        available: true
        exclusive: true
      - region: "Europe"
        available: true
        exclusive: false
      - region: "China"
        available: false
        reason: "No distribution agreement"
    
  # Artist Rights
  artists:
    - name: "Famous Artist"
      role: "primary"
      royalty_share: 15%
    - name: "Guest Feature"
      role: "featured"
      royalty_share: 3%

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

-- Track availability by territory
CREATE TABLE track_availability (
    track_id            VARCHAR(64) NOT NULL,
    territory_code      CHAR(2) NOT NULL,        -- ISO 3166-1 alpha-2
    availability        ENUM('available', 'blocked', 'premium_only', 'restricted'),
    
    -- Restriction details
    restriction_reason  VARCHAR(256),            -- Why restricted (for CS reference)
    restriction_type    ENUM('licensing', 'legal', 'label_request', 'rights_expired'),
    
    -- Time-based restrictions
    available_from      TIMESTAMP,               -- If delayed release
    available_until     TIMESTAMP,               -- If rights expire
    
    -- Platform restrictions (some licenses are mobile-only, etc.)
    platforms           JSON,                     -- ['mobile', 'desktop', 'web']
    
    -- Metadata
    updated_at          TIMESTAMP,
    updated_by          VARCHAR(128),            -- Source system
    
    PRIMARY KEY (track_id, territory_code)
);
 
-- Album-level restrictions (more efficient for full-album blocks)
CREATE TABLE album_availability (
    album_id            VARCHAR(64) NOT NULL,
    territory_code      CHAR(2) NOT NULL,
    availability        ENUM('available', 'blocked', 'partial'),
    blocked_track_count INT DEFAULT 0,
    
    PRIMARY KEY (album_id, territory_code)
);
 
-- Artist-level blocks (sanctions, legal issues)
CREATE TABLE artist_restrictions (
    artist_id           VARCHAR(64) NOT NULL,
    territory_code      CHAR(2) NOT NULL,
    restriction_type    ENUM('blocked', 'content_warning', 'age_restricted'),
    reason              VARCHAR(256),
    
    PRIMARY KEY (artist_id, territory_code)
);
 
-- Index for efficient lookups during playback
CREATE INDEX idx_track_territory ON track_availability(territory_code, track_id);

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

┌─────────────────────────────────────────────────────────────────────┐
│                    GEO-RESTRICTION ENFORCEMENT LAYERS                │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  LAYER 1: CLIENT                                                     │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  • Filter unavailable content from UI                         │   │
│  │  • Cache availability per user's country                      │   │
│  │  • Show "Not available in your country" message               │   │
│  │  • Acts as first defense, but NOT trusted                     │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│                               ▼                                      │
│  LAYER 2: API GATEWAY                                                │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  • Determine user's country from:                             │   │
│  │    - Account registration country (primary)                   │   │
│  │    - IP geolocation (secondary, for roaming)                  │   │
│  │    - SIM country (mobile)                                     │   │
│  │  • Check availability before serving metadata                 │   │
│  │  • Log attempts to access restricted content                  │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│                               ▼                                      │
│  LAYER 3: PLAYBACK SERVICE                                           │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  • Final check before generating stream URL                   │   │
│  │  • Validate license for track + territory                     │   │
│  │  • Generate signed URL with territory embedded                │   │
│  │  • Log approved playback for royalty calculation              │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                               │                                      │
│                               ▼                                      │
│  LAYER 4: CDN                                                        │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  • Validate signed URL hasn't been tampered                   │   │
│  │  • Can optionally verify IP matches token territory           │   │
│  │  • Rate limiting per account                                  │   │
│  │  • Last line of defense                                       │   │
│  └──────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘

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

class GeoRestrictionService:
    """
    Central service for geo-restriction checks.
    
    Optimized for low-latency lookups during playback.
    """
    
    def __init__(self):
        # In-memory cache of restrictions
        self.restriction_cache = LRUCache(maxsize=1_000_000)
        # Bloom filter for quick "definitely available" checks
        self.available_bloom = CountingBloomFilter()
        # Redis for distributed cache
        self.redis = RedisCluster()
    
    async def check_availability(
        self,
        track_id: str,
        territory: str,
        platform: str = 'all',
        user_tier: str = 'premium'
    ) -> AvailabilityResult:
        """
        Check if track is available in territory.
        
        Optimized path:
        1. Bloom filter for fast positive (available) cases
        2. Redis cache for recent lookups
        3. Database for cache misses
        """
        cache_key = f"{track_id}:{territory}"
        
        # Fast path: Bloom filter says definitely available
        if self.available_bloom.contains(cache_key):
            # Still verify platform and tier restrictions
            return await self._check_restrictions(track_id, territory, platform, user_tier)
        
        # Check local cache
        cached = self.restriction_cache.get(cache_key)
        if cached is not None:
            return self._apply_filters(cached, platform, user_tier)
        
        # Check distributed cache
        cached = await self.redis.get(cache_key)
        if cached:
            self.restriction_cache.set(cache_key, cached)
            return self._apply_filters(cached, platform, user_tier)
        
        # Database lookup
        availability = await self.db.fetch_one(
            """SELECT * FROM track_availability 
               WHERE track_id = %s AND territory_code = %s""",
            (track_id, territory)
        )
        
        if not availability:
            # No record means available (default available)
            availability = AvailabilityRecord(
                track_id=track_id,
                territory=territory,
                availability='available'
            )
        
        # Cache result
        await self._cache_availability(cache_key, availability)
        
        return self._apply_filters(availability, platform, user_tier)
    
    async def get_unavailable_territories(
        self,
        track_id: str
    ) -> List[str]:
        """
        Get list of territories where track is NOT available.
        
        Used for displaying "Not available in X" messages.
        """
        blocked = await self.db.fetch_all(
            """SELECT territory_code FROM track_availability
               WHERE track_id = %s AND availability = 'blocked'""",
            (track_id,)
        )
        return [r['territory_code'] for r in blocked]
    
    async def filter_playlist_by_territory(
        self,
        track_ids: List[str],
        territory: str
    ) -> Tuple[List[str], List[str]]:
        """
        Filter a list of tracks by territory availability.
        
        Returns (available_tracks, unavailable_tracks).
        Used when displaying playlists that might have blocked content.
        """
        available = []
        unavailable = []
        
        # Batch lookup for efficiency
        availabilities = await self._batch_check(track_ids, territory)
        
        for track_id, is_available in availabilities.items():
            if is_available:
                available.append(track_id)
            else:
                unavailable.append(track_id)
        
        return available, unavailable
    
    def determine_user_territory(
        self,
        user: User,
        request_context: RequestContext
    ) -> str:
        """
        Determine user's effective territory for licensing.
        
        Logic:
        1. Registration country is baseline
        2. If roaming, use temporary location with limits
        3. Mobile: registration country always (prevents VPN)
        4. Desktop/Web: IP geolocation checked against registration
        """
        registration_country = user.registration_country
        ip_country = request_context.ip_country
        
        if request_context.platform == 'mobile':
            # Mobile users always use registration country
            # (SIM lock + app store region provide verification)
            return registration_country
        
        # Check if user is roaming
        if ip_country != registration_country:
            # Allow roaming for 14 days (EU regulation)
            roaming_status = self._check_roaming_status(user, ip_country)
            
            if roaming_status.days_in_foreign_territory < 14:
                # Use temporary location for better experience
                # But log as roaming for royalty calculations
                return ip_country
            else:
                # Extended roaming: must use registration country
                # or update registration
                return registration_country
        
        return registration_country

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

class RoyaltyTrackingPipeline:
    """
    Processes streaming events into royalty calculations.
    
    Requirements:
    - Process 10B+ streams/day
    - No data loss (financial accuracy required)
    - Territory-accurate attribution
    - Handle complex ownership splits
    """
    
    def __init__(self):
        self.event_stream = KafkaCluster('streaming-events')
        self.royalty_store = BigQueryClient()
    
    def process_stream(self, event: StreamingEvent) -> RoyaltyRecord:
        """
        Convert streaming event to royalty record.
        
        A play is "royalty-eligible" if:
        - Duration >= 30 seconds (industry standard)
        - User has valid subscription or saw ad
        - Not flagged as fraudulent
        """
        if not self._is_royalty_eligible(event):
            return None
        
        # Get rights holder information
        rights = self.rights_database.get_rights(event.track_id)
        
        # Determine royalty rate for territory
        rates = self.rate_service.get_rates(
            territory=event.territory,
            subscription_tier=event.user_tier,
            platform=event.platform
        )
        
        # Calculate royalties for each stakeholder
        royalties = []
        
        # Master recording royalties (to label)
        master_royalty = rates.master_rate * rates.stream_value
        for owner in rights.master_owners:
            royalties.append(RoyaltyRecord(
                event_id=event.id,
                track_id=event.track_id,
                recipient_type='master',
                recipient_id=owner.id,
                amount=master_royalty * owner.share,
                territory=event.territory,
                timestamp=event.timestamp
            ))
        
        # Composition royalties (to publisher)
        composition_royalty = rates.composition_rate * rates.stream_value
        for owner in rights.composition_owners:
            # Get territory-specific publisher
            publisher = self._get_territory_publisher(owner, event.territory)
            royalties.append(RoyaltyRecord(
                event_id=event.id,
                track_id=event.track_id,
                recipient_type='composition',
                recipient_id=publisher.id,
                amount=composition_royalty * owner.share,
                territory=event.territory,
                timestamp=event.timestamp
            ))
        
        return royalties
    
    def _is_royalty_eligible(self, event: StreamingEvent) -> bool:
        """
        Determine if stream counts for royalties.
        """
        # Minimum 30 seconds played
        if event.duration_ms < 30000:
            return False
        
        # Free tier requires ad was served
        if event.user_tier == 'free' and not event.ad_served:
            return False
        
        # Check fraud signals
        if self._is_fraudulent(event):
            return False
        
        # Offline play during valid license period
        if event.offline and not self._valid_offline_license(event):
            return False
        
        return True
    
    async def aggregate_monthly_royalties(
        self,
        year_month: str  # "2024-01"
    ) -> MonthlyRoyaltyReport:
        """
        Aggregate royalties for monthly payment to rights holders.
        
        This is a massive aggregation job run monthly.
        """
        # Run BigQuery aggregation
        query = '''
        SELECT 
            recipient_id,
            recipient_type,
            territory,
            SUM(amount) as total_amount,
            COUNT(*) as stream_count
        FROM royalty_records
        WHERE FORMAT_TIMESTAMP('%Y-%m', timestamp) = @year_month
        GROUP BY recipient_id, recipient_type, territory
        '''
        
        results = await self.royalty_store.query(query, {'year_month': year_month})
        
        # Generate payment files for each distributor
        payment_files = self._generate_payment_files(results)
        
        return MonthlyRoyaltyReport(
            period=year_month,
            total_streams=sum(r['stream_count'] for r in results),
            total_royalties=sum(r['total_amount'] for r in results),
            payment_files=payment_files
        )
 
class FraudDetectionService:
    """
    Detect artificial streaming (click farms, bots).
    
    Fraudulent streams waste royalty pool and violate ToS.
    """
    
    def analyze_streaming_patterns(self, user_id: str) -> FraudScore:
        """
        Analyze user's streaming patterns for fraud signals.
        """
        history = self.streaming_history.get_recent(user_id, days=7)
        
        signals = []
        
        # Signal: Unusual listening hours (24/7 streaming)
        hour_distribution = self._get_hour_distribution(history)
        if self._is_uniform_24h(hour_distribution):
            signals.append(('uniform_hours', 0.3))
        
        # Signal: Same tracks on loop
        track_counts = Counter(e.track_id for e in history)
        most_common_ratio = track_counts.most_common(1)[0][1] / len(history)
        if most_common_ratio > 0.5:
            signals.append(('track_concentration', 0.4))
        
        # Signal: Streams always exactly 30 seconds
        durations = [e.duration_ms for e in history]
        if np.std(durations) < 1000:  # Very consistent durations
            signals.append(('uniform_duration', 0.5))
        
        # Signal: No user interactions (never skip, save, add to playlist)
        interaction_rate = len([e for e in history if e.had_interaction]) / len(history)
        if interaction_rate < 0.01:
            signals.append(('no_interactions', 0.3))
        
        # Signal: IP associated with known farms
        if self.ip_reputation.is_suspicious(history[0].ip):
            signals.append(('suspicious_ip', 0.6))
        
        # Combine signals
        fraud_score = 1 - np.prod([1 - s[1] for s in signals])
        
        return FraudScore(
            user_id=user_id,
            score=fraud_score,
            signals=signals,
            should_flag=fraud_score > 0.7
        )

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 ContentTakedownService:
    """
    Manages content takedowns across all systems.
    
    Requirements:
    - Takedown must propagate to all systems within SLA
    - Must handle cascading effects (playlists, offline, etc.)
    - Must maintain audit trail
    - Must support counter-notice reinstatement
    """
    
    async def process_takedown(
        self,
        takedown_request: TakedownRequest
    ) -> TakedownResult:
        """
        Process a content takedown across all systems.
        """
        # Validate request
        if not self._validate_request(takedown_request):
            raise InvalidTakedownRequest()
        
        # Log for audit
        audit_record = await self._create_audit_record(takedown_request)
        
        # Determine affected content
        affected_tracks = await self._resolve_affected_tracks(takedown_request)
        affected_territories = takedown_request.territories or ['ALL']
        
        results = TakedownResult(
            request_id=takedown_request.id,
            tracks_affected=len(affected_tracks)
        )
        
        # 1. Update availability database
        await self._update_availability_db(
            affected_tracks,
            affected_territories,
            availability='blocked',
            reason=takedown_request.type
        )
        
        # 2. Invalidate CDN caches
        await self._purge_cdn_cache(affected_tracks, affected_territories)
        
        # 3. Stop active streams
        active_streams = await self._find_active_streams(affected_tracks)
        for stream in active_streams:
            if stream.territory in affected_territories or 'ALL' in affected_territories:
                await self._terminate_stream(stream)
                results.streams_terminated += 1
        
        # 4. Remove from search index
        await self._update_search_index(affected_tracks, action='remove')
        
        # 5. Update playlists (mark tracks as unavailable)
        await self._update_affected_playlists(affected_tracks)
        
        # 6. Handle offline downloads
        affected_users = await self._find_users_with_offline(affected_tracks)
        for user in affected_users:
            await self._notify_offline_removal(user, affected_tracks)
        
        # 7. Notify relevant internal teams
        await self._notify_stakeholders(takedown_request, results)
        
        # Record completion
        audit_record.completed_at = datetime.utcnow()
        audit_record.results = results
        await self._save_audit_record(audit_record)
        
        return results
    
    async def _purge_cdn_cache(
        self,
        track_ids: List[str],
        territories: List[str]
    ):
        """
        Purge content from all CDN edge caches.
        
        This ensures no cached content continues to be served.
        """
        for track_id in track_ids:
            # Purge by path pattern
            purge_patterns = []
            for territory in territories:
                if territory == 'ALL':
                    purge_patterns.append(f"/audio/{track_id}/*")
                else:
                    purge_patterns.append(f"/{territory}/audio/{track_id}/*")
            
            # Send to CDN purge API
            for cdn in self.cdn_providers:
                await cdn.purge(patterns=purge_patterns)
    
    async def process_reinstatement(
        self,
        reinstatement_request: ReinstatementRequest
    ) -> ReinstatementResult:
        """
        Reinstate previously taken-down content.
        
        May happen due to counter-notice, license renewal, etc.
        """
        # Verify reinstatement is authorized
        original_takedown = await self._get_original_takedown(
            reinstatement_request.original_takedown_id
        )
        
        if not self._can_reinstate(original_takedown, reinstatement_request):
            raise ReinstatementNotAllowed()
        
        affected_tracks = original_takedown.affected_tracks
        territories = reinstatement_request.territories or original_takedown.territories
        
        # 1. Update availability back to available
        await self._update_availability_db(
            affected_tracks,
            territories,
            availability='available',
            reason='reinstated'
        )
        
        # 2. Warm CDN caches for popular tracks
        popular_tracks = await self._filter_popular(affected_tracks)
        await self._warm_cdn_cache(popular_tracks, territories)
        
        # 3. Update search index
        await self._update_search_index(affected_tracks, action='add')
        
        # 4. Update playlists
        await self._update_affected_playlists(affected_tracks, action='restore')
        
        return ReinstatementResult(
            tracks_reinstated=len(affected_tracks),
            territories_affected=territories
        )
    
    async def _update_affected_playlists(
        self,
        track_ids: List[str],
        action: str = 'mark_unavailable'
    ):
        """
        Update playlists containing affected tracks.
        
        Don't remove tracks (user might want them when reinstated),
        but mark as unavailable in that territory.
        """
        # Find playlists containing these tracks
        affected_playlists = await self.db.fetch_all(
            """SELECT DISTINCT playlist_id
               FROM playlist_tracks
               WHERE track_uri = ANY(%s)""",
            (track_ids,)
        )
        
        for playlist_id in affected_playlists:
            # Increment playlist version to trigger client refresh
            await self.playlist_service.increment_version(playlist_id)
            
            # Clients will re-fetch and see updated availability

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

class ComplianceMonitoringService:
    """
    Continuous monitoring of licensing and royalty compliance.
    """
    
    async def run_daily_compliance_checks(self):
        """
        Daily automated compliance checks.
        """
        results = ComplianceReport(date=date.today())
        
        # Check 1: Territorial violations
        violations = await self._check_territorial_violations()
        results.territorial_violations = violations
        if violations:
            await self._alert_compliance_team(
                'territorial_violations',
                violations
            )
        
        # Check 2: Royalty calculation discrepancies
        discrepancies = await self._check_royalty_discrepancies()
        results.royalty_discrepancies = discrepancies
        
        # Check 3: Pending takedowns past SLA
        overdue_takedowns = await self._check_overdue_takedowns()
        results.overdue_takedowns = overdue_takedowns
        if overdue_takedowns:
            await self._escalate_overdue_takedowns(overdue_takedowns)
        
        # Check 4: Stream count reconciliation
        stream_issues = await self._reconcile_stream_counts()
        results.stream_count_issues = stream_issues
        
        # Store report
        await self._store_compliance_report(results)
        
        return results
    
    async def _check_territorial_violations(self) -> List[TerritorialViolation]:
        """
        Find streams that shouldn't have been allowed.
        
        This catches edge cases where restrictions weren't enforced.
        """
        # Query for streams of blocked content
        violations = await self.db.fetch_all('''
            SELECT 
                se.event_id,
                se.track_id,
                se.territory,
                se.user_id,
                se.timestamp,
                ta.restriction_reason
            FROM streaming_events se
            JOIN track_availability ta ON 
                se.track_id = ta.track_id AND 
                se.territory = ta.territory_code
            WHERE 
                se.timestamp > NOW() - INTERVAL '24 hours'
                AND ta.availability = 'blocked'
        ''')
        
        return [
            TerritorialViolation(
                event_id=v['event_id'],
                track_id=v['track_id'],
                territory=v['territory'],
                reason=v['restriction_reason']
            )
            for v in violations
        ]
    
    async def generate_rights_holder_report(
        self,
        rights_holder_id: str,
        period: str  # "2024-Q1"
    ) -> RightsHolderReport:
        """
        Generate detailed report for rights holder.
        
        Rights holders often have audit rights and request detailed data.
        """
        report = RightsHolderReport(
            rights_holder_id=rights_holder_id,
            period=period
        )
        
        # Get all tracks owned by rights holder
        owned_tracks = await self._get_owned_tracks(rights_holder_id)
        
        # Stream counts by territory
        territory_streams = await self.db.fetch_all('''
            SELECT 
                territory,
                COUNT(*) as stream_count,
                SUM(royalty_amount) as total_royalties
            FROM royalty_records
            WHERE 
                recipient_id = %s
                AND period = %s
            GROUP BY territory
        ''', (rights_holder_id, period))
        
        report.streams_by_territory = territory_streams
        
        # Top tracks
        report.top_tracks = await self._get_top_tracks(
            owned_tracks, 
            period,
            limit=100
        )
        
        # Demographic breakdown (aggregated, not individual)
        report.demographics = await self._get_demographic_breakdown(
            owned_tracks,
            period
        )
        
        # Comparison to previous period
        report.period_comparison = await self._compare_to_previous(
            rights_holder_id,
            period
        )
        
        return report
    
    async def prepare_audit_package(
        self,
        audit_request: AuditRequest
    ) -> AuditPackage:
        """
        Prepare data package for external audit.
        
        Must include all data needed to verify royalty calculations.
        """
        package = AuditPackage(
            audit_id=audit_request.id,
            period=audit_request.period
        )
        
        # Raw streaming events (anonymized)
        package.streaming_events = await self._export_streaming_events(
            audit_request.period,
            anonymize=True
        )
        
        # Royalty calculation logic documentation
        package.calculation_methodology = self._get_royalty_methodology()
        
        # Rate cards by territory
        package.rate_cards = await self._export_rate_cards(audit_request.period)
        
        # Sample transactions for verification
        package.sample_transactions = await self._get_sample_transactions(
            audit_request.period,
            sample_size=10000
        )
        
        return package

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

class RegionalComplianceHandler:
    """
    Handles region-specific compliance requirements.
    """
    
    async def check_region_requirements(
        self,
        stream_event: StreamEvent
    ) -> RegionalComplianceResult:
        """
        Apply region-specific compliance checks.
        """
        territory = stream_event.territory
        
        result = RegionalComplianceResult()
        
        # EU: Check roaming regulations
        if territory in EU_COUNTRIES:
            result.roaming_check = await self._check_eu_roaming(stream_event)
        
        # China: Content approval status
        if territory == 'CN':
            result.content_approved = await self._check_china_approval(
                stream_event.track_id
            )
            if not result.content_approved:
                result.block_stream = True
                result.reason = 'content_not_approved'
        
        # Germany: GEMA specific reporting
        if territory == 'DE':
            result.gema_report = self._generate_gema_record(stream_event)
        
        # US: Check for compulsory license vs negotiated
        if territory == 'US':
            result.license_type = await self._determine_us_license_type(
                stream_event.track_id
            )
        
        return result
    
    async def _check_eu_roaming(self, stream_event: StreamEvent) -> RoamingCheck:
        """
        EU Portability Regulation (2017/1128):
        Users can access their home country catalog while temporarily 
        in another EU country for up to 90 days per year.
        """
        user = await self.user_service.get(stream_event.user_id)
        home_country = user.registration_country
        current_country = stream_event.territory
        
        if home_country not in EU_COUNTRIES or current_country not in EU_COUNTRIES:
            return RoamingCheck(applicable=False)
        
        if home_country == current_country:
            return RoamingCheck(applicable=False)
        
        # Check roaming history
        roaming_days = await self._count_roaming_days(
            user.id,
            current_country,
            year=datetime.now().year
        )
        
        if roaming_days > 90:
            # User may need to verify residency or change home country
            return RoamingCheck(
                applicable=True,
                status='exceeded',
                days_used=roaming_days,
                action='prompt_residency_verification'
            )
        
        # Roaming allowed - serve home country catalog
        return RoamingCheck(
            applicable=True,
            status='allowed',
            effective_territory=home_country,  # Use home catalog
            days_used=roaming_days
        )
    
    def get_regional_rate_card(self, territory: str) -> RoyaltyRates:
        """
        Get royalty rates for a territory.
        
        Rates vary significantly by region based on:
        - Average Revenue Per User (ARPU)
        - Collective licensing agreements
        - Regulatory requirements
        """
        return REGIONAL_RATES.get(territory, REGIONAL_RATES['DEFAULT'])
 
# Example regional rates (simplified)
REGIONAL_RATES = {
    'US': RoyaltyRates(
        master_rate=0.0037,
        composition_rate=0.0007,
        stream_value=0.0044
    ),
    'UK': RoyaltyRates(
        master_rate=0.0031,
        composition_rate=0.0006,
        stream_value=0.0037
    ),
    'IN': RoyaltyRates(  # India - lower ARPU
        master_rate=0.0008,
        composition_rate=0.0002,
        stream_value=0.001
    ),
    'DEFAULT': RoyaltyRates(
        master_rate=0.002,
        composition_rate=0.0004,
        stream_value=0.0024
    )
}