Move to FK pragma and prep other code points in light of that

app/database.py

@@ -66,7 +66,7 @@ CREATE TABLE IF NOT EXISTS raw_packets (
     data BLOB NOT NULL,
     message_id INTEGER,
     payload_hash BLOB,
-    FOREIGN KEY (message_id) REFERENCES messages(id)
+    FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE SET NULL
 );
 
 CREATE TABLE IF NOT EXISTS contact_advert_paths (
@@ -78,7 +78,7 @@ CREATE TABLE IF NOT EXISTS contact_advert_paths (
     last_seen INTEGER NOT NULL,
     heard_count INTEGER NOT NULL DEFAULT 1,
     UNIQUE(public_key, path_hex, path_len),
-    FOREIGN KEY (public_key) REFERENCES contacts(public_key)
+    FOREIGN KEY (public_key) REFERENCES contacts(public_key) ON DELETE CASCADE
 );
 
 CREATE TABLE IF NOT EXISTS contact_name_history (
@@ -88,7 +88,7 @@ CREATE TABLE IF NOT EXISTS contact_name_history (
     first_seen INTEGER NOT NULL,
     last_seen INTEGER NOT NULL,
     UNIQUE(public_key, name),
-    FOREIGN KEY (public_key) REFERENCES contacts(public_key)
+    FOREIGN KEY (public_key) REFERENCES contacts(public_key) ON DELETE CASCADE
 );
 
 CREATE INDEX IF NOT EXISTS idx_messages_received ON messages(received_at);
@@ -132,6 +132,12 @@ class Database:
         # migration 20 handles the one-time VACUUM to restructure the file.
         await self._connection.execute("PRAGMA auto_vacuum = INCREMENTAL")
 
+        # Foreign key enforcement: must be set per-connection (not persisted).
+        # Disabled during schema init and migrations to avoid issues with
+        # historical table-rebuild migrations that may temporarily violate
+        # constraints, then re-enabled for all subsequent application queries.
+        await self._connection.execute("PRAGMA foreign_keys = OFF")
+
         await self._connection.executescript(SCHEMA)
         await self._connection.commit()
         logger.debug("Database schema initialized")
@@ -141,6 +147,10 @@ class Database:
 
         await run_migrations(self._connection)
 
+        # Enable FK enforcement for all application queries from this point on.
+        await self._connection.execute("PRAGMA foreign_keys = ON")
+        logger.debug("Foreign key enforcement enabled")
+
     async def disconnect(self) -> None:
         if self._connection:
             await self._connection.close()

app/migrations.py

@@ -374,6 +374,14 @@ async def run_migrations(conn: aiosqlite.Connection) -> int:
         await set_version(conn, 48)
         applied += 1
 
+    # Migration 49: Enable foreign key enforcement — rebuild tables with
+    # CASCADE / SET NULL and clean up any orphaned rows first.
+    if version < 49:
+        logger.info("Applying migration 49: add foreign key cascade/set-null and clean orphans")
+        await _migrate_049_foreign_key_cascade(conn)
+        await set_version(conn, 49)
+        applied += 1
+
     if applied > 0:
         logger.info(
             "Applied %d migration(s), schema now at version %d", applied, await get_version(conn)
@@ -2938,3 +2946,159 @@ async def _migrate_048_discovery_blocked_types(conn: aiosqlite.Connection) -> No
         else:
             raise
     await conn.commit()
+
+
+async def _migrate_049_foreign_key_cascade(conn: aiosqlite.Connection) -> None:
+    """Rebuild FK tables with CASCADE/SET NULL and clean orphaned rows.
+
+    SQLite cannot ALTER existing FK constraints, so each table is rebuilt.
+    Orphaned child rows are cleaned up before the rebuild to ensure the
+    INSERT...SELECT into the new table (which has enforced FKs) succeeds.
+    """
+    import shutil
+    from pathlib import Path
+
+    # Back up the database before table rebuilds (skip for in-memory DBs).
+    cursor = await conn.execute("PRAGMA database_list")
+    db_row = await cursor.fetchone()
+    db_path = db_row[2] if db_row else ""
+    if db_path and db_path != ":memory:" and Path(db_path).exists():
+        backup_path = db_path + ".pre-fk-migration.bak"
+        shutil.copy2(db_path, backup_path)
+        logger.info("Database backed up to %s before FK migration", backup_path)
+
+    # --- Phase 1: clean orphans (guard each table's existence) ---
+    tables_cursor = await conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table'"
+    )
+    existing_tables = {row[0] for row in await tables_cursor.fetchall()}
+
+    if "contact_advert_paths" in existing_tables and "contacts" in existing_tables:
+        await conn.execute(
+            "DELETE FROM contact_advert_paths "
+            "WHERE public_key NOT IN (SELECT public_key FROM contacts)"
+        )
+    if "contact_name_history" in existing_tables and "contacts" in existing_tables:
+        await conn.execute(
+            "DELETE FROM contact_name_history "
+            "WHERE public_key NOT IN (SELECT public_key FROM contacts)"
+        )
+    if "raw_packets" in existing_tables and "messages" in existing_tables:
+        # Guard: message_id column may not exist on very old schemas
+        col_cursor = await conn.execute("PRAGMA table_info(raw_packets)")
+        raw_cols = {row[1] for row in await col_cursor.fetchall()}
+        if "message_id" in raw_cols:
+            await conn.execute(
+                "UPDATE raw_packets SET message_id = NULL WHERE message_id IS NOT NULL "
+                "AND message_id NOT IN (SELECT id FROM messages)"
+            )
+    await conn.commit()
+    logger.debug("Cleaned orphaned child rows before FK rebuild")
+
+    # --- Phase 2: rebuild raw_packets with ON DELETE SET NULL ---
+    # Skip if raw_packets doesn't have message_id (pre-migration-18 schema)
+    raw_has_message_id = False
+    if "raw_packets" in existing_tables:
+        col_cursor2 = await conn.execute("PRAGMA table_info(raw_packets)")
+        raw_has_message_id = "message_id" in {row[1] for row in await col_cursor2.fetchall()}
+
+    if raw_has_message_id:
+        # Dynamically build column list based on what the old table actually has,
+        # since very old schemas may lack payload_hash (added in migration 28).
+        col_cursor3 = await conn.execute("PRAGMA table_info(raw_packets)")
+        old_cols = [row[1] for row in await col_cursor3.fetchall()]
+
+        new_col_defs = [
+            "id INTEGER PRIMARY KEY AUTOINCREMENT",
+            "timestamp INTEGER NOT NULL",
+            "data BLOB NOT NULL",
+            "message_id INTEGER",
+        ]
+        copy_cols = ["id", "timestamp", "data", "message_id"]
+        if "payload_hash" in old_cols:
+            new_col_defs.append("payload_hash BLOB")
+            copy_cols.append("payload_hash")
+        new_col_defs.append(
+            "FOREIGN KEY (message_id) REFERENCES messages(id) ON DELETE SET NULL"
+        )
+
+        cols_sql = ", ".join(new_col_defs)
+        copy_sql = ", ".join(copy_cols)
+        await conn.execute(f"CREATE TABLE raw_packets_fk ({cols_sql})")
+        await conn.execute(
+            f"INSERT INTO raw_packets_fk ({copy_sql}) SELECT {copy_sql} FROM raw_packets"
+        )
+        await conn.execute("DROP TABLE raw_packets")
+        await conn.execute("ALTER TABLE raw_packets_fk RENAME TO raw_packets")
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_raw_packets_message_id ON raw_packets(message_id)"
+        )
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_raw_packets_timestamp ON raw_packets(timestamp)"
+        )
+        if "payload_hash" in old_cols:
+            await conn.execute(
+                "CREATE UNIQUE INDEX IF NOT EXISTS idx_raw_packets_payload_hash ON raw_packets(payload_hash)"
+            )
+        await conn.commit()
+        logger.debug("Rebuilt raw_packets with ON DELETE SET NULL")
+
+    # --- Phase 3: rebuild contact_advert_paths with ON DELETE CASCADE ---
+    if "contact_advert_paths" in existing_tables:
+        await conn.execute(
+            """
+            CREATE TABLE contact_advert_paths_fk (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                public_key TEXT NOT NULL,
+                path_hex TEXT NOT NULL,
+                path_len INTEGER NOT NULL,
+                first_seen INTEGER NOT NULL,
+                last_seen INTEGER NOT NULL,
+                heard_count INTEGER NOT NULL DEFAULT 1,
+                UNIQUE(public_key, path_hex, path_len),
+                FOREIGN KEY (public_key) REFERENCES contacts(public_key) ON DELETE CASCADE
+            )
+            """
+        )
+        await conn.execute(
+            "INSERT INTO contact_advert_paths_fk (id, public_key, path_hex, path_len, first_seen, last_seen, heard_count) "
+            "SELECT id, public_key, path_hex, path_len, first_seen, last_seen, heard_count FROM contact_advert_paths"
+        )
+        await conn.execute("DROP TABLE contact_advert_paths")
+        await conn.execute("ALTER TABLE contact_advert_paths_fk RENAME TO contact_advert_paths")
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_contact_advert_paths_recent "
+            "ON contact_advert_paths(public_key, last_seen DESC)"
+        )
+        await conn.commit()
+        logger.debug("Rebuilt contact_advert_paths with ON DELETE CASCADE")
+
+    # --- Phase 4: rebuild contact_name_history with ON DELETE CASCADE ---
+    if "contact_name_history" in existing_tables:
+        await conn.execute(
+            """
+            CREATE TABLE contact_name_history_fk (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                public_key TEXT NOT NULL,
+                name TEXT NOT NULL,
+                first_seen INTEGER NOT NULL,
+                last_seen INTEGER NOT NULL,
+                UNIQUE(public_key, name),
+                FOREIGN KEY (public_key) REFERENCES contacts(public_key) ON DELETE CASCADE
+            )
+            """
+        )
+        await conn.execute(
+            "INSERT INTO contact_name_history_fk (id, public_key, name, first_seen, last_seen) "
+            "SELECT id, public_key, name, first_seen, last_seen FROM contact_name_history"
+        )
+        await conn.execute("DROP TABLE contact_name_history")
+        await conn.execute("ALTER TABLE contact_name_history_fk RENAME TO contact_name_history")
+        await conn.execute(
+            "CREATE INDEX IF NOT EXISTS idx_contact_name_history_key "
+            "ON contact_name_history(public_key, last_seen DESC)"
+        )
+        await conn.commit()
+        logger.debug("Rebuilt contact_name_history with ON DELETE CASCADE")
+
+

