Deploy public authenticated AutoML gateway

auth_storage.py

@@ -113,6 +113,110 @@ def _copy_valid_database(source: Path, destination: Path) -> bool:
         staged.unlink(missing_ok=True)
 
 
+def _salvage_database(source: Path, destination: Path) -> bool:
+    """Rebuild readable application tables without trusting damaged indexes."""
+
+    if not source.is_file():
+        return False
+    staged = destination.with_suffix(destination.suffix + ".salvage")
+    staged.unlink(missing_ok=True)
+    source_conn = None
+    destination_conn = None
+    tables = (
+        (
+            "users",
+            frozenset(
+                {
+                    "user_id",
+                    "username",
+                    "username_normalized",
+                    "password_hash",
+                    "password_salt",
+                }
+            ),
+        ),
+        (
+            "saved_models",
+            frozenset(
+                {
+                    "model_id",
+                    "user_id",
+                    "manifest_json",
+                    "storage_path",
+                    "sha256",
+                }
+            ),
+        ),
+        (
+            "scheduled_jobs",
+            frozenset(
+                {
+                    "request_id",
+                    "user_id",
+                    "idempotency_key",
+                    "provider",
+                    "status",
+                }
+            ),
+        ),
+    )
+    try:
+        uri = f"file:{source.resolve()}?mode=ro&immutable=1"
+        source_conn = sqlite3.connect(uri, uri=True, timeout=5)
+        source_conn.row_factory = sqlite3.Row
+        source_conn.execute("PRAGMA writable_schema=ON")
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        destination_conn = sqlite3.connect(str(staged), timeout=30)
+        destination_conn.row_factory = sqlite3.Row
+        destination_conn.execute("PRAGMA journal_mode=DELETE")
+        destination_conn.execute("PRAGMA foreign_keys=ON")
+        _create_schema(destination_conn)
+
+        for table, required_columns in tables:
+            columns = [
+                str(row["name"])
+                for row in source_conn.execute(
+                    f"PRAGMA table_info({table})"
+                ).fetchall()
+            ]
+            if not required_columns.issubset(columns) or any(
+                not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", column)
+                for column in columns
+            ):
+                return False
+            rows = source_conn.execute(f"SELECT * FROM {table}").fetchall()
+            source_count = int(
+                source_conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+            )
+            if len(rows) != source_count:
+                return False
+            if rows:
+                quoted_columns = ", ".join(f'"{column}"' for column in columns)
+                placeholders = ", ".join("?" for _ in columns)
+                destination_conn.executemany(
+                    f"INSERT INTO {table} ({quoted_columns}) VALUES ({placeholders})",
+                    [tuple(row[column] for column in columns) for row in rows],
+                )
+
+        if destination_conn.execute("PRAGMA foreign_key_check").fetchone():
+            return False
+        destination_conn.commit()
+        destination_conn.close()
+        destination_conn = None
+        if not _database_integrity_ok(staged):
+            return False
+        os.replace(staged, destination)
+        return True
+    except (OSError, sqlite3.DatabaseError, TypeError, ValueError):
+        return False
+    finally:
+        if source_conn is not None:
+            source_conn.close()
+        if destination_conn is not None:
+            destination_conn.close()
+        staged.unlink(missing_ok=True)
+
+
 def _prepare_runtime_database() -> None:
     global _RUNTIME_DB_PREPARED
     if _RUNTIME_DB_PREPARED or not _snapshot_database_enabled():
@@ -133,7 +237,10 @@ def _prepare_runtime_database() -> None:
     )
     if any(candidate.is_file() for candidate in candidates):
         for candidate in candidates:
-            if _copy_valid_database(candidate, runtime_path):
+            if _copy_valid_database(candidate, runtime_path) or _salvage_database(
+                candidate,
+                runtime_path,
+            ):
                 _RUNTIME_DB_PREPARED = True
                 return
         raise StorageConfigurationError(