danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

Harden metadata cache: prevent simplified data from overwriting full entries, fix connection leaks, and add inline TTL enforcement

Browse source
This commit is contained in:
Broque Thomas 2026-03-14 13:39:12 -07:00
parent 0b8bfa1e6b
commit cf917279c2
3 changed files with 411 additions and 347 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
core/itunes_client.py
View file

@ -403,16 +403,31 @@ class iTunesClient:
""" """
Perform a batched lookup of artist IDs to get clean artist names. Perform a batched lookup of artist IDs to get clean artist names.
Returns a map of {artist_id: clean_artist_name} Returns a map of {artist_id: clean_artist_name}
Checks cache first to avoid unnecessary API calls.
""" """
if not artist_ids: if not artist_ids:
return {} return {}
clean_names = {} clean_names = {}
uncached_ids = []
# Check cache first
cache = get_metadata_cache()
for aid in artist_ids:
cached = cache.get_entity('itunes', 'artist', aid)
if cached and cached.get('artistName'):
clean_names[aid] = cached['artistName']
else:
uncached_ids.append(aid)
if not uncached_ids:
return clean_names
# iTunes lookup allows comma-separated IDs, but keep batch size reasonable (e.g. 50) # iTunes lookup allows comma-separated IDs, but keep batch size reasonable (e.g. 50)
batch_size = 50 batch_size = 50
for i in range(0, len(artist_ids), batch_size): for i in range(0, len(uncached_ids), batch_size):
batch = artist_ids[i:i+batch_size] batch = uncached_ids[i:i+batch_size]
ids_str = ",".join(batch) ids_str = ",".join(batch)
try: try:
@ -425,6 +440,8 @@ class iTunesClient:
a_name = item.get('artistName', '') a_name = item.get('artistName', '')
if a_id and a_name: if a_id and a_name:
clean_names[a_id] = a_name clean_names[a_id] = a_name
# Populate artist cache from lookup results
cache.store_entity('itunes', 'artist', a_id, item)
except Exception as e: except Exception as e:
logger.warning(f"Failed batch artist lookup: {e}") logger.warning(f"Failed batch artist lookup: {e}")
@ -571,10 +588,10 @@ class iTunesClient:
result = albums[:limit] result = albums[:limit]
# Cache individual albums + search mapping # Cache individual albums + search mapping (skip if full data already cached)
entries = [(str(ad.get('collectionId', '')), ad) for ad in raw_items if ad.get('collectionId')] entries = [(str(ad.get('collectionId', '')), ad) for ad in raw_items if ad.get('collectionId')]
if entries: if entries:
cache.store_entities_bulk('itunes', 'album', entries) cache.store_entities_bulk('itunes', 'album', entries, skip_if_exists=True)
# Only cache IDs for the albums we're actually returning # Only cache IDs for the albums we're actually returning
result_ids = [str(ad.get('collectionId', '')) for ad in raw_items[:limit] if ad.get('collectionId')] result_ids = [str(ad.get('collectionId', '')) for ad in raw_items[:limit] if ad.get('collectionId')]
if result_ids: if result_ids:
@ -782,13 +799,13 @@ class iTunesClient:
# Cache the album tracks listing # Cache the album tracks listing
cache.store_entity('itunes', 'album', f"{album_id}_tracks", result) cache.store_entity('itunes', 'album', f"{album_id}_tracks", result)
# Also cache individual tracks from the raw results # Also cache individual tracks from the raw results (skip if full data already cached)
track_entries = [] track_entries = []
for item in results: for item in results:
if item.get('wrapperType') == 'track' and item.get('kind') == 'song' and item.get('trackId'): if item.get('wrapperType') == 'track' and item.get('kind') == 'song' and item.get('trackId'):
track_entries.append((str(item['trackId']), item)) track_entries.append((str(item['trackId']), item))
if track_entries: if track_entries:
cache.store_entities_bulk('itunes', 'track', track_entries) cache.store_entities_bulk('itunes', 'track', track_entries, skip_if_exists=True)
return result return result
@ -1032,14 +1049,14 @@ class iTunesClient:
# Extract albums from dict # Extract albums from dict
albums = [item['album'] for item in seen_albums.values()] albums = [item['album'] for item in seen_albums.values()]
# Cache individual albums opportunistically # Cache individual albums opportunistically (skip if full data already cached)
album_entries = [] album_entries = []
for album_data in results: for album_data in results:
if album_data.get('wrapperType') == 'collection' and album_data.get('collectionId'): if album_data.get('wrapperType') == 'collection' and album_data.get('collectionId'):
album_entries.append((str(album_data['collectionId']), album_data)) album_entries.append((str(album_data['collectionId']), album_data))
if album_entries: if album_entries:
cache = get_metadata_cache() cache = get_metadata_cache()
cache.store_entities_bulk('itunes', 'album', album_entries) cache.store_entities_bulk('itunes', 'album', album_entries, skip_if_exists=True)
logger.info(f"Retrieved {len(albums)} unique albums for artist {artist_id} (filtered from {len(results)} results)") logger.info(f"Retrieved {len(albums)} unique albums for artist {artist_id} (filtered from {len(results)} results)")
return albums[:limit] return albums[:limit]

