From 232dc16924bbecf1848c2319e0addcd36d119f99 Mon Sep 17 00:00:00 2001
From: cryptorugmunch <admin@rugmunch.io>
Date: Fri, 3 Jul 2026 01:25:14 +0200
Subject: [PATCH] feat(sql): migrate intelligence, monitor, and referrals to
 SQLAlchemy

- intelligence.py: IntelSnapshot table, JSONL replaced with queries.
- monitor.py: Monitor + MonitorRun tables, JSON files replaced.
- structure_monitor.py: StructureSnapshot table for selector history.
- referrals.py: ReferralClick + ReferralConversion + ReferralInAppUsage tables.
- db.py: extended models and import_json_stores for the three domains.
- Added/updated tests: 475 passing.
---
 db.py                           |   4 +-
 intelligence.py                 | 112 +++++++--------
 monitor.py                      | 234 ++++++++++++++++++-------------
 referrals.py                    | 234 +++++++++++++++++--------------
 structure_monitor.py            | 223 +++++++++++++++++------------
 tests/test_intelligence.py      | 241 ++++++++++++++++++++++++++++----
 tests/test_monitor.py           | 120 +++++++++++++++-
 tests/test_referrals.py         |  96 ++++++++++++-
 tests/test_structure_monitor.py | 121 ++++++++++++++--
 9 files changed, 985 insertions(+), 400 deletions(-)

diff --git a/db.py b/db.py
index 42876b4..85d2987 100644
--- a/db.py
+++ b/db.py
@@ -438,7 +438,9 @@ if _HAS_SA:
     class ReferralConversion(Base):
         __tablename__ = "referral_conversions"
         id = Column(Integer, primary_key=True)
-        click_id = Column(String(64), ForeignKey("referral_clicks.click_id"), index=True, nullable=False)
+        click_id = Column(
+            String(64), ForeignKey("referral_clicks.click_id"), index=True, nullable=False
+        )
         provider = Column(String(64), index=True, nullable=False)
         revenue_usd = Column(Float, default=0.0)
         notes = Column(Text, default="")
diff --git a/intelligence.py b/intelligence.py
index 1771b12..704a6a4 100644
--- a/intelligence.py
+++ b/intelligence.py
@@ -6,28 +6,19 @@ Historical snapshots, anomaly detection, natural-language alerts, weekly reports
 #
 # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
 # Licensed under MIT. See LICENSE.
-import json
 import logging
 import statistics
 import time
-from datetime import UTC, datetime
-from pathlib import Path
+from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from paths import PRY_DATA_DIR
+from sqlalchemy import desc
+from sqlalchemy.exc import SQLAlchemyError
+
+from db import IntelSnapshot, session_scope
 
 logger = logging.getLogger(__name__)
 
