"""Collector Worker Lambda — Resonance Engine Phase 3a / 5b Processes fan records from SQS and produces Spotify data to Kafka (MSK). Supports multiple message formats: - Fan batch: {"source": "backfill", "fans": [...], "fan_count": N} Pre-extracted fan records from the Manifest Parser dispatcher. - S3 file path: {"bucket": "...", "key": "..."} Legacy format for streaming gzipped DynamoDB export files. - Snowflake re-collection: fan batches with UPPERCASE keys (Phase 5b). For each fan: refreshes the OAuth token, calls Spotify API endpoints, and produces results (including token updates) to Kafka. Trigger: SQS event source mapping from the manifest-files queue. """ import base64 import gzip import json import logging import os import random import time from datetime import datetime, timezone import boto3 import requests logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) # --------------------------------------------------------------------------- # Custom Exceptions # --------------------------------------------------------------------------- class TokenRefreshError(Exception): """Spotify token refresh failed (revoked, expired, or network error).""" class SpotifyApiError(Exception): """Spotify API call failed (401, 429 exhausted, 5xx, or timeout).""" class SpotifyRateLimitError(SpotifyApiError): """Raised when Retry-After exceeds threshold, indicating severe throttling.""" # --------------------------------------------------------------------------- # DynamoDB JSON Deserialization # --------------------------------------------------------------------------- def _deserialize_ddb_item(raw_item): """Convert DynamoDB JSON (with type markers) to a plain Python dict. Handles S, N, BOOL, NULL, L, M type descriptors. """ if isinstance(raw_item, dict): if len(raw_item) == 1: type_key = next(iter(raw_item)) value = raw_item[type_key] if type_key == "S": return value if type_key == "N": try: return int(value) except ValueError: return float(value) if type_key == "BOOL": return value if type_key == "NULL": return None if type_key == "L": return [_deserialize_ddb_item(item) for item in value] if type_key == "M": return {k: _deserialize_ddb_item(v) for k, v in value.items()} # Not a type descriptor — regular dict with nested DDB items return {k: _deserialize_ddb_item(v) for k, v in raw_item.items()} return raw_item # --------------------------------------------------------------------------- # Spotify OAuth # --------------------------------------------------------------------------- def _refresh_token(refresh_token, client_id, client_secret): """Refresh a Spotify OAuth access token. Returns: tuple: (access_token, new_refresh_token or None) Raises: TokenRefreshError: If the refresh fails. """ auth_str = f"{client_id}:{client_secret}" auth_b64 = base64.b64encode(auth_str.encode("ascii")).decode("ascii") try: resp = requests.post( "https://accounts.spotify.com/api/token", data={ "grant_type": "refresh_token", "refresh_token": refresh_token, }, headers={ "Authorization": f"Basic {auth_b64}", "Content-Type": "application/x-www-form-urlencoded", }, timeout=10, ) resp.raise_for_status() data = resp.json() return data["access_token"], data.get("refresh_token") except requests.RequestException as exc: error_msg = "request failed" if hasattr(exc, "response") and exc.response is not None: try: error_data = exc.response.json() error_desc = error_data.get("error_description", "") # Sanitize: only pass through known safe error descriptions safe_errors = { "invalid_grant", "invalid_client", "invalid_request", "unauthorized_client", "unsupported_grant_type", } error_type = error_data.get("error", "") if error_type in safe_errors: error_msg = error_desc or error_type else: error_msg = f"HTTP {exc.response.status_code}" except ValueError: error_msg = f"HTTP {exc.response.status_code}" raise TokenRefreshError( f"Token refresh failed: {error_msg}" ) from exc # --------------------------------------------------------------------------- # Spotify API # --------------------------------------------------------------------------- _MAX_RATE_LIMIT_RETRIES = int( os.environ.get("SPOTIFY_RATE_LIMIT_RETRIES", "3") ) _RATE_LIMIT_THRESHOLD_SECONDS = 10 def _call_spotify_api(access_token, endpoint, params=None): """Call a Spotify API endpoint with Bearer auth. Handles 429 (rate limit) with exponential backoff + jitter. Raises SpotifyApiError on 401 — caller logs and skips the fan. Returns: dict: JSON response. Raises: SpotifyApiError: On non-recoverable failure. """ url = f"https://api.spotify.com/v1{endpoint}" headers = {"Authorization": f"Bearer {access_token}"} try: for attempt in range(_MAX_RATE_LIMIT_RETRIES + 1): resp = requests.get( url, headers=headers, params=params, timeout=30 ) if resp.status_code != 429: break if attempt == _MAX_RATE_LIMIT_RETRIES: raise SpotifyApiError( f"Rate limit exhausted on {endpoint} after " f"{attempt + 1} attempts" ) try: retry_after = int(resp.headers.get("Retry-After", 5)) except (ValueError, TypeError): retry_after = 5 if retry_after > _RATE_LIMIT_THRESHOLD_SECONDS: raise SpotifyRateLimitError( f"Retry-After {retry_after}s exceeds threshold " f"({_RATE_LIMIT_THRESHOLD_SECONDS}s) on {endpoint}" ) sleep_time = min( retry_after * (2 ** attempt), 30 ) + random.uniform(0, 1) logger.warning( "Rate limited on %s attempt %d/%d, sleeping %.1fs", endpoint, attempt + 1, _MAX_RATE_LIMIT_RETRIES + 1, sleep_time, ) time.sleep(sleep_time) if resp.status_code == 401: raise SpotifyApiError(f"Unauthorized (401) on {endpoint}") resp.raise_for_status() return resp.json() except SpotifyApiError: raise except requests.RequestException as exc: raise SpotifyApiError( f"Spotify API error on {endpoint}: {exc}" ) from exc # --------------------------------------------------------------------------- # Kafka # --------------------------------------------------------------------------- def _create_kafka_producer(): """Create a Kafka producer if KAFKA_BOOTSTRAP_SERVERS is configured. Returns: Producer or None: Kafka producer, or None if not configured. """ servers = os.environ.get("KAFKA_BOOTSTRAP_SERVERS", "") if not servers: logger.info("KAFKA_BOOTSTRAP_SERVERS not set — Kafka disabled") return None try: from confluent_kafka import Producer except ImportError: logger.warning("confluent_kafka not installed — Kafka disabled") return None config = {"bootstrap.servers": servers} security_protocol = os.environ.get("KAFKA_SECURITY_PROTOCOL", "SSL") if security_protocol: config["security.protocol"] = security_protocol sasl_mechanism = os.environ.get("KAFKA_SASL_MECHANISM", "") if sasl_mechanism: config["sasl.mechanism"] = sasl_mechanism # TODO: When MSK is provisioned, implement OAUTHBEARER callback for # AWS_MSK_IAM auth using aws-msk-iam-sasl-signer package. return Producer(config) def _delivery_report(err, msg): """Kafka delivery callback — logs failures.""" if err is not None: key = msg.key().decode("utf-8") if msg.key() else "unknown" logger.error("Kafka delivery failed for key %s: %s", key, err) def _produce_to_kafka(producer, topic, key, message): """Produce a message to Kafka. Args: producer: confluent_kafka.Producer instance. topic: Kafka topic name. key: Message key (string). message: Message value (dict, will be JSON-serialized). """ producer.produce( topic, value=json.dumps(message).encode("utf-8"), key=key.encode("utf-8"), on_delivery=_delivery_report, ) producer.poll(0) # --------------------------------------------------------------------------- # Fan Record Normalization # --------------------------------------------------------------------------- # Mapping from Snowflake UPPERCASE keys to camelCase fan record keys. _SNOWFLAKE_KEY_MAP = { "SPOTIFY_USER_ID": "spotifyUserId", "REFRESH_TOKEN": "refreshToken", "PARTITION_KEY": "partitionKey", "SORT_KEY": "sortKey", } def _normalize_fan_record(raw): """Normalize a fan record from different sources to a common format. Backfill/DDB format (camelCase keys) is returned as-is. Snowflake re-collection format (UPPERCASE keys) is mapped to camelCase. Returns: dict: Normalized fan record with camelCase keys. """ # Detect Snowflake format by checking for uppercase keys if "SPOTIFY_USER_ID" in raw or "REFRESH_TOKEN" in raw: return { camel: raw.get(upper, "") for upper, camel in _SNOWFLAKE_KEY_MAP.items() } # Backfill/DDB format — return as-is, defaulting optional keys result = dict(raw) result.setdefault("partitionKey", "") result.setdefault("sortKey", "") return result # --------------------------------------------------------------------------- # Fan Processing # --------------------------------------------------------------------------- def _process_fan(fan_record, client_id, client_secret, producer, topic): """Process a single fan: refresh token, call Spotify, produce to Kafka. Returns: dict: {success, timings} where timings has per-step ms values, or {success: False} on error. """ spotify_user_id = fan_record.get("spotifyUserId", "unknown") refresh_tok = fan_record.get("refreshToken") timings = {} if not refresh_tok: logger.warning( "No refresh token for fan %s, skipping", spotify_user_id ) return {"success": False, "timings": timings} try: t0 = time.monotonic() access_token, new_refresh = _refresh_token( refresh_tok, client_id, client_secret ) timings["token_ms"] = round((time.monotonic() - t0) * 1000) if new_refresh and new_refresh != refresh_tok: # TODO (Phase 3b): Write new_refresh back to Songwhip DDB # using ConditionExpression to prevent concurrent refresh race. logger.info( "Refresh token rotated for fan %s — DDB update deferred", spotify_user_id, ) t0 = time.monotonic() top_artists = _call_spotify_api( access_token, "/me/top/artists", {"limit": 50, "time_range": "medium_term"}, ) timings["top_artists_ms"] = round((time.monotonic() - t0) * 1000) t0 = time.monotonic() recently_played = _call_spotify_api( access_token, "/me/player/recently-played", {"limit": 50}, ) timings["recently_played_ms"] = round((time.monotonic() - t0) * 1000) message = { "spotify_user_id": spotify_user_id, "partition_key": fan_record.get("partitionKey", ""), "sort_key": fan_record.get("sortKey", ""), "refresh_token": refresh_tok, "new_refresh_token": new_refresh if new_refresh and new_refresh != refresh_tok else None, "endpoints": { "top_artists": top_artists, "recently_played": recently_played, }, "collected_at": datetime.now(timezone.utc).isoformat(), } if producer is not None: _produce_to_kafka(producer, topic, spotify_user_id, message) else: logger.info( "Kafka disabled — collected data for fan %s", spotify_user_id, ) return {"success": True, "timings": timings} except TokenRefreshError: logger.warning( "Token refresh failed for fan %s, skipping", spotify_user_id ) return {"success": False, "timings": timings} except SpotifyRateLimitError: raise # Let circuit breaker handle this except SpotifyApiError as exc: logger.warning( "Spotify API error for fan %s: %s", spotify_user_id, exc ) return {"success": False, "timings": timings} except Exception: logger.exception( "Unexpected error processing fan %s", spotify_user_id ) return {"success": False, "timings": timings} # --------------------------------------------------------------------------- # Fan Batch Processing # --------------------------------------------------------------------------- def _process_fan_batch( fans, client_id, client_secret, producer, topic, context ): """Process a batch of pre-extracted fan records with circuit breaker. Iterates through fan records, normalizing each to camelCase format, and processing via _process_fan(). Tracks consecutive rate-limit errors and trips a circuit breaker if the threshold is reached. Returns: dict: {processed, errors, rate_limited, timed_out} Raises: SpotifyRateLimitError: If circuit breaker trips (consecutive rate-limit errors reach the threshold). """ timeout_buffer_ms = int( os.environ.get("TIMEOUT_BUFFER_MS", "90000") ) circuit_breaker_threshold = int( os.environ.get("CIRCUIT_BREAKER_THRESHOLD", "3") ) processed = 0 errors = 0 rate_limited = 0 consecutive_rate_limits = 0 timed_out = False all_token_ms = [] all_top_artists_ms = [] all_recently_played_ms = [] for raw_fan in fans: # Timeout check if context and hasattr(context, "get_remaining_time_in_millis"): if context.get_remaining_time_in_millis() < timeout_buffer_ms: logger.warning( "Timeout approaching (%dms remaining), stopping", context.get_remaining_time_in_millis(), ) timed_out = True break fan = _normalize_fan_record(raw_fan) try: result = _process_fan( fan, client_id, client_secret, producer, topic ) timings = result.get("timings", {}) if timings.get("token_ms") is not None: all_token_ms.append(timings["token_ms"]) if timings.get("top_artists_ms") is not None: all_top_artists_ms.append(timings["top_artists_ms"]) if timings.get("recently_played_ms") is not None: all_recently_played_ms.append(timings["recently_played_ms"]) if result["success"]: processed += 1 consecutive_rate_limits = 0 # Reset on success else: errors += 1 consecutive_rate_limits = 0 # Non-rate-limit error resets except SpotifyRateLimitError: rate_limited += 1 consecutive_rate_limits += 1 if consecutive_rate_limits >= circuit_breaker_threshold: logger.warning( "Circuit breaker tripped: %d consecutive rate limits, " "failing batch for retry", consecutive_rate_limits, ) raise # Below threshold — skip this fan, continue with next logger.warning( "Rate limited on fan %s, skipping", fan.get("spotifyUserId", "unknown"), ) # Log per-endpoint timing summary (one line per batch) def _avg(vals): return round(sum(vals) / len(vals)) if vals else 0 def _p95(vals): if not vals: return 0 s = sorted(vals) return s[int(len(s) * 0.95)] logger.info( "Batch timing: %d ok, %d err | " "token_refresh avg=%dms p95=%dms (n=%d) | " "top_artists avg=%dms p95=%dms (n=%d) | " "recently_played avg=%dms p95=%dms (n=%d)", processed, errors, _avg(all_token_ms), _p95(all_token_ms), len(all_token_ms), _avg(all_top_artists_ms), _p95(all_top_artists_ms), len(all_top_artists_ms), _avg(all_recently_played_ms), _p95(all_recently_played_ms), len(all_recently_played_ms), ) return { "processed": processed, "errors": errors, "rate_limited": rate_limited, "timed_out": timed_out, } # --------------------------------------------------------------------------- # File Processing # --------------------------------------------------------------------------- def _process_file( s3, bucket, key, producer, topic, client_id, client_secret, context ): """Stream a .json.gz DDB export file and process each presave fan. Returns: dict: {processed, skipped, errors, timed_out} """ timeout_buffer_ms = int(os.environ.get("TIMEOUT_BUFFER_MS", "90000")) logger.info("Processing s3://%s/%s", bucket, key) response = s3.get_object(Bucket=bucket, Key=key) processed = 0 skipped = 0 errors = 0 timed_out = False with gzip.GzipFile(fileobj=response["Body"]) as gz: for line in gz: if context and hasattr(context, "get_remaining_time_in_millis"): remaining = context.get_remaining_time_in_millis() if remaining < timeout_buffer_ms: logger.warning( "Timeout approaching (%dms remaining), stopping", remaining, ) timed_out = True break line = line.decode("utf-8").strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: logger.warning("Invalid JSON line, skipping") skipped += 1 continue raw_item = raw.get("Item", raw) item = _deserialize_ddb_item(raw_item) sort_key = item.get("sortKey", "") if not sort_key.startswith("task:spotify-presave"): skipped += 1 continue result = _process_fan( item, client_id, client_secret, producer, topic ) if result["success"]: processed += 1 else: errors += 1 logger.info( "File complete: processed=%d skipped=%d errors=%d timed_out=%s", processed, skipped, errors, timed_out, ) return { "processed": processed, "skipped": skipped, "errors": errors, "timed_out": timed_out, } # --------------------------------------------------------------------------- # Lambda Handler # --------------------------------------------------------------------------- def handler(event, context): """SQS-triggered Lambda handler for collecting Spotify data. Dispatches based on SQS message format: - ``{"fans": [...], ...}`` — fan batch from dispatcher or re-collection - ``{"bucket": "...", "key": "..."}`` — legacy S3 file path Returns partial batch failure response for failed SQS messages. """ client_id = os.environ["SPOTIFY_CLIENT_ID"] client_secret = os.environ["SPOTIFY_CLIENT_SECRET"] topic = os.environ.get("KAFKA_TOPIC", "resonance-engine.spotify-data") environment = os.environ.get("ENVIRONMENT", "dev") logger.info( "Collector Worker invoked: %d SQS records, env=%s", len(event.get("Records", [])), environment, ) s3 = boto3.client("s3") producer = _create_kafka_producer() batch_item_failures = [] for record in event.get("Records", []): message_id = record["messageId"] try: body = json.loads(record["body"]) if "fans" in body: # Fan batch from dispatcher (backfill or re-collection) logger.info( "Processing SQS message %s: fan batch " "(%d fans, source=%s)", message_id, body.get("fan_count", len(body["fans"])), body.get("source", "unknown"), ) result = _process_fan_batch( body["fans"], client_id, client_secret, producer, topic, context, ) elif "bucket" in body and "key" in body: # Legacy S3 file path bucket = body["bucket"] key = body["key"] logger.info( "Processing SQS message %s: s3://%s/%s", message_id, bucket, key, ) result = _process_file( s3, bucket, key, producer, topic, client_id, client_secret, context, ) else: # Intentionally consume (not retry) unknown formats — these # are not retryable without a code fix. Sends to DLQ would # just accumulate unprocessable messages. logger.warning( "Unknown message format in SQS message %s, skipping", message_id, ) continue if result["timed_out"]: logger.warning( "SQS message %s timed out — reporting as " "failure for retry", message_id, ) batch_item_failures.append( {"itemIdentifier": message_id} ) except SpotifyRateLimitError: logger.warning( "Circuit breaker tripped for SQS message %s — " "reporting as failure for retry", message_id, ) batch_item_failures.append({"itemIdentifier": message_id}) except Exception: logger.exception( "Failed to process SQS message %s", message_id ) batch_item_failures.append({"itemIdentifier": message_id}) if producer is not None: undelivered = producer.flush(timeout=30) if undelivered > 0: logger.error( "Kafka flush: %d messages undelivered — failing all " "SQS messages for retry", undelivered, ) failed_ids = { f["itemIdentifier"] for f in batch_item_failures } for record in event.get("Records", []): if record["messageId"] not in failed_ids: batch_item_failures.append( {"itemIdentifier": record["messageId"]} ) else: logger.info("Kafka producer flushed") if batch_item_failures: logger.warning( "%d of %d SQS messages failed", len(batch_item_failures), len(event.get("Records", [])), ) return {"batchItemFailures": batch_item_failures}