"""Manifest Parser Lambda — Resonance Engine Phase 2b (Dispatcher) Parses DynamoDB export manifest, streams .json.gz data files, extracts spotify-presave fan records, and dispatches them as batched SQS messages for downstream processing by the Collector Worker. Trigger: S3 event on manifest-summary.json creation, or manual invocation. """ import gzip import json import logging import os from itertools import islice from urllib.parse import unquote_plus import boto3 logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) def handler(event, context): """Parse DDB export manifest, stream data files, and dispatch fan batches. Accepts either: - S3 event notification (ObjectCreated on manifest-summary.json) - Manual invocation: {"bucket": "...", "export_prefix": "..."} """ bucket, export_prefix = _parse_event(event) s3 = boto3.client("s3") sqs = boto3.client("sqs") queue_url = os.environ["SQS_QUEUE_URL"] try: batch_size = int(os.environ.get("FAN_BATCH_SIZE", "500")) except (TypeError, ValueError): logger.warning("Invalid FAN_BATCH_SIZE; falling back to 500") batch_size = 500 batch_size = max(1, min(batch_size, 1000)) manifest_key = f"{export_prefix}manifest-files.json" logger.info("Reading manifest from s3://%s/%s", bucket, manifest_key) response = s3.get_object(Bucket=bucket, Key=manifest_key) manifest_body = response["Body"].read().decode("utf-8") files = _parse_manifest(manifest_body) if not files: logger.warning("No data files found in manifest") return {"statusCode": 200, "fansDispatched": 0, "batchesSent": 0} total_fans = 0 total_batches = 0 for file_entry in files: s3_key = _reconstruct_key(export_prefix, file_entry["dataFileS3Key"]) logger.info( "Streaming s3://%s/%s (%d items)", bucket, s3_key, file_entry["itemCount"], ) file_fans = 0 file_batches = 0 messages_buffer = [] buffer_bytes = 0 fan_stream = _stream_presave_fans(s3, bucket, s3_key) for batch in _batch_iterator(fan_stream, batch_size): msg = { "source": "backfill", "fans": batch, "fan_count": len(batch), } msg_bytes = len(json.dumps(msg).encode("utf-8")) # SQS send_message_batch limit: 10 messages or 1MB total if messages_buffer and ( len(messages_buffer) >= 10 or buffer_bytes + msg_bytes > 1_000_000 ): sent = _send_batches(sqs, queue_url, messages_buffer) file_batches += sent messages_buffer = [] buffer_bytes = 0 messages_buffer.append(msg) buffer_bytes += msg_bytes file_fans += len(batch) if messages_buffer: sent = _send_batches(sqs, queue_url, messages_buffer) file_batches += sent logger.info( "File complete: s3://%s/%s — %d fans extracted, %d batches sent", bucket, s3_key, file_fans, file_batches, ) total_fans += file_fans total_batches += file_batches logger.info( "Dispatch complete: %d fans dispatched in %d batches from %d files", total_fans, total_batches, len(files), ) return { "statusCode": 200, "fansDispatched": total_fans, "batchesSent": total_batches, } def _parse_event(event): """Extract bucket and export_prefix from S3 event or manual invocation.""" if "Records" in event: record = event["Records"][0] bucket = record["s3"]["bucket"]["name"] key = unquote_plus(record["s3"]["object"]["key"]) export_prefix = key.rsplit("manifest-summary.json", 1)[0] return bucket, export_prefix bucket = event.get("bucket", os.environ.get("DDB_EXPORT_BUCKET", "dev-mymac80")) export_prefix = event.get("export_prefix") if not export_prefix: raise ValueError( "Manual invocation requires 'export_prefix' in the event payload. " 'Example: {"bucket": "dev-mymac80", ' '"export_prefix": "resonance-engine/ddb-export/01771804222395-8dd0c6a7/"}' ) if not export_prefix.endswith("/"): export_prefix += "/" return bucket, export_prefix def _parse_manifest(manifest_body): """Parse manifest-files.json (JSON Lines format — one object per line).""" files = [] for line in manifest_body.strip().splitlines(): line = line.strip() if not line: continue entry = json.loads(line) if "dataFileS3Key" in entry and "itemCount" in entry: files.append(entry) return files def _reconstruct_key(export_prefix, data_file_s3_key): """Reconstruct S3 key relative to the export prefix. The dataFileS3Key in the manifest references the original export path (Songwhip prod bucket). We extract the data/ relative path and prepend the current export prefix. """ marker = "/data/" idx = data_file_s3_key.find(marker) if idx != -1: relative_path = data_file_s3_key[idx + 1:] # strip leading slash elif data_file_s3_key.startswith("data/"): relative_path = data_file_s3_key else: relative_path = "data/" + data_file_s3_key.rsplit("/", 1)[-1] return f"{export_prefix}{relative_path}" # --------------------------------------------------------------------------- # 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 # --------------------------------------------------------------------------- # Fan Streaming & Batching # --------------------------------------------------------------------------- def _stream_presave_fans(s3, bucket, key): """Stream a .json.gz file from S3 and yield presave fan dicts. For each line, parses JSON, deserializes DDB type markers, filters for spotify-presave records, and yields only the 4 required fields: spotifyUserId, refreshToken, partitionKey, sortKey. Yields: dict: Fan record with only the essential fields. """ response = s3.get_object(Bucket=bucket, Key=key) with gzip.GzipFile(fileobj=response["Body"]) as gz: for line in gz: line = line.decode("utf-8").strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: logger.warning("Invalid JSON line in %s, skipping", key) 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"): continue spotify_user_id = item.get("spotifyUserId") refresh_token = item.get("refreshToken") partition_key = item.get("partitionKey") if not spotify_user_id or not refresh_token or not partition_key: logger.warning( "Presave fan in %s missing required fields; skipping. " "partitionKey=%r spotifyUserId=%r", key, partition_key, spotify_user_id, ) continue yield { "spotifyUserId": spotify_user_id, "refreshToken": refresh_token, "partitionKey": partition_key, "sortKey": sort_key, } def _batch_iterator(iterable, batch_size): """Yield lists of up to batch_size items from iterable. Args: iterable: Any iterable. batch_size: Maximum items per batch. Yields: list: A batch of up to batch_size items. """ it = iter(iterable) while batch := list(islice(it, batch_size)): yield batch # --------------------------------------------------------------------------- # SQS Dispatch # --------------------------------------------------------------------------- def _send_batches(sqs, queue_url, messages): """Send messages to SQS in batches of 10 (SQS maximum). Each message is a dict that will be JSON-serialized as the SQS body. Returns: int: Number of messages successfully enqueued. """ enqueued = 0 it = iter(messages) while batch := list(islice(it, 10)): entries = [ {"Id": str(i), "MessageBody": json.dumps(msg)} for i, msg in enumerate(batch) ] response = sqs.send_message_batch( QueueUrl=queue_url, Entries=entries ) failed = response.get("Failed", []) if failed: raise RuntimeError( f"SQS batch send had {len(failed)} failure(s): {failed}" ) enqueued += len(batch) return enqueued