Harden Amazon worker schema migration

Ensure the Amazon enrichment worker verifies its required columns before querying pending work or progress, preventing upgraded installs from spamming no-such-column errors when amazon_match_status is missing.

Add regression coverage for legacy databases without Amazon enrichment columns.

core/amazon_worker.py

@@ -19,6 +19,7 @@ class AmazonWorker:
     def __init__(self, database: MusicDatabase):
         self.db = database
         self.client = AmazonClient()
+        self._amazon_schema_checked = False
 
         self.running = False
         self.paused = False
@@ -40,6 +41,38 @@ class AmazonWorker:
 
         logger.info("Amazon background worker initialized")
 
+    def _ensure_amazon_schema(self, cursor) -> None:
+        """Ensure upgraded installs have the Amazon enrichment columns.
+
+        MusicDatabase normally runs this migration during startup, but the
+        worker should still be defensive because it is the code path that
+        repeatedly queries these columns in the background.
+        """
+        if self._amazon_schema_checked:
+            return
+
+        table_columns = {
+            'artists': ('amazon_id', 'amazon_match_status', 'amazon_last_attempted'),
+            'albums': ('amazon_id', 'amazon_match_status', 'amazon_last_attempted'),
+            'tracks': ('amazon_id', 'amazon_match_status', 'amazon_last_attempted'),
+        }
+        for table, columns in table_columns.items():
+            cursor.execute(f"PRAGMA table_info({table})")
+            existing = {row[1] for row in cursor.fetchall()}
+            for column in columns:
+                if column not in existing:
+                    column_type = 'TIMESTAMP' if column == 'amazon_last_attempted' else 'TEXT'
+                    cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {column_type}")
+
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_amazon_id ON artists (amazon_id)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_artists_amazon_status ON artists (amazon_match_status)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_amazon_id ON albums (amazon_id)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_albums_amazon_status ON albums (amazon_match_status)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_amazon_id ON tracks (amazon_id)")
+        cursor.execute("CREATE INDEX IF NOT EXISTS idx_tracks_amazon_status ON tracks (amazon_match_status)")
+        cursor.connection.commit()
+        self._amazon_schema_checked = True
+
     def start(self):
         if self.running:
             logger.warning("Worker already running")
@@ -131,6 +164,7 @@ class AmazonWorker:
         try:
             conn = self.db._get_connection()
             cursor = conn.cursor()
+            self._ensure_amazon_schema(cursor)
 
             # Priority 1: Unattempted artists
             cursor.execute("""
@@ -517,6 +551,7 @@ class AmazonWorker:
         try:
             conn = self.db._get_connection()
             cursor = conn.cursor()
+            self._ensure_amazon_schema(cursor)
             cursor.execute("""
                 SELECT
                     (SELECT COUNT(*) FROM artists WHERE amazon_match_status IS NULL AND id IS NOT NULL) +
@@ -538,6 +573,7 @@ class AmazonWorker:
         try:
             conn = self.db._get_connection()
             cursor = conn.cursor()
+            self._ensure_amazon_schema(cursor)
             progress = {}
 
             for table in ('artists', 'albums', 'tracks'):

tests/tools/test_amazon_worker_schema.py

@@ -0,0 +1,63 @@
+import sqlite3
+
+from core.amazon_worker import AmazonWorker
+
+
+class _LegacyDatabase:
+    def __init__(self, path):
+        self.path = str(path)
+        with sqlite3.connect(self.path) as conn:
+            conn.executescript(
+                """
+                CREATE TABLE artists (
+                    id INTEGER PRIMARY KEY,
+                    name TEXT
+                );
+                CREATE TABLE albums (
+                    id INTEGER PRIMARY KEY,
+                    title TEXT,
+                    artist_id INTEGER
+                );
+                CREATE TABLE tracks (
+                    id INTEGER PRIMARY KEY,
+                    title TEXT,
+                    artist_id INTEGER
+                );
+                INSERT INTO artists (id, name) VALUES (1, 'Artist A');
+                INSERT INTO albums (id, title, artist_id) VALUES (10, 'Album A', 1);
+                INSERT INTO tracks (id, title, artist_id) VALUES (100, 'Track A', 1);
+                """
+            )
+
+    def _get_connection(self):
+        return sqlite3.connect(self.path)
+
+
+def _columns(db_path, table):
+    with sqlite3.connect(str(db_path)) as conn:
+        return {row[1] for row in conn.execute(f"PRAGMA table_info({table})")}
+
+
+def test_amazon_worker_self_heals_legacy_schema_before_selecting_next_item(tmp_path):
+    db_path = tmp_path / "legacy.db"
+    db = _LegacyDatabase(db_path)
+    worker = AmazonWorker(db)
+
+    item = worker._get_next_item()
+
+    assert item == {"type": "artist", "id": 1, "name": "Artist A"}
+    for table in ("artists", "albums", "tracks"):
+        assert {"amazon_id", "amazon_match_status", "amazon_last_attempted"} <= _columns(db_path, table)
+
+
+def test_amazon_worker_stats_self_heal_legacy_schema(tmp_path):
+    db_path = tmp_path / "legacy.db"
+    db = _LegacyDatabase(db_path)
+    worker = AmazonWorker(db)
+
+    assert worker._count_pending_items() == 3
+    assert worker._get_progress_breakdown() == {
+        "artists": {"matched": 0, "total": 1, "percent": 0},
+        "albums": {"matched": 0, "total": 1, "percent": 0},
+        "tracks": {"matched": 0, "total": 1, "percent": 0},
+    }