feat: persist profile-scoped multi-source station provenance (#276)

fluxtuner/core/custom_stations.py

@@ -7,6 +7,7 @@
 from urllib.parse import urlsplit
 
 from fluxtuner.core import db
+from fluxtuner.core import station_provenance as provenance
 from fluxtuner.core.stations import station_metadata
 
 MAX_NAME_LENGTH = 200
@@ -63,6 +64,7 @@ def add_custom_station(
     profile_id: int,
     name: str,
     url: str,
+    record_manual_provenance: bool = False,
 ) -> bool:
     """Add a manual station to one existing profile's favorites.
 
@@ -71,6 +73,8 @@ def add_custom_station(
     belongs to the new profile favorite, so other profiles remain unchanged.
     The caller owns the transaction and must commit successful changes.
     """
+    if type(record_manual_provenance) is not bool:
+        raise ValueError("Invalid manual provenance option.")
     clean_name = _validate_name(name)
     clean_url = _validate_url(url)
     if type(profile_id) is not int or profile_id <= 0:
@@ -124,6 +128,14 @@ def add_custom_station(
             (profile_id, int(row["id"]), clean_name, now, now),
         )
         added = cursor.rowcount > 0
+        if added and record_manual_provenance:
+            source_id = provenance.create_source(conn, profile_id=profile_id, kind="manual_entry")
+            provenance.record_observation(
+                conn,
+                profile_id=profile_id,
+                source_id=source_id,
+                station_id=int(row["id"]),
+            )
     except Exception:
         conn.execute("ROLLBACK TO SAVEPOINT add_custom_station")
         conn.execute("RELEASE SAVEPOINT add_custom_station")

fluxtuner/core/db.py

@@ -20,7 +20,7 @@
 DEFAULT_USER_NAME = "default"
 DEFAULT_PROFILE_NAME = "default"
 SCHEMA_MIGRATION_NAME = "schema_v6"
-CURRENT_SCHEMA_VERSION = 3
+CURRENT_SCHEMA_VERSION = 4
 APPROVAL_APPROVED = "approved"
 APPROVAL_PENDING = "pending"
 APPROVAL_REJECTED = "rejected"
@@ -157,6 +157,12 @@ def migrate_schema(conn: sqlite3.Connection) -> None:
                 conn.execute("PRAGMA user_version = 3")
                 continue
 
+            if stored_version == 3:
+                _create_station_provenance_schema(conn)
+                stored_version = 4
+                conn.execute("PRAGMA user_version = 4")
+                continue
+
             raise RuntimeError(f"Unsupported database schema version: {stored_version}.")
 
         conn.commit()
@@ -428,6 +434,72 @@ def create_schema(conn: sqlite3.Connection) -> None:
         ON playlist_stations(playlist_id, position);
         """,
     )
+    _create_station_provenance_schema(conn)
+
+
+def _create_station_provenance_schema(conn: sqlite3.Connection) -> None:
+    """Create profile-scoped station provenance storage, without committing."""
+    _execute_schema_script(
+        conn,
+        """
+        CREATE TABLE IF NOT EXISTS station_sources (
+            id INTEGER PRIMARY KEY,
+            profile_id INTEGER NOT NULL,
+            kind TEXT NOT NULL,
+            source_ref TEXT NOT NULL,
+            display_name TEXT,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            FOREIGN KEY (profile_id)
+                REFERENCES profiles(id) ON DELETE CASCADE,
+            UNIQUE (profile_id, kind, source_ref),
+            CHECK (
+                kind IN ('local_collection', 'remote_collection', 'manual_entry')
+            ),
+            CHECK (
+                length(source_ref) BETWEEN 1 AND 128
+                AND length(trim(source_ref)) = length(source_ref)
+            ),
+            CHECK (
+                display_name IS NULL
+                OR length(display_name) BETWEEN 1 AND 120
+            )
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_station_sources_profile_kind
+        ON station_sources(profile_id, kind);
+
+        CREATE TABLE IF NOT EXISTS station_source_memberships (
+            source_id INTEGER NOT NULL,
+            station_id INTEGER NOT NULL,
+            first_observed_at TEXT NOT NULL,
+            last_observed_at TEXT NOT NULL,
+            PRIMARY KEY (source_id, station_id),
+            FOREIGN KEY (source_id)
+                REFERENCES station_sources(id) ON DELETE CASCADE,
+            FOREIGN KEY (station_id)
+                REFERENCES stations(id) ON DELETE CASCADE
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_station_source_memberships_station
+        ON station_source_memberships(station_id);
+
+        CREATE TABLE IF NOT EXISTS remote_import_receipts (
+            token_hash TEXT PRIMARY KEY,
+            user_id INTEGER NOT NULL,
+            profile_id INTEGER NOT NULL,
+            session_hash TEXT NOT NULL,
+            payload_hash TEXT NOT NULL,
+            expires_at TEXT NOT NULL,
+            consumed_at TEXT,
+            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
+            FOREIGN KEY (profile_id) REFERENCES profiles(id) ON DELETE CASCADE
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_remote_import_receipts_expires
+        ON remote_import_receipts(expires_at);
+        """,
+    )
 
 
 def _column_names(conn: sqlite3.Connection, table_name: str) -> set[str]:

fluxtuner/core/m3u_import.py

@@ -11,6 +11,7 @@
 from urllib.parse import urlsplit
 
 from fluxtuner.core import db
+from fluxtuner.core import station_provenance as provenance
 from fluxtuner.core.custom_stations import (
     MAX_NAME_LENGTH,
     _has_controls,
@@ -53,6 +54,14 @@ class M3UImportResult:
     already_in_playlist: int
 
 
+@dataclass(frozen=True, slots=True)
+class M3UImportWithSourceResult:
+    """A newly created provenance source and the unchanged import result."""
+
+    result: M3UImportResult
+    source_id: int
+
+
 def _extinf_name(line: str) -> str:
     """Find the title separator without splitting quoted attribute values."""
     quoted = False
@@ -306,6 +315,7 @@ def import_m3u(
     profile_id: int,
     playlist_name: str,
     entries: Sequence[M3UEntry],
+    source_id: int | None = None,
 ) -> M3UImportResult:
     """Import stations into favorites and one playlist without committing.
 
@@ -325,6 +335,8 @@ def import_m3u(
     if conn.execute("SELECT 1 FROM profiles WHERE id = ?", (profile_id,)).fetchone() is None:
         raise ValueError("The selected profile no longer exists.")
     accepted = _validated_entries(entries)
+    if source_id is not None:
+        provenance.get_source(conn, profile_id=profile_id, source_id=source_id)
 
     if not conn.in_transaction:
         conn.execute("BEGIN")
@@ -367,6 +379,14 @@ def import_m3u(
             if station is None:
                 raise RuntimeError("Could not find the imported station.")
 
+            if source_id is not None:
+                provenance.record_observation(
+                    conn,
+                    profile_id=profile_id,
+                    source_id=source_id,
+                    station_id=int(station["id"]),
+                )
+
             # Link the existing row directly: the normal playlist helper
             # upserts station metadata, which imports must preserve.
             cursor = conn.execute(
@@ -393,3 +413,42 @@ def import_m3u(
         raise
     conn.execute("RELEASE SAVEPOINT m3u_import")
     return result
+
+
+def import_m3u_with_new_source(
+    conn: sqlite3.Connection,
+    *,
+    profile_id: int,
+    playlist_name: str,
+    entries: Sequence[M3UEntry],
+    display_name: str | None = None,
+    kind: str = "local_collection",
+) -> M3UImportWithSourceResult:
+    """Atomically create a collection source and import its stations.
+
+    The caller controls commit/rollback. Remote kind is reserved for
+    callers that have obtained a validated remote collection snapshot.
+    """
+    if not conn.in_transaction:
+        conn.execute("BEGIN")
+    conn.execute("SAVEPOINT m3u_import_new_source")
+    try:
+        source_id = provenance.create_source(
+            conn,
+            profile_id=profile_id,
+            kind=kind,
+            display_name=display_name,
+        )
+        result = import_m3u(
+            conn,
+            profile_id=profile_id,
+            playlist_name=playlist_name,
+            entries=entries,
+            source_id=source_id,
+        )
+    except Exception:
+        conn.execute("ROLLBACK TO SAVEPOINT m3u_import_new_source")
+        conn.execute("RELEASE SAVEPOINT m3u_import_new_source")
+        raise
+    conn.execute("RELEASE SAVEPOINT m3u_import_new_source")
+    return M3UImportWithSourceResult(result=result, source_id=source_id)