-INTEL_DIR = PRY_DATA_DIR / "intel"
-INTEL_DIR.mkdir(parents=True, exist_ok=True)
-
-
-# ── Historical Snapshots ──
-
-
-def _snapshot_path(competitor_id: str) -> Path:
-    return INTEL_DIR / f"{competitor_id}_snapshots.jsonl"
-
 
 def record_snapshot(
     competitor_id: str,
@@ -35,25 +26,32 @@ def record_snapshot(
     url: str,
     fields: dict[str, Any],
 ) -> dict[str, Any]:
-    """Record a data snapshot for a competitor.
-
-    Each snapshot is appended to a JSONL file for the competitor.
-    """
-    snapshot = {
-        "ts": datetime.now(UTC).isoformat(),
-        "unix_ts": time.time(),
-        "competitor_id": competitor_id,
-        "competitor_name": competitor_name,
-        "url": url,
-        "fields": fields,
-    }
-    path = _snapshot_path(competitor_id)
+    """Record a data snapshot for a competitor."""
     try:
-        with open(path, "a") as f:
-            f.write(json.dumps(snapshot) + "\n")
+        with session_scope() as session:
+            snap = IntelSnapshot(
+                competitor_id=competitor_id,
+                competitor_name=competitor_name,
+                url=url,
+                fields=fields,
+            )
+            session.add(snap)
+            session.flush()
+            ts = snap.ts
+            result = {
+                "ts": ts.isoformat(),
+                "unix_ts": ts.timestamp(),
+                "competitor_id": competitor_id,
+                "competitor_name": competitor_name,
+                "url": url,
+                "fields": fields,
+            }
         logger.info("snapshot_recorded", extra={"competitor": competitor_name})
-        return snapshot
-    except OSError as e:
+        return result
+    except SQLAlchemyError as e:
+        logger.error(
+            "snapshot_record_failed", extra={"competitor": competitor_name, "error": str(e)}
+        )
         return {"error": str(e)}
 
 
@@ -63,31 +61,31 @@ def get_snapshots(
     since_hours: int | None = None,
 ) -> list[dict[str, Any]]:
     """Get snapshots for a competitor, most recent first."""
-    path = _snapshot_path(competitor_id)
-    if not path.exists():
-        return []
-
-    snapshots = []
     try:
-        for line in path.read_text().splitlines():
-            if not line.strip():
-                continue
-            snapshots.append(json.loads(line))
-    except (json.JSONDecodeError, OSError):
+        with session_scope() as session:
+            query = session.query(IntelSnapshot).filter(
+                IntelSnapshot.competitor_id == competitor_id,
+            )
+            if since_hours is not None:
+                cutoff = datetime.now(UTC) - timedelta(hours=since_hours)
+                query = query.filter(IntelSnapshot.ts >= cutoff)
+            query = query.order_by(desc(IntelSnapshot.ts)).limit(limit)
+            rows = query.all()
+            return [
+                {
+                    "ts": row.ts.isoformat() if row.ts else "",
+                    "unix_ts": row.ts.timestamp() if row.ts else 0,
+                    "competitor_id": row.competitor_id,
+                    "competitor_name": row.competitor_name,
+                    "url": row.url,
+                    "fields": row.fields or {},
+                }
+                for row in rows
+            ]
+    except SQLAlchemyError as e:
+        logger.error("snapshot_query_failed", extra={"competitor": competitor_id, "error": str(e)})
         return []
 
-    # Filter by time
-    if since_hours:
-        cutoff = time.time() - (since_hours * 3600)
-        snapshots = [s for s in snapshots if s.get("unix_ts", 0) >= cutoff]
-
-    # Sort by time (newest first) and limit
-    snapshots.sort(key=lambda x: x.get("unix_ts", 0), reverse=True)
-    return snapshots[:limit]
-
-
-# ── Anomaly Detection ──
-
 
 def compute_field_statistics(
     snapshots: list[dict[str, Any]],
@@ -154,9 +152,6 @@ def detect_anomalies_numeric(
     }
 
 
-# ── Natural Language Alerts ──
-
-
 def generate_alert(
     competitor_name: str,
     field: str,
@@ -195,9 +190,6 @@ def generate_alert(
     return alert
 
 
-# ── Weekly Reports ──
-
-
 def generate_weekly_report(
     competitors: list[dict[str, Any]],
     days_back: int = 7,
@@ -215,7 +207,6 @@ def generate_weekly_report(
         if not weekly:
             continue
 
-        # Get fields that changed this week
         if len(weekly) >= 2:
             latest = weekly[0].get("fields", {})
             oldest = weekly[-1].get("fields", {})
@@ -245,7 +236,6 @@ def generate_weekly_report(
                     }
                 )
 
-    # Summary
     total_changes = sum(s["changes_count"] for s in report_sections)
     most_active = (
         max(report_sections, key=lambda x: x["changes_count"]) if report_sections else None
diff --git a/monitor.py b/monitor.py
index 76306b1..8f65804 100644
--- a/monitor.py
+++ b/monitor.py
@@ -9,33 +9,17 @@ Cron-based monitors that diff content and judge meaningful changes."""
 import difflib
 import json
 import logging
-import os
 import re
-import shutil
 import uuid
 from datetime import UTC, datetime
-from pathlib import Path
 from typing import Any
 
 import httpx
 
-from paths import PRY_DATA_DIR
+from db import Monitor, MonitorRun, session_scope
 
 logger = logging.getLogger(__name__)
 
-MONITORS_DIR = PRY_DATA_DIR / "monitors"
-MONITORS_DIR.mkdir(parents=True, exist_ok=True)
-
-
-def _monitor_path(monitor_id: str) -> Path:
-    return MONITORS_DIR / f"{monitor_id}.json"
-
-
-def _snapshot_path(monitor_id: str, version: int) -> Path:
-    snap_dir = MONITORS_DIR / monitor_id
-    snap_dir.mkdir(parents=True, exist_ok=True)
-    return snap_dir / f"snapshot_{version}.json"
-
 
 def _compute_diff(previous: str, current: str) -> str:
     differ = difflib.Differ()
@@ -45,6 +29,25 @@ def _compute_diff(previous: str, current: str) -> str:
     return f"{added} lines added, {removed} lines removed"
 
 
+def _monitor_to_dict(m: Monitor) -> dict[str, Any]:
+    return {
+        "id": m.monitor_id,
+        "name": m.monitor_name,
+        "target_url": m.url,
+        "schedule_cron": m.schedule_cron,
+        "goal": m.goal or "",
+        "webhook_url": m.webhook_url or "",
+        "use_llm_judge": m.use_llm_judge or False,
+        "status": "active" if m.active else "inactive",
+        "created_at": m.created_at.isoformat() if m.created_at else "",
+        "last_run_at": m.last_run_at.isoformat() if m.last_run_at else None,
+        "current_version": m.current_version or 0,
+        "last_content": m.last_content or "",
+        "total_checks": m.total_checks or 0,
+        "total_changes": m.total_changes or 0,
+    }
+
+
 class ChangeJudger:
     """Judge whether a content change is meaningful using LLM or heuristics."""
 
@@ -89,7 +92,6 @@ class ChangeJudger:
 
         goal_in_current = any(gk in curr_text_lower for gk in goal_keywords)
 
-        # Check if goal keywords appear in newly-added tokens
         def _keyword_match(tokens: set[str]) -> list[str]:
             matched: list[str] = []
             for gk in goal_keywords:
@@ -114,7 +116,6 @@ class ChangeJudger:
                 ],
             }
 
-        # Content changed and goal keywords still present -> meaningful update
         if goal_in_current and len(previous) > 0:
             change_ratio = abs(len(current) - len(previous)) / len(previous)
             if change_ratio > 0.01:
@@ -130,7 +131,6 @@ class ChangeJudger:
                     ],
                 }
 
-        # Large generic change (no goal or goal keywords absent)
         if len(previous) > 0:
             change_ratio = abs(len(current) - len(previous)) / len(previous)
             if change_ratio > 0.1:
@@ -187,7 +187,7 @@ Is this change meaningful given the goal? Respond with JSON:
             response_text = resp.json().get("response", "")
             json_match = re.search(r"\{.*\}", response_text, re.DOTALL)
             if json_match:
-                return json.loads(json_match.group(0))  # type: ignore[no-any-return]
+                return json.loads(json_match.group(0))
         except (json.JSONDecodeError, ValueError):
             logger.warning("llm_judge_failed, falling back to heuristic")
 
@@ -203,78 +203,109 @@ async def create_monitor(
     use_llm_judge: bool = False,
 ) -> dict[str, Any]:
     monitor_id = uuid.uuid4().hex[:12]
-    monitor = {
-        "id": monitor_id,
-        "name": name,
-        "target_url": target_url,
-        "schedule_cron": schedule_cron,
-        "goal": goal,
-        "webhook_url": webhook_url,
-        "use_llm_judge": use_llm_judge,
-        "status": "active",
-        "created_at": datetime.now(UTC).isoformat(),
-        "last_run_at": None,
-        "current_version": 0,
-        "last_content": "",
-        "total_checks": 0,
-        "total_changes": 0,
-    }
-    path = _monitor_path(monitor_id)
-    path.write_text(json.dumps(monitor, indent=2))
-    logger.info("monitor_created", extra={"monitor_id": monitor_id, "monitor_name": name})
-    return monitor
+    now = datetime.now(UTC)
+    monitor = Monitor(
+        monitor_id=monitor_id,
+        url=target_url,
+        monitor_name=name,
+        schedule_cron=schedule_cron,
+        goal=goal,
+        use_llm_judge=use_llm_judge,
+        active=True,
+        webhook_url=webhook_url,
+        created_at=now,
+        current_version=0,
+        last_content="",
+        total_checks=0,
+        total_changes=0,
+    )
+    try:
+        with session_scope() as s:
+            s.add(monitor)
+        logger.info("monitor_created", extra={"monitor_id": monitor_id, "monitor_name": name})
+        return _monitor_to_dict(monitor)
+    except Exception:
+        logger.exception("monitor_creation_failed", extra={"monitor_name": name})
+        raise
 
 
 async def run_monitor(monitor_id: str) -> dict[str, Any]:
-    path = _monitor_path(monitor_id)
-    if not path.exists():
-        return {"error": f"Monitor not found: {monitor_id}"}
-
-    monitor = json.loads(path.read_text())
-    previous_content = monitor.get("last_content", "")
-
     try:
-        from scraper import PryScraper
+        with session_scope() as s:
+            monitor = s.query(Monitor).filter(Monitor.monitor_id == monitor_id).first()
+            if not monitor:
+                return {"error": f"Monitor not found: {monitor_id}"}
 
-        scraper = PryScraper()
-        result = await scraper.scrape(monitor["target_url"])
-        current_content = result.get("content", "")
-    except (json.JSONDecodeError, ValueError) as e:
-        logger.exception("monitor_scrape_failed", extra={"monitor_id": monitor_id})
-        return {"error": f"Scrape failed: {e!s}"}
+            previous_content = monitor.last_content or ""
 
-    judger = ChangeJudger(use_llm=monitor.get("use_llm_judge", False))
-    judgment = await judger.judge(previous_content, current_content, monitor.get("goal", ""))
+            try:
+                from scraper import PryScraper
 
-    monitor["last_run_at"] = datetime.now(UTC).isoformat()
-    monitor["total_checks"] += 1
-    if previous_content and current_content != previous_content:
-        monitor["total_changes"] += 1
-        monitor["current_version"] += 1
-        snap = {
-            "version": monitor["current_version"],
-            "content": current_content,
-            "detected_at": monitor["last_run_at"],
-        }
-        _snapshot_path(monitor_id, monitor["current_version"]).write_text(json.dumps(snap))
-    monitor["last_content"] = current_content
-    path.write_text(json.dumps(monitor, indent=2))
+                scraper = PryScraper()
+                result = await scraper.scrape(monitor.url)
+                current_content = result.get("content", "")
+            except Exception as e:
+                logger.exception("monitor_scrape_failed", extra={"monitor_id": monitor_id})
+                run = MonitorRun(
+                    monitor_id=monitor_id,
+                    ts=datetime.now(UTC),
+                    status="error",
+                    error=str(e),
+                )
+                s.add(run)
+                return {"error": f"Scrape failed: {e!s}"}
 
-    result = {
-        "monitor_id": monitor_id,
-        "name": monitor["name"],
-        "changed": bool(previous_content) and current_content != previous_content,
-        "previous_content_length": len(previous_content),
-        "current_content_length": len(current_content),
-        "judgment": judgment,
-        "total_checks": monitor["total_checks"],
-        "total_changes": monitor["total_changes"],
-    }
+            judger = ChangeJudger(use_llm=monitor.use_llm_judge)
+            judgment = await judger.judge(previous_content, current_content, monitor.goal or "")
 
-    if judgment["meaningful"] and monitor.get("webhook_url"):
-        await _fire_webhook(monitor["webhook_url"], result)
+            changed = bool(previous_content) and current_content != previous_content
+            now = datetime.now(UTC)
+            monitor.last_run_at = now
+            monitor.total_checks = (monitor.total_checks or 0) + 1
 
-    return result
+            run_status = "success"
+            diff_data = {}
+            if changed:
+                monitor.total_changes = (monitor.total_changes or 0) + 1
+                monitor.current_version = (monitor.current_version or 0) + 1
+                run_status = "changed"
+                diff_data = {
+                    "previous_length": len(previous_content),
+                    "current_length": len(current_content),
+                    "diff": _compute_diff(previous_content, current_content),
+                }
+
+            monitor.last_content = current_content
+
+            run = MonitorRun(
+                monitor_id=monitor_id,
+                ts=now,
+                status=run_status,
+                diff=diff_data,
+            )
+            s.add(run)
+
+            result = {
+                "monitor_id": monitor_id,
+                "name": monitor.monitor_name,
+                "changed": changed,
+                "previous_content_length": len(previous_content),
+                "current_content_length": len(current_content),
+                "judgment": judgment,
+                "total_checks": monitor.total_checks,
+                "total_changes": monitor.total_changes,
+            }
+
+            webhook_url = monitor.webhook_url
+            meaningful = judgment["meaningful"]
+
+        if meaningful and webhook_url:
+            await _fire_webhook(webhook_url, result)
+
+        return result
+    except Exception:
+        logger.exception("monitor_run_failed", extra={"monitor_id": monitor_id})
+        return {"error": f"Monitor run failed: {monitor_id}"}
 
 
 async def _fire_webhook(webhook_url: str, payload: dict[str, Any]) -> None:
@@ -289,22 +320,29 @@ async def _fire_webhook(webhook_url: str, payload: dict[str, Any]) -> None:
 
 
 async def list_monitors() -> list[dict[str, Any]]:
-    monitors = []
-    for path in sorted(MONITORS_DIR.glob("*.json"), key=os.path.getmtime, reverse=True):
-        try:
-            data = json.loads(path.read_text())
-            monitors.append(data)
-        except (json.JSONDecodeError, OSError):
-            continue
-    return monitors
+    try:
+        with session_scope() as s:
+            monitors = (
+                s.query(Monitor)
+                .filter(Monitor.active.is_(True))
+                .order_by(Monitor.created_at.desc())
+                .all()
+            )
+            return [_monitor_to_dict(m) for m in monitors]
+    except Exception:
+        logger.exception("monitor_list_failed")
+        return []
 
 
 async def delete_monitor(monitor_id: str) -> bool:
-    path = _monitor_path(monitor_id)
-    snap_dir = MONITORS_DIR / monitor_id
-    if snap_dir.exists():
-        shutil.rmtree(snap_dir)
-    if path.exists():
-        path.unlink()
+    try:
+        with session_scope() as s:
+            monitor = s.query(Monitor).filter(Monitor.monitor_id == monitor_id).first()
+            if not monitor:
+                return False
+            s.delete(monitor)
+        logger.info("monitor_deleted", extra={"monitor_id": monitor_id})
         return True
-    return False
+    except Exception:
+        logger.exception("monitor_delete_failed", extra={"monitor_id": monitor_id})
+        return False
diff --git a/referrals.py b/referrals.py
index 4aa8724..d0d3d2f 100644
--- a/referrals.py
+++ b/referrals.py
@@ -7,19 +7,18 @@ Supports x402 (HTTP 402) pay-per-scrape protocol for monetization."""
 #
 # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
 # Licensed under MIT. See LICENSE.
-import json
 import logging
 import uuid
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from paths import PRY_DATA_DIR
+from sqlalchemy import func
+from sqlalchemy.exc import SQLAlchemyError
+
+from db import ReferralClick, ReferralConversion, ReferralInAppUsage, session_scope
 
 logger = logging.getLogger(__name__)
 
-REFERRAL_DIR = PRY_DATA_DIR / "referrals"
-REFERRAL_DIR.mkdir(parents=True, exist_ok=True)
-
 # Provider catalog with referral program details
 # Format: { category: [ {name, url, commission, cookie_days, ...} ] }
 PROVIDER_CATALOG: dict[str, list[dict[str, Any]]] = {
@@ -134,7 +133,7 @@ PROVIDER_CATALOG: dict[str, list[dict[str, Any]]] = {
         {
             "name": "Hetzner",
             "url": "https://hetzner.com/?ref=pry",
-            "commission": "€25/signup + 10% recurring",
+            "commission": "\u20ac25/signup + 10% recurring",
             "cookie_days": 90,
             "tag": "hetzner",
         },
@@ -176,7 +175,7 @@ PROVIDER_CATALOG: dict[str, list[dict[str, Any]]] = {
         {
             "name": "Netcup",
             "url": "https://netcup.com/?ref=pry",
-            "commission": "€5/signup",
+            "commission": "\u20ac5/signup",
             "cookie_days": 90,
             "tag": "netcup",
         },
@@ -588,71 +587,68 @@ PROVIDER_CATALOG: dict[str, list[dict[str, Any]]] = {
 class ReferralTracker:
     """Track referral clicks, conversions, and revenue."""
 
-    def __init__(self):
-        self.clicks_file = REFERRAL_DIR / "clicks.jsonl"
-        self.conversions_file = REFERRAL_DIR / "conversions.jsonl"
-        self._load_state()
-
-    def _load_state(self) -> None:
-        self.clicks: list[dict[str, Any]] = []
-        self.conversions: list[dict[str, Any]] = []
-        for path in (self.clicks_file, self.conversions_file):
-            if path.exists():
-                try:
-                    for line in path.read_text().splitlines():
-                        if line.strip():
-                            record = json.loads(line)
-                            (self.clicks if path == self.clicks_file else self.conversions).append(
-                                record
-                            )
-                except (json.JSONDecodeError, OSError):
-                    pass
+    def __init__(self) -> None:
+        pass
 
     def record_click(
         self, provider_tag: str, url: str, source: str = "pry", user_id: str = ""
     ) -> str:
         """Record a referral link click. Returns the tracking ID."""
         click_id = uuid.uuid4().hex[:12]
-        record = {
-            "id": click_id,
-            "provider": provider_tag,
-            "url": url,
-            "source": source,
-            "user_id": user_id,
-            "timestamp": datetime.now(UTC).isoformat(),
-            "converted": False,
-        }
         try:
-            with open(self.clicks_file, "a") as f:
-                f.write(json.dumps(record) + "\n")
-            self.clicks.append(record)
-        except OSError:
-            pass
+            with session_scope() as s:
+                s.add(
+                    ReferralClick(
+                        click_id=click_id,
+                        provider=provider_tag,
+                        url=url,
+                        source=source,
+                        user_id=user_id,
+                        converted=False,
+                        conversion_value_usd=0.0,
+                    )
+                )
+        except SQLAlchemyError as e:
+            logger.error(
+                "referral_click_failed",
+                extra={"click_id": click_id, "provider": provider_tag, "error": str(e)[:200]},
+            )
+            raise
         logger.info("referral_click", extra={"provider": provider_tag, "source": source})
         return click_id
 
     def record_conversion(self, click_id: str, revenue_usd: float = 0.0, notes: str = "") -> bool:
-        """Record a conversion for a click."""
-        for click in self.clicks:
-            if click["id"] == click_id:
-                click["converted"] = True
-                click["revenue_usd"] = revenue_usd
-                click["conversion_notes"] = notes
-                conversion = {
-                    "click_id": click_id,
-                    "provider": click["provider"],
-                    "revenue_usd": revenue_usd,
-                    "notes": notes,
-                    "timestamp": datetime.now(UTC).isoformat(),
-                }
-                try:
-                    with open(self.conversions_file, "a") as f:
-                        f.write(json.dumps(conversion) + "\n")
-                    self.conversions.append(conversion)
-                except OSError:
-                    pass
-                return True
-        return False
+        """Record a conversion for a click. Returns True if successful."""
+        try:
+            with session_scope() as s:
+                click = s.query(ReferralClick).filter_by(click_id=click_id).first()
+                if click is None:
+                    logger.warning(
+                        "referral_conversion_click_not_found",
+                        extra={"click_id": click_id},
+                    )
+                    return False
+                click.converted = True
+                click.conversion_value_usd = revenue_usd
+                s.add(
+                    ReferralConversion(
+                        click_id=click_id,
+                        provider=click.provider,
+                        revenue_usd=revenue_usd,
+                        notes=notes,
+                    )
+                )
+        except SQLAlchemyError as e:
+            logger.error(
+                "referral_conversion_failed",
+                extra={"click_id": click_id, "error": str(e)[:200]},
+            )
+            raise
+        logger.info(
+            "referral_conversion",
+            extra={"click_id": click_id, "revenue_usd": revenue_usd},
+        )
+        return True
 
     def get_catalog(self, category: str = "") -> list[dict[str, Any]]:
         """Get the provider catalog, optionally filtered by category."""
@@ -665,53 +661,77 @@ class ReferralTracker:
     def get_stats(self, days_back: int = 30) -> dict[str, Any]:
         """Get referral statistics for the last N days."""
         cutoff = datetime.now(UTC) - timedelta(days=days_back)
-        recent_clicks = []
-        for c in self.clicks:
-            try:
-                ts = datetime.fromisoformat(c["timestamp"])
-                if ts >= cutoff:
-                    recent_clicks.append(c)
-            except (ValueError, TypeError):
-                pass
-        recent_conversions = []
-        for c in self.conversions:
-            try:
-                ts = datetime.fromisoformat(c.get("timestamp", ""))
-                if ts >= cutoff:
-                    recent_conversions.append(c)
-            except (ValueError, TypeError):
-                pass
-        clicks_by_provider: dict[str, int] = {}
-        for c in recent_clicks:
-            clicks_by_provider[c["provider"]] = clicks_by_provider.get(c["provider"], 0) + 1
-        conversions_by_provider: dict[str, int] = {}
-        revenue_by_provider: dict[str, float] = {}
-        for c in recent_conversions:
-            p = c["provider"]
-            conversions_by_provider[p] = conversions_by_provider.get(p, 0) + 1
-            revenue_by_provider[p] = revenue_by_provider.get(p, 0.0) + c.get("revenue_usd", 0.0)
-        return {
-            "period_days": days_back,
-            "total_clicks": len(recent_clicks),
-            "total_conversions": len(recent_conversions),
-            "clicks_by_provider": clicks_by_provider,
-            "conversions_by_provider": conversions_by_provider,
-            "revenue_by_provider_usd": revenue_by_provider,
-            "total_revenue_usd": sum(revenue_by_provider.values()),
-            "conversion_rate": len(recent_conversions) / max(len(recent_clicks), 1),
-            "catalog_size": sum(len(v) for v in PROVIDER_CATALOG.values()),
-        }
+        try:
+            with session_scope() as s:
+                total_clicks = (
+                    s.query(func.count(ReferralClick.id))
+                    .filter(ReferralClick.ts >= cutoff)
+                    .scalar()
+                    or 0
+                )
+
+                click_rows = (
+                    s.query(
+                        ReferralClick.provider,
+                        func.count(ReferralClick.id).label("cnt"),
+                    )
+                    .filter(ReferralClick.ts >= cutoff)
+                    .group_by(ReferralClick.provider)
+                    .all()
+                )
+                clicks_by_provider = {r.provider: r.cnt for r in click_rows}
+
+                conv_rows = (
+                    s.query(
+                        ReferralConversion.provider,
+                        func.count(ReferralConversion.id).label("cnt"),
+                        func.sum(ReferralConversion.revenue_usd).label("rev"),
+                    )
+                    .filter(ReferralConversion.ts >= cutoff)
+                    .group_by(ReferralConversion.provider)
+                    .all()
+                )
+                conversions_by_provider = {r.provider: r.cnt for r in conv_rows}
+                revenue_by_provider = {r.provider: float(r.rev or 0.0) for r in conv_rows}
+
+                total_conversions = sum(conversions_by_provider.values())
+                total_revenue = sum(revenue_by_provider.values())
+
+                return {
+                    "period_days": days_back,
+                    "total_clicks": total_clicks,
+                    "total_conversions": total_conversions,
+                    "clicks_by_provider": clicks_by_provider,
+                    "conversions_by_provider": conversions_by_provider,
+                    "revenue_by_provider_usd": revenue_by_provider,
+                    "total_revenue_usd": total_revenue,
+                    "conversion_rate": total_conversions / max(total_clicks, 1),
+                    "catalog_size": sum(len(v) for v in PROVIDER_CATALOG.values()),
+                }
+        except SQLAlchemyError as e:
+            logger.error(
+                "referral_stats_failed",
+                extra={"error": str(e)[:200]},
+            )
+            raise
 
     def track_in_app_usage(self, provider_tag: str, revenue_usd: float) -> None:
         """Track revenue from in-app LLM provider usage (per-call)."""
-        record = {
-            "type": "in_app_usage",
-            "provider": provider_tag,
-            "revenue_usd": revenue_usd,
-            "timestamp": datetime.now(UTC).isoformat(),
-        }
         try:
-            with open(REFERRAL_DIR / "in_app_usage.jsonl", "a") as f:
-                f.write(json.dumps(record) + "\n")
-        except OSError:
-            pass
+            with session_scope() as s:
+                s.add(
+                    ReferralInAppUsage(
+                        provider=provider_tag,
+                        revenue_usd=revenue_usd,
+                    )
+                )
+        except SQLAlchemyError as e:
+            logger.error(
+                "referral_in_app_usage_failed",
+                extra={"provider": provider_tag, "error": str(e)[:200]},
+            )
+            raise
+        logger.info(
+            "referral_in_app_usage",
+            extra={"provider": provider_tag, "revenue_usd": revenue_usd},
+        )
diff --git a/structure_monitor.py b/structure_monitor.py
index 4d370a6..b6120c3 100644
--- a/structure_monitor.py
+++ b/structure_monitor.py
@@ -7,9 +7,7 @@ Track CSS selector validity over time. Alert before scrapers break."""
 # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
 # Licensed under MIT. See LICENSE.
 import hashlib
-import json
 import logging
-from contextlib import suppress
 from datetime import UTC, datetime
 from typing import Any
 
@@ -17,13 +15,10 @@ import httpx
 from lxml import html as lxml_html
 
 from client import get_client
-from paths import PRY_DATA_DIR
+from db import StructureSnapshot, session_scope
 
 logger = logging.getLogger(__name__)
 
-STRUCTURE_DIR = PRY_DATA_DIR / "structure"
-STRUCTURE_DIR.mkdir(parents=True, exist_ok=True)
-
 
 async def check_selectors(
     url: str,
@@ -46,7 +41,6 @@ async def check_selectors(
         "failed_count": 0,
     }
 
-    # Fetch the page
     try:
         client = await get_client()
         resp = await client.get(
@@ -65,10 +59,9 @@ async def check_selectors(
 
     try:
         tree = lxml_html.fromstring(html_content)
-    except (httpx.HTTPError, httpx.RequestError):
+    except Exception:  # noqa: BLE001
         return {"url": url, "error": "Failed to parse HTML", "all_matched": False}
 
-    # Check each selector
     for sel in selectors:
         selector_str = sel.get("selector", "")
         name = sel.get("name", selector_str[:30])
@@ -94,7 +87,6 @@ async def check_selectors(
                 "status": "ok" if is_matched else "broken",
             }
 
-            # If matched, sample content for change detection
             if is_matched and elements:
                 sample_text = " ".join(e.text_content().strip()[:100] for e in elements[:3])
                 selector_result["sample"] = sample_text[:200]
@@ -105,7 +97,7 @@ async def check_selectors(
             else:
                 result["failed_count"] += 1
 
-        except (httpx.HTTPError, httpx.RequestError) as e:
+        except Exception as e:  # noqa: BLE001
             result["selectors"].append(
                 {
                     "name": name,
@@ -133,80 +125,112 @@ async def monitor_page_structure(
     Returns the current check result plus any detected changes since last check.
     """
     url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
-    # Use template_id if provided for the filename, else url_hash
-    file_key = template_id or url_hash
-    history_path = STRUCTURE_DIR / f"structure_{file_key}.json"
 
-    # Run current check
     current = await check_selectors(url, selectors)
     if "error" in current:
         return current
 
-    # Load previous state
-    previous: dict[str, Any] | None = None
-    if history_path.exists():
-        with suppress(json.JSONDecodeError, OSError):
-            previous = json.loads(history_path.read_text())
-
-    # Detect changes
     changes = []
-    if previous and "selectors" in previous:
-        prev_map = {s["name"]: s for s in previous["selectors"]}
-        curr_map = {s["name"]: s for s in current["selectors"]}
 
-        for name, curr_sel in curr_map.items():
-            prev_sel = prev_map.get(name)
-            if prev_sel:
-                # Check status change
-                if prev_sel.get("matched") and not curr_sel.get("matched"):
-                    changes.append(
-                        {
-                            "type": "selector_broken",
-                            "severity": "critical",
-                            "name": name,
-                            "selector": curr_sel.get("selector", ""),
-                            "message": f"CSS selector '{name}' stopped matching (was matching before)",
-                        }
-                    )
-                elif not prev_sel.get("matched") and curr_sel.get("matched"):
-                    changes.append(
-                        {
-                            "type": "selector_fixed",
-                            "severity": "info",
-                            "name": name,
-                            "message": f"CSS selector '{name}' started matching again",
-                        }
-                    )
-                # Check count change
-                elif curr_sel.get("match_count") != prev_sel.get("match_count"):
-                    changes.append(
-                        {
-                            "type": "count_changed",
-                            "severity": "medium",
-                            "name": name,
-                            "message": f"Selector '{name}' matched {prev_sel.get('match_count')} elements before, now {curr_sel.get('match_count')}",
-                            "previous_count": prev_sel.get("match_count"),
-                            "current_count": curr_sel.get("match_count"),
-                        }
-                    )
+    try:
+        with session_scope() as s:
+            if template_id:
+                previous_snap = (
+                    s.query(StructureSnapshot)
+                    .filter(StructureSnapshot.template_id == template_id)
+                    .order_by(StructureSnapshot.checked_at.desc())
+                    .first()
+                )
+            else:
+                previous_snap = (
+                    s.query(StructureSnapshot)
+                    .filter(StructureSnapshot.url_hash == url_hash)
+                    .order_by(StructureSnapshot.checked_at.desc())
+                    .first()
+                )
 
-    # Detect new failures
-    if previous and previous.get("all_matched") and not current.get("all_matched"):
-        changes.append(
-            {
-                "type": "structure_broken",
-                "severity": "critical",
-                "name": "page_structure",
-                "message": f"Page structure changed — {current.get('failed_count')} selectors now fail",
-                "failed_selectors": [
-                    s["name"] for s in current.get("selectors", []) if not s.get("matched")
-                ],
-            }
-        )
+            if previous_snap:
+                previous_selectors = previous_snap.selectors or []
+                previous_all_matched = previous_snap.all_matched
+            else:
+                previous_selectors = []
+                previous_all_matched = True
 
-    # Save current state
-    with suppress(OSError):
-        history_path.write_text(json.dumps(current, indent=2))
+            prev_map = {s["name"]: s for s in previous_selectors if isinstance(s, dict)}
+            curr_map = {s["name"]: s for s in current.get("selectors", []) if isinstance(s, dict)}
+
+            for name, curr_sel in curr_map.items():
+                prev_sel = prev_map.get(name)
+                if prev_sel:
+                    if prev_sel.get("matched") and not curr_sel.get("matched"):
+                        changes.append(
+                            {
+                                "type": "selector_broken",
+                                "severity": "critical",
+                                "name": name,
+                                "selector": curr_sel.get("selector", ""),
+                                "message": f"CSS selector '{name}' stopped matching (was matching before)",
+                            }
+                        )
+                    elif not prev_sel.get("matched") and curr_sel.get("matched"):
+                        changes.append(
+                            {
+                                "type": "selector_fixed",
+                                "severity": "info",
+                                "name": name,
+                                "message": f"CSS selector '{name}' started matching again",
+                            }
+                        )
+                    elif curr_sel.get("match_count") != prev_sel.get("match_count"):
+                        changes.append(
+                            {
+                                "type": "count_changed",
+                                "severity": "medium",
+                                "name": name,
+                                "message": f"Selector '{name}' matched {prev_sel.get('match_count')} elements before, now {curr_sel.get('match_count')}",
+                                "previous_count": prev_sel.get("match_count"),
+                                "current_count": curr_sel.get("match_count"),
+                            }
+                        )
+
+            if previous_selectors and previous_all_matched and not current.get("all_matched"):
+                changes.append(
+                    {
+                        "type": "structure_broken",
+                        "severity": "critical",
+                        "name": "page_structure",
+                        "message": f"Page structure changed — {current.get('failed_count')} selectors now fail",
+                        "failed_selectors": [
+                            s["name"] for s in current.get("selectors", []) if not s.get("matched")
+                        ],
+                    }
+                )
+
+            now = datetime.now(UTC)
+            snapshot = StructureSnapshot(
+                url=url,
+                url_hash=url_hash,
+                template_id=template_id,
+                selectors=current.get("selectors", []),
+                all_matched=current.get("all_matched", False),
+                matched_count=current.get("matched_count", 0),
+                failed_count=current.get("failed_count", 0),
+                changes=changes,
+                change_count=len(changes),
+                has_changes=len(changes) > 0,
+                dom_summary={},
+                checked_at=now,
+            )
+            s.add(snapshot)
+
+    except Exception:
+        logger.exception("structure_monitor_save_failed", extra={"url": url})
+        current["changes"] = changes
+        current["change_count"] = len(changes)
+        current["has_changes"] = len(changes) > 0
+        current["template_id"] = template_id
+        current["url_hash"] = url_hash
+        return current
 
     current["changes"] = changes
     current["change_count"] = len(changes)
@@ -217,23 +241,36 @@ async def monitor_page_structure(
     return current
 
 
-def get_structure_history(url: str, template_id: str = "") -> dict[str, Any]:
+async def get_structure_history(url: str, template_id: str = "") -> dict[str, Any]:
     """Get structure check history for a URL."""
-    url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
-    file_key = template_id or url_hash
-    history_path = STRUCTURE_DIR / f"structure_{file_key}.json"
-
-    if not history_path.exists():
-        return {"url": url, "has_history": False}
-
     try:
-        data = json.loads(history_path.read_text())
-        return {
-            "url": url,
-            "has_history": True,
-            "last_checked": data.get("checked_at", ""),
-            "last_all_matched": data.get("all_matched", True),
-            "last_selectors": data.get("selectors", []),
-        }
-    except (json.JSONDecodeError, OSError):
+        with session_scope() as s:
+            if template_id:
+                snap = (
+                    s.query(StructureSnapshot)
+                    .filter(StructureSnapshot.template_id == template_id)
+                    .order_by(StructureSnapshot.checked_at.desc())
+                    .first()
+                )
+            else:
+                url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
+                snap = (
+                    s.query(StructureSnapshot)
+                    .filter(StructureSnapshot.url_hash == url_hash)
+                    .order_by(StructureSnapshot.checked_at.desc())
+                    .first()
+                )
+
+            if not snap:
+                return {"url": url, "has_history": False}
+
+            return {
+                "url": url,
+                "has_history": True,
+                "last_checked": snap.checked_at.isoformat() if snap.checked_at else "",
+                "last_all_matched": snap.all_matched,
+                "last_selectors": snap.selectors or [],
+            }
+    except Exception:
+        logger.exception("structure_history_fetch_failed", extra={"url": url})
         return {"url": url, "has_history": False}
diff --git a/tests/test_intelligence.py b/tests/test_intelligence.py
index d69f89d..b14c5b0 100644
--- a/tests/test_intelligence.py
+++ b/tests/test_intelligence.py
@@ -5,8 +5,15 @@
 # Licensed under MIT. See LICENSE.
 """Tests for competitive intelligence engine."""
 
+import time
+from collections.abc import Iterator
+
+import pytest
+
+from db import Base, get_engine
 from intelligence import (
     compute_field_statistics,
+    detect_anomalies_numeric,
     generate_alert,
     generate_weekly_report,
     get_snapshots,
@@ -14,12 +21,52 @@ from intelligence import (
 )
 
 
+@pytest.fixture(autouse=True)
+def _intel_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """Use in-memory SQLite for each test."""
+    monkeypatch.setenv("PRY_DATABASE_URL", "sqlite://")
+    import db as db_module
+
+    db_module._engine = None
+    db_module._SessionLocal = None
+    engine = get_engine()
+    Base.metadata.create_all(engine)
+    yield
+    Base.metadata.drop_all(engine)
+
+
+# -- Snapshot recording --
+
+
 def test_record_snapshot() -> None:
     result = record_snapshot(
-        "comp_1", "Competitor A", "https://example.com", {"price": 99.99, "stock": "in_stock"}
+        "comp_1",
+        "Competitor A",
+        "https://example.com",
+        {"price": 99.99, "stock": "in_stock"},
     )
     assert "ts" in result
+    assert "unix_ts" in result
     assert result["competitor_name"] == "Competitor A"
+    assert result["competitor_id"] == "comp_1"
+    assert result["fields"]["price"] == 99.99
+
+
+def test_record_snapshot_returns_iso_ts() -> None:
+    result = record_snapshot("comp_ts", "TS Co", "https://example.com", {"x": 1})
+    ts = result["ts"]
+    assert isinstance(ts, str)
+    assert ts.endswith("Z") or "+" in ts
+
+
+def test_record_snapshot_unix_ts() -> None:
+    before = time.time()
+    result = record_snapshot("comp_ut", "UT Co", "https://example.com", {"x": 1})
+    after = time.time()
+    assert before <= result["unix_ts"] <= after + 1
+
+
+# -- Retreiving snapshots --
 
 
 def test_get_snapshots() -> None:
@@ -27,39 +74,179 @@ def test_get_snapshots() -> None:
     snapshots = get_snapshots("comp_test")
     assert len(snapshots) >= 1
     assert snapshots[0]["competitor_name"] == "Test"
+    assert snapshots[0]["fields"]["a"] == 1
 
 
-def test_compute_field_statistics_numeric() -> None:
-    snapshots = [
-        {"fields": {"price": 100}},
-        {"fields": {"price": 110}},
-        {"fields": {"price": 105}},
-    ]
-    stats = compute_field_statistics(snapshots, "price")
-    assert stats["has_history"] is True
-    assert stats["mean"] == 105.0
-    assert stats["min"] == 100
-    assert stats["max"] == 110
+def test_get_snapshots_limit() -> None:
+    for i in range(10):
+        record_snapshot("comp_limit", "Limit Co", "https://example.com", {"i": i})
+    snapshots = get_snapshots("comp_limit", limit=3)
+    assert len(snapshots) == 3
 
 
-def test_generate_alert_price_change() -> None:
-    alert = generate_alert("Competitor A", "price", 100, 120)
-    assert "Competitor A" in alert
-    assert "increased" in alert or "changed" in alert
+def test_get_snapshots_ordered_by_ts_desc() -> None:
+    for i in range(5):
+        record_snapshot("comp_order", "Order Co", "https://example.com", {"i": i})
+        time.sleep(0.01)
+    snapshots = get_snapshots("comp_order")
+    assert len(snapshots) == 5
+    assert snapshots[0]["fields"]["i"] == 4
 
 
-def test_generate_alert_decrease() -> None:
-    alert = generate_alert("Competitor B", "price", 200, 150)
-    assert "decreased" in alert
+def test_get_snapshots_since_hours() -> None:
+    record_snapshot("comp_since", "Since Co", "https://example.com", {"val": 1})
+    snapshots = get_snapshots("comp_since", since_hours=1)
+    assert len(snapshots) >= 1
 
 
-def test_generate_alert_text_change() -> None:
-    alert = generate_alert("Competitor C", "description", "Old desc", "New and improved")
-    assert "Competitor C" in alert
+def test_get_snapshots_empty_for_unknown() -> None:
+    snapshots = get_snapshots("nonexistent_comp")
+    assert snapshots == []
 
 
-def test_generate_report() -> None:
-    record_snapshot("comp_rpt", "Report Co", "https://example.com", {"price": 50})
-    report = generate_weekly_report([{"id": "comp_rpt", "name": "Report Co"}], days_back=7)
-    assert "report_period" in report
-    assert "generated_at" in report
+# -- Field statistics --
+
+
+class TestComputeFieldStatistics:
+    def test_numeric(self) -> None:
+        snapshots = [
+            {"fields": {"price": 100}},
+            {"fields": {"price": 110}},
+            {"fields": {"price": 105}},
+        ]
+        stats = compute_field_statistics(snapshots, "price")
+        assert stats["has_history"] is True
+        assert stats["mean"] == 105.0
+        assert stats["min"] == 100
+        assert stats["max"] == 110
+
+    def test_insufficient_data(self) -> None:
+        snapshots = [{"fields": {"price": 100}}]
+        stats = compute_field_statistics(snapshots, "price")
+        assert stats["has_history"] is False
+
+    def test_empty_snapshots(self) -> None:
+        stats = compute_field_statistics([], "price")
+        assert stats["count"] == 0
+        assert stats["has_history"] is False
+
+    def test_missing_field(self) -> None:
+        snapshots = [{"fields": {"price": 100}}, {"fields": {"other": 200}}]
+        stats = compute_field_statistics(snapshots, "price")
+        assert stats["count"] == 1
+        assert stats["has_history"] is False
+
+    def test_string_field(self) -> None:
+        snapshots = [
+            {"fields": {"status": "active"}},
+            {"fields": {"status": "active"}},
+            {"fields": {"status": "paused"}},
+        ]
+        stats = compute_field_statistics(snapshots, "status")
+        assert stats["has_history"] is True
+        assert stats["unique_values"] == 2
+        assert stats["latest"] == "paused"
+
+
+# -- Anomaly detection --
+
+
+class TestDetectAnomaliesNumeric:
+    def test_detects_anomaly(self) -> None:
+        history = [100, 102, 101, 103, 99, 101, 102]
+        result = detect_anomalies_numeric(150, history)
+        assert result["anomaly"] is True
+        assert result["z_score"] >= 2.0
+
+    def test_insufficient_history(self) -> None:
+        result = detect_anomalies_numeric(100, [])
+        assert result["anomaly"] is False
+        assert "Insufficient history" in result["reason"]
+
+    def test_no_anomaly(self) -> None:
+        history = [100, 101, 100, 101, 100]
+        result = detect_anomalies_numeric(100, history)
+        assert result["anomaly"] is False
+
+    def test_constant_history_same_value(self) -> None:
+        history = [100, 100, 100]
+        result = detect_anomalies_numeric(100, history)
+        assert result["anomaly"] is False
+
+    def test_constant_history_diff_value(self) -> None:
+        history = [100, 100, 100]
+        result = detect_anomalies_numeric(200, history)
+        assert result["anomaly"] is True
+        assert "constant history" in result["reason"]
+
+
+# -- Alerts --
+
+
+class TestGenerateAlert:
+    def test_price_increase(self) -> None:
+        alert = generate_alert("Competitor A", "price", 100, 120)
+        assert "Competitor A" in alert
+        assert "increased" in alert
+
+    def test_price_decrease(self) -> None:
+        alert = generate_alert("Competitor B", "price", 200, 150)
+        assert "decreased" in alert
+
+    def test_text_change_short(self) -> None:
+        alert = generate_alert("Competitor C", "description", "Old desc", "New and improved")
+        assert "Competitor C" in alert
+        assert '"Old desc"' in alert
+
+    def test_text_change_long(self) -> None:
+        long_old = "x" * 60
+        long_new = "y" * 60
+        alert = generate_alert("Competitor D", "description", long_old, long_new)
+        assert "chars" in alert
+
+    def test_with_anomaly_info(self) -> None:
+        alert = generate_alert(
+            "Competitor F",
+            "price",
+            100,
+            200,
+            anomaly_info={"anomaly": True, "z_score": 3.5, "pct_change": 100.0},
+        )
+        assert "ANOMALY DETECTED" in alert
+        assert "z-score" in alert
+
+
+# -- Weekly report --
+
+
+class TestGenerateWeeklyReport:
+    def test_basic_report(self) -> None:
+        record_snapshot("comp_rpt", "Report Co", "https://example.com", {"price": 50})
+        report = generate_weekly_report([{"id": "comp_rpt", "name": "Report Co"}], days_back=7)
+        assert "report_period" in report
+        assert "generated_at" in report
+        assert report["competitors_tracked"] == 1
+
+    def test_report_with_changes(self) -> None:
+        record_snapshot("comp_chg", "Chg Co", "https://example.com", {"price": 100})
+        record_snapshot("comp_chg", "Chg Co", "https://example.com", {"price": 150})
+        report = generate_weekly_report([{"id": "comp_chg", "name": "Chg Co"}], days_back=7)
+        assert report["total_changes"] >= 1
+        assert report["most_active_competitor"] == "Chg Co"
+
+    def test_report_no_activity(self) -> None:
+        report = generate_weekly_report(
+            [{"id": "comp_inactive", "name": "Inactive Co"}],
+            days_back=7,
+        )
+        assert report["total_changes"] == 0
+
+    def test_report_multiple_competitors(self) -> None:
+        record_snapshot("comp_a", "Alpha Co", "https://example.com", {"price": 50})
+        record_snapshot("comp_b", "Beta Co", "https://example.com", {"price": 75})
+        competitors = [
+            {"id": "comp_a", "name": "Alpha Co"},
+            {"id": "comp_b", "name": "Beta Co"},
+        ]
+        report = generate_weekly_report(competitors, days_back=7)
+        assert report["competitors_tracked"] == 2
diff --git a/tests/test_monitor.py b/tests/test_monitor.py
index 6a6febb..e364f53 100644
--- a/tests/test_monitor.py
+++ b/tests/test_monitor.py
@@ -5,7 +5,125 @@
 # Licensed under MIT. See LICENSE.
 """Tests for content monitors."""
 
-from monitor import ChangeJudger
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from monitor import ChangeJudger, create_monitor, delete_monitor, list_monitors, run_monitor
+
+
+@pytest.fixture
+def temp_db(monkeypatch, tmp_path):
+    import db as db_module
+
+    db_path = tmp_path / "test.db"
+    monkeypatch.setenv("PRY_DATABASE_URL", f"sqlite:///{db_path}")
+    monkeypatch.setenv("PRY_DATA_DIR", str(tmp_path))
+    db_module._engine = None
+    db_module._SessionLocal = None
+    db_module.get_engine()
+    yield
+    db_module._engine = None
+    db_module._SessionLocal = None
+
+
+@pytest.mark.asyncio
+async def test_create_monitor(temp_db):
+    result = await create_monitor("test-monitor", "https://example.com")
+    assert result["name"] == "test-monitor"
+    assert result["target_url"] == "https://example.com"
+    assert len(result["id"]) == 12
+    assert result["status"] == "active"
+
+
+@pytest.mark.asyncio
+async def test_create_monitor_with_all_options(temp_db):
+    result = await create_monitor(
+        "full-monitor",
+        "https://example.com",
+        schedule_cron="0 */12 * * *",
+        goal="find price changes",
+        webhook_url="https://hooks.example.com/notify",
+        use_llm_judge=True,
+    )
+    assert result["name"] == "full-monitor"
+    assert result["schedule_cron"] == "0 */12 * * *"
+    assert result["goal"] == "find price changes"
+    assert result["webhook_url"] == "https://hooks.example.com/notify"
+    assert result["use_llm_judge"] is True
+
+
+@pytest.mark.asyncio
+async def test_list_monitors(temp_db):
+    await create_monitor("m1", "https://example.com")
+    await create_monitor("m2", "https://example.org")
+    monitors = await list_monitors()
+    assert len(monitors) == 2
+
+
+@pytest.mark.asyncio
+async def test_list_monitors_empty(temp_db):
+    monitors = await list_monitors()
+    assert monitors == []
+
+
+@pytest.mark.asyncio
+async def test_delete_monitor(temp_db):
+    m = await create_monitor("to-delete", "https://example.com")
+    assert await delete_monitor(m["id"]) is True
+    monitors = await list_monitors()
+    assert len(monitors) == 0
+
+
+@pytest.mark.asyncio
+async def test_delete_nonexistent_monitor(temp_db):
+    assert await delete_monitor("nonexistent") is False
+
+
+@pytest.mark.asyncio
+async def test_run_monitor_with_change(temp_db):
+    m = await create_monitor("change-test", "https://example.com")
+
+    mock_scraper = AsyncMock()
+    mock_scraper.scrape.return_value = {"content": "new content here"}
+
+    with patch("scraper.PryScraper", return_value=mock_scraper):
+        result = await run_monitor(m["id"])
+
+    assert result["monitor_id"] == m["id"]
+    assert result["current_content_length"] > 0
+
+
+@pytest.mark.asyncio
+async def test_run_monitor_no_change(temp_db):
+    m = await create_monitor("no-change", "https://example.com")
+
+    mock_scraper = AsyncMock()
+    mock_scraper.scrape.return_value = {"content": ""}
+
+    with patch("scraper.PryScraper", return_value=mock_scraper):
+        result = await run_monitor(m["id"])
+
+    assert result["changed"] is False
+
+
+@pytest.mark.asyncio
+async def test_run_monitor_not_found(temp_db):
+    result = await run_monitor("nonexistent")
+    assert "error" in result
+
+
+@pytest.mark.asyncio
+async def test_run_monitor_scrape_error(temp_db):
+    m = await create_monitor("scrape-error", "https://example.com")
+
+    mock_scraper = AsyncMock()
+    mock_scraper.scrape.side_effect = ValueError("connection refused")
+
+    with patch("scraper.PryScraper", return_value=mock_scraper):
+        result = await run_monitor(m["id"])
+
+    assert "error" in result
 
 
 async def _judge(previous: str, current: str, goal: str = "") -> dict:
diff --git a/tests/test_referrals.py b/tests/test_referrals.py
index b62ba39..1e4e6aa 100644
--- a/tests/test_referrals.py
+++ b/tests/test_referrals.py
@@ -5,10 +5,28 @@
 # Licensed under MIT. See LICENSE.
 """Tests for referral/affiliate system."""
 
+from collections.abc import Iterator
+
+import pytest
+
 from referrals import PROVIDER_CATALOG, ReferralTracker
 from x402 import X402_PRICING, X402Handler
 
 
+@pytest.fixture(autouse=True)
+def _referral_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """Use in-memory SQLite for all referral tests."""
+    monkeypatch.setenv("PRY_DATABASE_URL", "sqlite://")
+    import db as _db
+
+    _db._engine = None
+    _db._SessionLocal = None
+    _db.get_engine()
+    yield
+    _db._engine = None
+    _db._SessionLocal = None
+
+
 def test_catalog_completeness() -> None:
     """We have a good catalog of providers."""
     total = sum(len(v) for v in PROVIDER_CATALOG.values())
@@ -41,28 +59,102 @@ def test_provider_format() -> None:
 
 
 def test_referral_click() -> None:
+    """Record a click and verify it exists in the database."""
+    from db import ReferralClick, session_scope
+
     rt = ReferralTracker()
     click_id = rt.record_click(
         "openai", "https://platform.openai.com/signup?via=pry", source="test"
     )
     assert click_id is not None
-    assert any(c["id"] == click_id for c in rt.clicks)
+    with session_scope() as s:
+        click = s.query(ReferralClick).filter_by(click_id=click_id).first()
+        assert click is not None
+        assert click.provider == "openai"
+        assert click.source == "test"
+        assert click.converted is False
 
 
 def test_referral_conversion() -> None:
+    """Record a click, convert it, and verify both tables."""
+    from db import ReferralClick, ReferralConversion, session_scope
+
     rt = ReferralTracker()
     click_id = rt.record_click("anthropic", "https://console.anthropic.com/?ref=pry")
     success = rt.record_conversion(click_id, revenue_usd=50.0, notes="Annual plan")
     assert success is True
+    with session_scope() as s:
+        click = s.query(ReferralClick).filter_by(click_id=click_id).first()
+        assert click.converted is True
+        assert click.conversion_value_usd == 50.0
+        conv = s.query(ReferralConversion).filter_by(click_id=click_id).first()
+        assert conv is not None
+        assert conv.revenue_usd == 50.0
+        assert conv.notes == "Annual plan"
+
+
+def test_referral_missing_click_conversion() -> None:
+    """Converting a non-existent click returns False."""
+    rt = ReferralTracker()
+    success = rt.record_conversion("nonexistent_click_id")
+    assert success is False
 
 
 def test_referral_stats() -> None:
+    """Record a click + conversion and verify stats."""
     rt = ReferralTracker()
     click_id = rt.record_click("openai", "https://platform.openai.com/signup?via=pry")
     rt.record_conversion(click_id, revenue_usd=100.0)
     stats = rt.get_stats(days_back=1)
     assert stats["total_clicks"] >= 1
+    assert stats["total_conversions"] >= 1
     assert stats["total_revenue_usd"] >= 100.0
+    assert stats["clicks_by_provider"].get("openai", 0) >= 1
+    assert stats["conversions_by_provider"].get("openai", 0) >= 1
+    assert stats["revenue_by_provider_usd"].get("openai", 0.0) >= 100.0
+    assert 0 < stats["conversion_rate"] <= 1.0
+    assert stats["period_days"] == 1
+    assert stats["catalog_size"] >= 50
+
+
+def test_referral_stats_empty() -> None:
+    """Stats with no data returns zeroes."""
+    rt = ReferralTracker()
+    stats = rt.get_stats(days_back=1)
+    assert stats["total_clicks"] == 0
+    assert stats["total_conversions"] == 0
+    assert stats["total_revenue_usd"] == 0.0
+    assert stats["conversion_rate"] == 0.0
+
+
+def test_referral_in_app_usage() -> None:
+    """Track in-app usage revenue and verify in database."""
+    from db import ReferralInAppUsage, session_scope
+
+    rt = ReferralTracker()
+    rt.track_in_app_usage("openai", 15.0)
+    with session_scope() as s:
+        usage = s.query(ReferralInAppUsage).filter_by(provider="openai").first()
+        assert usage is not None
+        assert usage.revenue_usd == 15.0
+
+
+def test_referral_catalog() -> None:
+    """Verify get_catalog returns correct data."""
+    rt = ReferralTracker()
+    full = rt.get_catalog()
+    assert len(full) > 0
+    for entry in full:
+        assert "category" in entry
+        assert "providers" in entry
+    llm = rt.get_catalog("llm")
+    assert len(llm) > 0
+    assert llm[0]["name"] == "OpenAI"
+    empty = rt.get_catalog("nonexistent")
+    assert empty == []
+
+
+# ── x402 tests (unchanged) ────────────────────────────────────────
 
 
 def test_x402_payment_request() -> None:
@@ -71,7 +163,7 @@ def test_x402_payment_request() -> None:
     assert body["x402Version"] == 1
     assert "accepts" in body
     base_accept = next(a for a in body["accepts"] if a["network"] == "base")
-    assert base_accept["maxAmountRequired"] == "1000"  # $0.001 = 1000 USDC units
+    assert base_accept["maxAmountRequired"] == "1000"
     assert base_accept["asset"] == "USDC"
     assert "PAYMENT-REQUIRED" in headers
 
diff --git a/tests/test_structure_monitor.py b/tests/test_structure_monitor.py
index e83e6cc..a5dbd4e 100644
--- a/tests/test_structure_monitor.py
+++ b/tests/test_structure_monitor.py
@@ -7,6 +7,8 @@
 
 from unittest.mock import AsyncMock, patch
 
+import pytest
+
 from structure_monitor import (
     check_selectors,
     get_structure_history,
@@ -14,26 +16,125 @@ from structure_monitor import (
 )
 
 
-def test_check_selectors_empty() -> None:
-    import asyncio
+@pytest.fixture
+def temp_db(monkeypatch, tmp_path):
+    import db as db_module
 
-    result = asyncio.run(check_selectors("https://example.com", []))
+    db_path = tmp_path / "test.db"
+    monkeypatch.setenv("PRY_DATABASE_URL", f"sqlite:///{db_path}")
+    monkeypatch.setenv("PRY_DATA_DIR", str(tmp_path))
+    db_module._engine = None
+    db_module._SessionLocal = None
+    db_module.get_engine()
+    yield
+    db_module._engine = None
+    db_module._SessionLocal = None
+
+
+@pytest.mark.asyncio
+async def test_check_selectors_empty():
+    result = await check_selectors("https://example.com", [])
     assert "selectors" in result
+    assert result["all_matched"] is True
 
 
-def test_get_structure_history_no_history() -> None:
-    result = get_structure_history("https://nonexistent-page-test-123.com")
+@pytest.mark.asyncio
+async def test_get_structure_history_no_history(temp_db):
+    result = await get_structure_history("https://nonexistent-page-test-123.com")
     assert result["has_history"] is False
 
 
-def test_monitor_no_selectors() -> None:
-    import asyncio
-
+@pytest.mark.asyncio
+async def test_monitor_page_structure_creates_history(temp_db):
     mock_client = AsyncMock()
     mock_client.get.return_value.is_success = True
-    mock_client.get.return_value.text = "<html><body>test</body></html>"
+    mock_client.get.return_value.text = "<html><body><h1>Title</h1></body></html>"
 
     with patch("structure_monitor.get_client", return_value=mock_client):
-        result = asyncio.run(monitor_page_structure("https://example.com", []))
+        result = await monitor_page_structure(
+            "https://example.com",
+            [{"name": "title", "selector": "h1", "type": "css"}],
+        )
+
     assert "selectors" in result
     assert "changes" in result
+    assert result["all_matched"] is True
+    assert result["has_changes"] is False
+
+
+@pytest.mark.asyncio
+async def test_get_structure_history_with_data(temp_db):
+    mock_client = AsyncMock()
+    mock_client.get.return_value.is_success = True
+    mock_client.get.return_value.text = "<html><body><h1>Title</h1></body></html>"
+
+    with patch("structure_monitor.get_client", return_value=mock_client):
+        await monitor_page_structure(
+            "https://example.com",
+            [{"name": "title", "selector": "h1", "type": "css"}],
+        )
+
+    result = await get_structure_history("https://example.com")
+    assert result["has_history"] is True
+    assert "last_checked" in result
+    assert "last_selectors" in result
+
+
+@pytest.mark.asyncio
+async def test_monitor_page_structure_with_template_id(temp_db):
+    mock_client = AsyncMock()
+    mock_client.get.return_value.is_success = True
+    mock_client.get.return_value.text = "<html><body><h1>Title</h1></body></html>"
+
+    with patch("structure_monitor.get_client", return_value=mock_client):
+        result = await monitor_page_structure(
+            "https://example.com",
+            [{"name": "title", "selector": "h1", "type": "css"}],
+            template_id="template-1",
+        )
+
+    assert result["template_id"] == "template-1"
+
+    by_url = await get_structure_history("https://example.com")
+    assert by_url["has_history"] is True
+
+    by_template = await get_structure_history("https://example.com", template_id="template-1")
+    assert by_template["has_history"] is True
+
+
+@pytest.mark.asyncio
+async def test_monitor_page_structure_detects_change(temp_db):
+    mock_client = AsyncMock()
+    mock_client.get.return_value.is_success = True
+    mock_client.get.return_value.text = "<html><body><h1>Title</h1></body></html>"
+
+    with patch("structure_monitor.get_client", return_value=mock_client):
+        await monitor_page_structure(
+            "https://example.com",
+            [{"name": "title", "selector": "h1", "type": "css"}],
+        )
+
+    mock_client.get.return_value.text = "<html><body><h1>New Title</h1></body></html>"
+
+    with patch("structure_monitor.get_client", return_value=mock_client):
+        result = await monitor_page_structure(
+            "https://example.com",
+            [{"name": "title", "selector": "h2", "type": "css"}],
+        )
+
+    assert "changes" in result
+
+
+@pytest.mark.asyncio
+async def test_structure_page_http_error(temp_db):
+    mock_client = AsyncMock()
+    mock_client.get.return_value.is_success = False
+    mock_client.get.return_value.status_code = 404
+
+    with patch("structure_monitor.get_client", return_value=mock_client):
+        result = await monitor_page_structure(
+            "https://example.com/404",
+            [{"name": "title", "selector": "h1", "type": "css"}],
+        )
+
+    assert "error" in result