import sqlite3 import json from backend.db.connection import get_db_connection from backend.logging_config import _log def run_migrations() -> None: conn = get_db_connection() try: cursor = conn.cursor() # Enable WAL mode cursor.execute("PRAGMA journal_mode=WAL") # Check if schema_version exists cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='schema_version'") has_version_table = cursor.fetchone() is not None current_version = 0 if not has_version_table: # Check if products table exists to determine baseline cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='products'") has_products_table = cursor.fetchone() is not None # Create schema_version table cursor.execute("CREATE TABLE schema_version (version INTEGER PRIMARY KEY)") if has_products_table: # Check what columns are already present cursor.execute("PRAGMA table_info(products)") columns = [row[1] for row in cursor.fetchall()] # Check if discord_subscriptions exists cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='discord_subscriptions'") has_discord = cursor.fetchone() is not None if has_discord: current_version = 2 elif "full_specs" in columns: current_version = 1 else: current_version = 0 else: current_version = 0 cursor.execute("INSERT INTO schema_version (version) VALUES (?)", (current_version,)) conn.commit() _log(f"Database schema version initialized to baseline: {current_version}") else: cursor.execute("SELECT version FROM schema_version") current_version = cursor.fetchone()[0] _log(f"Current database schema version: {current_version}") # Run migrations sequentially if current_version < 1: _log("Applying migration 1: Create base products and price_history tables.") cursor.execute(""" CREATE TABLE IF NOT EXISTS products ( product_code TEXT PRIMARY KEY, product_name TEXT, condition TEXT, first_seen TEXT, last_seen TEXT, removed_at TEXT, original_price REAL, current_price REAL, save_percent REAL, active INTEGER, specs TEXT, rating_star REAL, comment_count INTEGER, thumbnail_url TEXT, full_specs TEXT ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_code TEXT, timestamp TEXT, price REAL, save_percent REAL, FOREIGN KEY(product_code) REFERENCES products(product_code) ) """) # Catch up columns if base table existed but columns didn't cursor.execute("PRAGMA table_info(products)") columns = [row[1] for row in cursor.fetchall()] migrations = [ ("removed_at", "TEXT"), ("specs", "TEXT"), ("rating_star", "REAL"), ("comment_count", "INTEGER"), ("thumbnail_url", "TEXT"), ("full_specs", "TEXT"), ] for col_name, col_type in migrations: if col_name not in columns: _log(f"Adding column '{col_name}' ({col_type}) to products table.") cursor.execute(f"ALTER TABLE products ADD COLUMN {col_name} {col_type}") cursor.execute("UPDATE schema_version SET version = 1") conn.commit() _log("Migration 1 applied successfully.") current_version = 1 if current_version < 2: _log("Applying migration 2: Create Discord tables.") cursor.execute(""" CREATE TABLE IF NOT EXISTS discord_subscriptions ( channel_id TEXT PRIMARY KEY, guild_id TEXT NOT NULL, min_savings REAL, max_price REAL, brands TEXT, keywords TEXT, conditions TEXT, event_types TEXT DEFAULT 'all', created_at TEXT, updated_by TEXT ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS discord_authorized_users ( user_id TEXT PRIMARY KEY, added_by TEXT, added_at TEXT ) """) cursor.execute("UPDATE schema_version SET version = 2") conn.commit() _log("Migration 2 applied successfully.") current_version = 2 if current_version < 3: _log("Applying migration 3: Create watchlist table.") cursor.execute(""" CREATE TABLE IF NOT EXISTS watchlists ( user_id TEXT, product_code TEXT, target_price REAL, created_at TEXT, PRIMARY KEY (user_id, product_code) ) """) cursor.execute("UPDATE schema_version SET version = 3") conn.commit() _log("Migration 3 applied successfully.") current_version = 3 if current_version < 4: _log("Applying migration 4: Add price_delta column to price_history.") # Add signed delta column: negative = price drop, positive = price hike cursor.execute("PRAGMA table_info(price_history)") ph_columns = [row[1] for row in cursor.fetchall()] if "price_delta" not in ph_columns: cursor.execute("ALTER TABLE price_history ADD COLUMN price_delta REAL") cursor.execute("UPDATE schema_version SET version = 4") conn.commit() _log("Migration 4 applied successfully.") current_version = 4 if current_version < 5: _log("Applying migration 5: Create push_subscriptions table.") cursor.execute(""" CREATE TABLE IF NOT EXISTS push_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, endpoint TEXT UNIQUE NOT NULL, p256dh TEXT NOT NULL, auth TEXT NOT NULL, user_agent TEXT, watchlist TEXT DEFAULT '[]', notify_price_drops INTEGER DEFAULT 1, notify_new_listings INTEGER DEFAULT 1, notify_back_in_stock INTEGER DEFAULT 1, notify_watchlist_only INTEGER DEFAULT 0, min_drop_percent REAL DEFAULT 0.0, created_at TEXT, last_used TEXT ) """) cursor.execute( "CREATE INDEX IF NOT EXISTS idx_push_endpoint ON push_subscriptions(endpoint)" ) cursor.execute("UPDATE schema_version SET version = 5") conn.commit() _log("Migration 5 applied successfully.") current_version = 5 if current_version < 6: _log("Applying migration 6: Reconstruct push_subscriptions, create subscription_watchlist.") # 1. Fetch current subscriptions and their watchlists cursor.execute(""" SELECT id, endpoint, p256dh, auth, user_agent, watchlist, notify_price_drops, notify_new_listings, notify_back_in_stock, notify_watchlist_only, min_drop_percent, created_at, last_used FROM push_subscriptions """) rows = cursor.fetchall() # Store existing data existing_subs = [] for r in rows: existing_subs.append({ "id": r["id"], "endpoint": r["endpoint"], "p256dh": r["p256dh"], "auth": r["auth"], "user_agent": r["user_agent"], "watchlist": r["watchlist"], "notify_price_drops": r["notify_price_drops"], "notify_new_listings": r["notify_new_listings"], "notify_back_in_stock": r["notify_back_in_stock"], "notify_watchlist_only": r["notify_watchlist_only"], "min_drop_percent": r["min_drop_percent"], "created_at": r["created_at"], "last_used": r["last_used"] }) # 2. Drop the old table and its index cursor.execute("DROP INDEX IF EXISTS idx_push_endpoint") cursor.execute("DROP TABLE IF EXISTS push_subscriptions") # 3. Recreate push_subscriptions table without watchlist column cursor.execute(""" CREATE TABLE IF NOT EXISTS push_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, endpoint TEXT UNIQUE NOT NULL, p256dh TEXT NOT NULL, auth TEXT NOT NULL, user_agent TEXT, notify_price_drops INTEGER DEFAULT 1, notify_new_listings INTEGER DEFAULT 1, notify_back_in_stock INTEGER DEFAULT 1, notify_watchlist_only INTEGER DEFAULT 0, min_drop_percent REAL DEFAULT 0.0, created_at TEXT, last_used TEXT ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_push_endpoint ON push_subscriptions(endpoint)") # 4. Create subscription_watchlist table cursor.execute(""" CREATE TABLE IF NOT EXISTS subscription_watchlist ( id INTEGER PRIMARY KEY AUTOINCREMENT, subscription_id INTEGER NOT NULL, product_code TEXT NOT NULL, created_at TEXT, FOREIGN KEY(subscription_id) REFERENCES push_subscriptions(id) ON DELETE CASCADE, UNIQUE(subscription_id, product_code) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_sub_watchlist_code ON subscription_watchlist(product_code)") # 5. Insert old subscriptions back, keeping their IDs, and populate subscription_watchlist for sub in existing_subs: cursor.execute(""" INSERT INTO push_subscriptions ( id, endpoint, p256dh, auth, user_agent, notify_price_drops, notify_new_listings, notify_back_in_stock, notify_watchlist_only, min_drop_percent, created_at, last_used ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( sub["id"], sub["endpoint"], sub["p256dh"], sub["auth"], sub["user_agent"], sub["notify_price_drops"], sub["notify_new_listings"], sub["notify_back_in_stock"], sub["notify_watchlist_only"], sub["min_drop_percent"], sub["created_at"], sub["last_used"] )) watchlist_json = sub["watchlist"] if watchlist_json: try: codes = json.loads(watchlist_json) if isinstance(codes, list): for code in codes: cursor.execute(""" INSERT OR IGNORE INTO subscription_watchlist (subscription_id, product_code, created_at) VALUES (?, ?, ?) """, (sub["id"], code, sub["created_at"])) except Exception as e: _log(f"Warning: Failed to parse watchlist JSON for subscription {sub['id']}: {e}") cursor.execute("UPDATE schema_version SET version = 6") conn.commit() _log("Migration 6 applied successfully.") current_version = 6 if current_version < 7: _log("Applying migration 7: Add individual ping role columns to discord_subscriptions.") cursor.execute("PRAGMA table_info(discord_subscriptions)") sub_columns = [row[1] for row in cursor.fetchall()] new_cols = [ ("ping_role_added_id", "TEXT"), ("ping_role_removed_id", "TEXT"), ("ping_role_price_drop_id", "TEXT"), ("ping_role_price_hike_id", "TEXT") ] for col_name, col_type in new_cols: if col_name not in sub_columns: _log(f"Adding column '{col_name}' to discord_subscriptions table.") cursor.execute(f"ALTER TABLE discord_subscriptions ADD COLUMN {col_name} {col_type}") cursor.execute("UPDATE schema_version SET version = 7") conn.commit() _log("Migration 7 applied successfully.") current_version = 7 if current_version < 8: _log("Applying migration 8: Create stock_history table and index.") cursor.execute(""" CREATE TABLE IF NOT EXISTS stock_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_code TEXT, event_type TEXT NOT NULL, timestamp TEXT NOT NULL, price REAL, FOREIGN KEY(product_code) REFERENCES products(product_code) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_stock_history_code ON stock_history(product_code)") cursor.execute("UPDATE schema_version SET version = 8") conn.commit() _log("Migration 8 applied successfully.") current_version = 8 if current_version < 9: _log("Applying migration 9: Add oscillation-filter columns to products.") cursor.execute("PRAGMA table_info(products)") prod_columns = [row[1] for row in cursor.fetchall()] new_cols = [ ("stability_count", "INTEGER DEFAULT 0"), ("pending_state", "TEXT DEFAULT NULL"), ("pending_removal_since", "TEXT DEFAULT NULL"), ] for col_name, col_def in new_cols: if col_name not in prod_columns: _log(f"Adding column '{col_name}' to products table.") cursor.execute(f"ALTER TABLE products ADD COLUMN {col_name} {col_def}") cursor.execute("UPDATE schema_version SET version = 9") conn.commit() _log("Migration 9 applied successfully.") current_version = 9 if current_version < 10: _log("Applying migration 10: Add debounce_duration and message tracking tables.") cursor.execute("PRAGMA table_info(products)") prod_columns = [row[1] for row in cursor.fetchall()] if "debounce_duration" not in prod_columns: _log("Adding column 'debounce_duration' to products table.") cursor.execute("ALTER TABLE products ADD COLUMN debounce_duration INTEGER DEFAULT NULL") cursor.execute(""" CREATE TABLE IF NOT EXISTS discord_sent_messages ( product_code TEXT NOT NULL, channel_id TEXT NOT NULL, message_id TEXT NOT NULL, event_type TEXT NOT NULL, sent_at TEXT NOT NULL, PRIMARY KEY (product_code, channel_id) ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS telegram_sent_messages ( product_code TEXT NOT NULL, chat_id TEXT NOT NULL, message_id TEXT NOT NULL, event_type TEXT NOT NULL, message_type TEXT NOT NULL, sent_at TEXT NOT NULL, PRIMARY KEY (product_code, chat_id) ) """) cursor.execute("UPDATE schema_version SET version = 10") conn.commit() _log("Migration 10 applied successfully.") current_version = 10 if current_version < 11: _log("Applying migration 11: Create removal_attempts table.") # Internal-only log of every raw "product went missing" cycle, independent # of stock_history's public 'removed' events (which now only get written # once a removal is confirmed past the debounce window — see # get_product_volatility in processor.py). Keeps flap/volatility tiering # blind to nothing: it still sees every attempt, not just confirmed ones. cursor.execute(""" CREATE TABLE IF NOT EXISTS removal_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_code TEXT, timestamp TEXT ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_removal_attempts_code ON removal_attempts(product_code)") cursor.execute("UPDATE schema_version SET version = 11") conn.commit() _log("Migration 11 applied successfully.") current_version = 11 if current_version < 12: _log("Applying migration 12: Create scrape_stats and ghost tracking tables.") # scrape_stats: single-row snapshot of the last scrape cycle — how many # products Lenovo's Search API listed vs. how many survived the Compare-API # ghost filter (verify_and_enrich). Backs GET /api/stats (Phase 9 landing/ # about "Only X of Y listed laptops are purchasable" copy, D-00085/D-00112). cursor.execute(""" CREATE TABLE IF NOT EXISTS scrape_stats ( id INTEGER PRIMARY KEY CHECK (id = 1), listed_count INTEGER NOT NULL, purchasable_count INTEGER NOT NULL, ghost_count INTEGER NOT NULL, updated_at TEXT NOT NULL ) """) # ghosts_current: the current ghost set (listed on Lenovo, not cartable). # ghost_events: change-only log (per user request, 2026-07-14) — one row # when a product becomes a ghost ('appeared') and one when it stops being # one ('resolved'), NOT a row per ghost per cycle. Analysis-oriented; # ghosts never enter the products table, so name/price are captured here. cursor.execute(""" CREATE TABLE IF NOT EXISTS ghosts_current ( product_code TEXT PRIMARY KEY, product_name TEXT, price REAL, first_ghosted_at TEXT NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS ghost_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT NOT NULL, product_code TEXT NOT NULL, product_name TEXT, price REAL, event_type TEXT NOT NULL CHECK (event_type IN ('appeared', 'resolved')) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_ghost_events_code ON ghost_events(product_code)") cursor.execute("UPDATE schema_version SET version = 12") conn.commit() _log("Migration 12 applied successfully.") current_version = 12 if current_version < 13: _log("Applying migration 13: Create psref_specs table.") # Caches Lenovo PSREF's LoadSpecData response per Machine Type (first 4 # chars of product_code — see project_psref_feasibility memory), keyed by # MT rather than product_code since a spec sheet covers an entire product # line, not one SKU. Joined at read time in get_laptops_data via # substr(product_code,1,4) = machine_type, never baked into `products`. cursor.execute(""" CREATE TABLE IF NOT EXISTS psref_specs ( machine_type TEXT PRIMARY KEY, product_key TEXT, page_url TEXT, spec_data TEXT, updated_at TEXT ) """) cursor.execute("UPDATE schema_version SET version = 13") conn.commit() _log("Migration 13 applied successfully.") current_version = 13 if current_version < 14: _log("Applying migration 14: Consolidate discord_sent_messages/telegram_sent_messages into sent_messages.") cursor.execute(""" CREATE TABLE IF NOT EXISTS sent_messages ( platform TEXT NOT NULL, product_code TEXT NOT NULL, target_id TEXT NOT NULL, message_id TEXT NOT NULL, event_type TEXT NOT NULL, message_type TEXT, sent_at TEXT NOT NULL, PRIMARY KEY (platform, product_code, target_id) ) """) cursor.execute(""" INSERT OR REPLACE INTO sent_messages (platform, product_code, target_id, message_id, event_type, message_type, sent_at) SELECT 'discord', product_code, channel_id, message_id, event_type, NULL, sent_at FROM discord_sent_messages """) cursor.execute(""" INSERT OR REPLACE INTO sent_messages (platform, product_code, target_id, message_id, event_type, message_type, sent_at) SELECT 'telegram', product_code, chat_id, message_id, event_type, message_type, sent_at FROM telegram_sent_messages """) cursor.execute("DROP TABLE discord_sent_messages") cursor.execute("DROP TABLE telegram_sent_messages") cursor.execute("UPDATE schema_version SET version = 14") conn.commit() _log("Migration 14 applied successfully.") current_version = 14 if current_version < 15: _log("Applying migration 15: Create index on price_history(product_code).") # price_history is the largest, fastest-growing table and is queried # per product row via get_laptops_data's correlated latest_price_delta # subquery — EXPLAIN QUERY PLAN confirmed a full table scan per row # without this (fable audit finding B-DB-1). cursor.execute("CREATE INDEX IF NOT EXISTS idx_price_history_code ON price_history(product_code)") cursor.execute("UPDATE schema_version SET version = 15") conn.commit() _log("Migration 15 applied successfully.") current_version = 15 if current_version < 16: _log("Applying migration 16: Create timestamp indexes on stock_history/price_history.") # Backs get_activity's per-table "ORDER BY timestamp DESC LIMIT ?" # subqueries (fable audit B-PERF-2) — without these, each subquery # still needs a full sort of its table before the LIMIT applies. cursor.execute("CREATE INDEX IF NOT EXISTS idx_stock_history_timestamp ON stock_history(timestamp DESC)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_price_history_timestamp ON price_history(timestamp DESC)") cursor.execute("UPDATE schema_version SET version = 16") conn.commit() _log("Migration 16 applied successfully.") current_version = 16 finally: conn.close()