app/packet_processor.py

@@ -476,15 +476,6 @@ async def _process_advertisement(
             )
             return
 
-    # Keep recent unique advert paths for all contacts.
-    await ContactAdvertPathRepository.record_observation(
-        public_key=advert.public_key.lower(),
-        path_hex=new_path_hex,
-        timestamp=timestamp,
-        max_paths=10,
-        hop_count=new_path_len,
-    )
-
     contact_upsert = ContactUpsert(
         public_key=advert.public_key.lower(),
         name=advert.name,
@@ -496,7 +487,18 @@ async def _process_advertisement(
         first_seen=timestamp,  # COALESCE in upsert preserves existing value
     )
 
+    # Upsert the contact BEFORE recording advert paths so the parent row
+    # exists when foreign key enforcement is enabled.
     await ContactRepository.upsert(contact_upsert)
+
+    # Keep recent unique advert paths for all contacts.
+    await ContactAdvertPathRepository.record_observation(
+        public_key=advert.public_key.lower(),
+        path_hex=new_path_hex,
+        timestamp=timestamp,
+        max_paths=10,
+        hop_count=new_path_len,
+    )
     promoted_keys = await promote_prefix_contacts_for_contact(
         public_key=advert.public_key,
         log=logger,

app/repository/messages.py

@@ -572,6 +572,9 @@ class MessageRepository:
     @staticmethod
     async def delete_by_id(message_id: int) -> None:
         """Delete a message row by ID."""
+        await db.conn.execute(
+            "UPDATE raw_packets SET message_id = NULL WHERE message_id = ?", (message_id,)
+        )
         await db.conn.execute("DELETE FROM messages WHERE id = ?", (message_id,))
         await db.conn.commit()