81
core/metadata_cache.py
View file

@ -46,13 +46,25 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(""" cursor.execute("""
SELECT id, raw_json FROM metadata_cache_entities SELECT id, raw_json, updated_at, ttl_days FROM metadata_cache_entities
WHERE source = ? AND entity_type = ? AND entity_id = ? WHERE source = ? AND entity_type = ? AND entity_id = ?
""", (source, entity_type, entity_id)) """, (source, entity_type, entity_id))
row = cursor.fetchone() row = cursor.fetchone()
if row: if row:
# Inline TTL check — don't serve stale data
try:
updated = datetime.fromisoformat(row['updated_at'])
age_days = (datetime.now() - updated).days
if age_days > (row['ttl_days'] or 30):
cursor.execute("DELETE FROM metadata_cache_entities WHERE id = ?", (row['id'],))
conn.commit()
return None
except (ValueError, TypeError):
pass
# Touch: update access stats # Touch: update access stats
cursor.execute(""" cursor.execute("""
UPDATE metadata_cache_entities UPDATE metadata_cache_entities
@ -60,10 +72,10 @@ class MetadataCache:
WHERE id = ? WHERE id = ?
""", (row['id'],)) """, (row['id'],))
conn.commit() conn.commit()
conn.close()
return json.loads(row['raw_json']) return json.loads(row['raw_json'])
conn.close()
return None return None
finally:
conn.close()
except Exception as e: except Exception as e:
logger.debug(f"Cache lookup error ({source}/{entity_type}/{entity_id}): {e}") logger.debug(f"Cache lookup error ({source}/{entity_type}/{entity_id}): {e}")
return None return None
@ -77,6 +89,7 @@ class MetadataCache:
raw_json = json.dumps(raw_data, default=str) raw_json = json.dumps(raw_data, default=str)
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(""" cursor.execute("""
INSERT OR REPLACE INTO metadata_cache_entities INSERT OR REPLACE INTO metadata_cache_entities
@ -118,21 +131,39 @@ class MetadataCache:
source, entity_type, entity_id, source, entity_type, entity_id,
)) ))
conn.commit() conn.commit()
finally:
conn.close() conn.close()
except Exception as e: except Exception as e:
logger.debug(f"Cache store error ({source}/{entity_type}/{entity_id}): {e}") logger.debug(f"Cache store error ({source}/{entity_type}/{entity_id}): {e}")
def store_entities_bulk(self, source: str, entity_type: str, items: List[Tuple[str, dict]]) -> None: def store_entities_bulk(self, source: str, entity_type: str, items: List[Tuple[str, dict]],
"""Store multiple entities at once. items = [(entity_id, raw_data), ...]""" skip_if_exists: bool = False) -> None:
"""Store multiple entities at once. items = [(entity_id, raw_data), ...]
Args:
skip_if_exists: If True, don't overwrite existing entries. Use this for
opportunistic caching of simplified data (e.g. from list endpoints)
to avoid replacing richer data from detail endpoints.
"""
if not items: if not items:
return return
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
for entity_id, raw_data in items: for entity_id, raw_data in items:
if not entity_id or not raw_data: if not entity_id or not raw_data:
continue continue
if skip_if_exists:
cursor.execute("""
SELECT 1 FROM metadata_cache_entities
WHERE source = ? AND entity_type = ? AND entity_id = ?
""", (source, entity_type, entity_id))
if cursor.fetchone():
continue
fields = self._extract_fields(source, entity_type, raw_data) fields = self._extract_fields(source, entity_type, raw_data)
raw_json = json.dumps(raw_data, default=str) raw_json = json.dumps(raw_data, default=str)
cursor.execute(""" cursor.execute("""
@ -141,12 +172,14 @@ class MetadataCache:
genres, popularity, followers, genres, popularity, followers,
artist_name, artist_id, release_date, total_tracks, album_type, label, artist_name, artist_id, release_date, total_tracks, album_type, label,
album_name, album_id, duration_ms, track_number, disc_number, explicit, isrc, preview_url, album_name, album_id, duration_ms, track_number, disc_number, explicit, isrc, preview_url,
raw_json, updated_at, last_accessed_at) raw_json, updated_at, last_accessed_at, access_count)
VALUES (?, ?, ?, ?, ?, ?, VALUES (?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP,
COALESCE((SELECT access_count FROM metadata_cache_entities
WHERE source = ? AND entity_type = ? AND entity_id = ?), 0) + 1)
""", ( """, (
source, entity_type, entity_id, source, entity_type, entity_id,
fields.get('name', ''), fields.get('name', ''),
@ -170,8 +203,10 @@ class MetadataCache:
fields.get('isrc'), fields.get('isrc'),
fields.get('preview_url'), fields.get('preview_url'),
raw_json, raw_json,
source, entity_type, entity_id,
)) ))
conn.commit() conn.commit()
finally:
conn.close() conn.close()
except Exception as e: except Exception as e:
logger.debug(f"Cache bulk store error ({source}/{entity_type}): {e}") logger.debug(f"Cache bulk store error ({source}/{entity_type}): {e}")
@ -186,6 +221,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
# Batch query in chunks of 500 to avoid SQLite variable limit # Batch query in chunks of 500 to avoid SQLite variable limit
for i in range(0, len(entity_ids), 500): for i in range(0, len(entity_ids), 500):
@ -208,6 +244,7 @@ class MetadataCache:
WHERE source = ? AND entity_type = ? AND entity_id IN ({ph2}) WHERE source = ? AND entity_type = ? AND entity_id IN ({ph2})
""", [source, entity_type] + found_in_chunk) """, [source, entity_type] + found_in_chunk)
conn.commit() conn.commit()
finally:
conn.close() conn.close()
missing = [eid for eid in entity_ids if eid not in found] missing = [eid for eid in entity_ids if eid not in found]
except Exception as e: except Exception as e:
@ -226,6 +263,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(""" cursor.execute("""
SELECT id, result_ids, created_at FROM metadata_cache_searches SELECT id, result_ids, created_at FROM metadata_cache_searches
@ -233,7 +271,6 @@ class MetadataCache:
""", (source, search_type, normalized, limit)) """, (source, search_type, normalized, limit))
row = cursor.fetchone() row = cursor.fetchone()
if not row: if not row:
conn.close()
return None return None
# Check TTL (7 days for searches) # Check TTL (7 days for searches)
@ -244,7 +281,6 @@ class MetadataCache:
# Expired — delete and return miss # Expired — delete and return miss
cursor.execute("DELETE FROM metadata_cache_searches WHERE id = ?", (row['id'],)) cursor.execute("DELETE FROM metadata_cache_searches WHERE id = ?", (row['id'],))
conn.commit() conn.commit()
conn.close()
return None return None
except (ValueError, TypeError): except (ValueError, TypeError):
pass pass
@ -260,7 +296,6 @@ class MetadataCache:
# Resolve entity IDs to full data # Resolve entity IDs to full data
result_ids = json.loads(row['result_ids']) result_ids = json.loads(row['result_ids'])
if not result_ids: if not result_ids:
conn.close()
return [] return []
results = [] results = []
@ -272,12 +307,13 @@ class MetadataCache:
erow = cursor.fetchone() erow = cursor.fetchone()
if erow: if erow:
results.append(json.loads(erow['raw_json'])) results.append(json.loads(erow['raw_json']))
conn.close()
# Only return if we found all (or most) entries — partial results are unreliable # Only return if we found all (or most) entries — partial results are unreliable
if len(results) >= len(result_ids) * 0.8: if len(results) >= len(result_ids) * 0.8:
return results return results
return None return None
finally:
conn.close()
except Exception as e: except Exception as e:
logger.debug(f"Search cache lookup error ({source}/{search_type}/{query}): {e}") logger.debug(f"Search cache lookup error ({source}/{search_type}/{query}): {e}")
return None return None
@ -291,6 +327,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(""" cursor.execute("""
INSERT OR REPLACE INTO metadata_cache_searches INSERT OR REPLACE INTO metadata_cache_searches
@ -302,6 +339,7 @@ class MetadataCache:
json.dumps(entity_ids), len(entity_ids), limit, json.dumps(entity_ids), len(entity_ids), limit,
)) ))
conn.commit() conn.commit()
finally:
conn.close() conn.close()
except Exception as e: except Exception as e:
logger.debug(f"Search cache store error ({source}/{search_type}/{query}): {e}") logger.debug(f"Search cache store error ({source}/{search_type}/{query}): {e}")
@ -315,6 +353,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
where_clauses = ['entity_type = ?'] where_clauses = ['entity_type = ?']
@ -370,8 +409,9 @@ class MetadataCache:
pass pass
items.append(item) items.append(item)
conn.close()
return {'items': items, 'total': total, 'offset': offset, 'limit': limit} return {'items': items, 'total': total, 'offset': offset, 'limit': limit}
finally:
conn.close()
except Exception as e: except Exception as e:
logger.error(f"Cache browse error: {e}") logger.error(f"Cache browse error: {e}")
return {'items': [], 'total': 0, 'offset': offset, 'limit': limit} return {'items': [], 'total': 0, 'offset': offset, 'limit': limit}
@ -381,6 +421,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
cursor.execute(""" cursor.execute("""
SELECT * FROM metadata_cache_entities SELECT * FROM metadata_cache_entities
@ -388,7 +429,6 @@ class MetadataCache:
""", (source, entity_type, entity_id)) """, (source, entity_type, entity_id))
row = cursor.fetchone() row = cursor.fetchone()
if not row: if not row:
conn.close()
return None return None
# Touch # Touch
@ -407,8 +447,9 @@ class MetadataCache:
item[json_field] = json.loads(item[json_field]) item[json_field] = json.loads(item[json_field])
except (json.JSONDecodeError, TypeError): except (json.JSONDecodeError, TypeError):
pass pass
conn.close()
return item return item
finally:
conn.close()
except Exception as e: except Exception as e:
logger.error(f"Cache detail error: {e}") logger.error(f"Cache detail error: {e}")
return None return None
@ -420,6 +461,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
stats = { stats = {
@ -461,8 +503,9 @@ class MetadataCache:
stats['oldest'] = row['oldest'] stats['oldest'] = row['oldest']
stats['newest'] = row['newest'] stats['newest'] = row['newest']
conn.close()
return stats return stats
finally:
conn.close()
except Exception as e: except Exception as e:
logger.error(f"Cache stats error: {e}") logger.error(f"Cache stats error: {e}")
return { return {
@ -480,6 +523,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
# Entities # Entities
@ -497,11 +541,12 @@ class MetadataCache:
search_count = cursor.rowcount search_count = cursor.rowcount
conn.commit() conn.commit()
conn.close()
total = entity_count + search_count total = entity_count + search_count
if total > 0: if total > 0:
logger.info(f"Evicted {total} expired cache entries ({entity_count} entities, {search_count} searches)") logger.info(f"Evicted {total} expired cache entries ({entity_count} entities, {search_count} searches)")
return total return total
finally:
conn.close()
except Exception as e: except Exception as e:
logger.error(f"Cache eviction error: {e}") logger.error(f"Cache eviction error: {e}")
return 0 return 0
@ -511,6 +556,7 @@ class MetadataCache:
try: try:
db = self._get_db() db = self._get_db()
conn = db._get_connection() conn = db._get_connection()
try:
cursor = conn.cursor() cursor = conn.cursor()
# Clear entities # Clear entities
@ -547,10 +593,11 @@ class MetadataCache:
search_count = cursor.rowcount search_count = cursor.rowcount
conn.commit() conn.commit()
conn.close()
total = entity_count + search_count total = entity_count + search_count
logger.info(f"Cleared {total} cache entries (source={source}, type={entity_type})") logger.info(f"Cleared {total} cache entries (source={source}, type={entity_type})")
return total return total
finally:
conn.close()
except Exception as e: except Exception as e:
logger.error(f"Cache clear error: {e}") logger.error(f"Cache clear error: {e}")
return 0 return 0

12
core/spotify_client.py
View file

@ -1049,10 +1049,10 @@ class SpotifyClient:
album = Album.from_spotify_album(album_data) album = Album.from_spotify_album(album_data)
albums.append(album) albums.append(album)
# Cache individual albums + search mapping # Cache individual albums + search mapping (skip if full data already cached)
entries = [(ad.get('id'), ad) for ad in raw_items if ad.get('id')] entries = [(ad.get('id'), ad) for ad in raw_items if ad.get('id')]
if entries: if entries:
cache.store_entities_bulk('spotify', 'album', entries) cache.store_entities_bulk('spotify', 'album', entries, skip_if_exists=True)
cache.store_search_results('spotify', 'album', query, min(limit, 10), cache.store_search_results('spotify', 'album', query, min(limit, 10),
[ad.get('id') for ad in raw_items if ad.get('id')]) [ad.get('id') for ad in raw_items if ad.get('id')])
@ -1234,14 +1234,14 @@ class SpotifyClient:
# Cache the aggregated result # Cache the aggregated result
cache.store_entity('spotify', 'album', cache_key, result) cache.store_entity('spotify', 'album', cache_key, result)
# Also cache individual tracks opportunistically # Also cache individual tracks opportunistically (skip if full data already cached)
track_entries = [] track_entries = []
for track in all_tracks: for track in all_tracks:
tid = track.get('id') tid = track.get('id')
if tid: if tid:
track_entries.append((tid, track)) track_entries.append((tid, track))
if track_entries: if track_entries:
cache.store_entities_bulk('spotify', 'track', track_entries) cache.store_entities_bulk('spotify', 'track', track_entries, skip_if_exists=True)
return result return result
@ -1278,11 +1278,11 @@ class SpotifyClient:
logger.info(f"Retrieved {len(albums)} albums for artist {artist_id}") logger.info(f"Retrieved {len(albums)} albums for artist {artist_id}")
# Cache individual albums opportunistically # Cache individual albums opportunistically (skip if full data already cached)
cache = get_metadata_cache() cache = get_metadata_cache()
entries = [(ad.get('id'), ad) for ad in raw_items if ad.get('id')] entries = [(ad.get('id'), ad) for ad in raw_items if ad.get('id')]
if entries: if entries:
cache.store_entities_bulk('spotify', 'album', entries) cache.store_entities_bulk('spotify', 'album', entries, skip_if_exists=True)
return albums return albums