from __future__ import annotations from datetime import UTC, datetime import hashlib import json from typing import Any from sqlalchemy import delete from sqlalchemy import insert from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.exceptions import ValidationError from app.models.entities import AuthSession from app.models.entities import Backend from app.models.entities import Blob from app.models.entities import BlobReplica from app.models.entities import CacheEntry from app.models.entities import Directory from app.models.entities import FileEntry from app.models.entities import FileVersion from app.models.entities import Job from app.models.entities import PreviewArtifact from app.models.entities import UploadSession from app.models.entities import UploadSessionPart from app.models.entities import User from app.services.storage_service import StorageService class ExportService: def __init__(self, session: AsyncSession) -> None: self.session = session self.storage_service = StorageService(session) async def export_metadata(self) -> dict: return { "users": await self._dump(User), "auth_sessions": await self._dump(AuthSession), "directories": await self._dump(Directory), "file_entries": await self._dump(FileEntry), "file_versions": await self._dump(FileVersion), "blobs": await self._dump(Blob), "blob_replicas": await self._dump(BlobReplica), "backends": await self._dump(Backend), "upload_sessions": await self._dump(UploadSession), "upload_session_parts": await self._dump(UploadSessionPart), "jobs": await self._dump(Job), "cache_entries": await self._dump(CacheEntry), "preview_artifacts": await self._dump(PreviewArtifact), "exported_at": datetime.now(UTC).isoformat(), } async def import_metadata(self, data: dict[str, list[dict[str, Any]]]) -> list[str]: validation = await self.validate_metadata_snapshot(data) if not validation["ok"]: raise ValidationError("Metadata snapshot is invalid and cannot be imported.") ordered_models = [ User, AuthSession, Directory, Backend, UploadSession, UploadSessionPart, FileEntry, FileVersion, Blob, BlobReplica, PreviewArtifact, Job, CacheEntry, ] for model in reversed(ordered_models): await self.session.execute(delete(model)) imported_tables: list[str] = [] for model in ordered_models: rows = data.get(model.__tablename__, []) if not rows: continue normalized_rows = [self._normalize_row(model, row) for row in rows] await self.session.execute(insert(model), normalized_rows) imported_tables.append(model.__tablename__) await self.session.commit() return imported_tables async def build_restore_plan(self, data: dict[str, list[dict[str, Any]]]) -> dict[str, Any]: validation = await self.validate_metadata_snapshot(data) summary = { "users": len(data.get("users", [])), "directories": len(data.get("directories", [])), "file_entries": len(data.get("file_entries", [])), "file_versions": len(data.get("file_versions", [])), "blobs": len(data.get("blobs", [])), "blob_replicas": len(data.get("blob_replicas", [])), "backends": len(data.get("backends", [])), "jobs": len(data.get("jobs", [])), "preview_artifacts": len(data.get("preview_artifacts", [])), } validation_token = None if validation["ok"]: validation_token = self._build_validation_token(data) return { "ok": validation["ok"], "issues": validation["issues"], "validation_token": validation_token, "summary": summary, } async def validate_metadata_snapshot(self, data: dict[str, list[dict[str, Any]]]) -> dict[str, Any]: issues: list[str] = [] users = data.get("users", []) directories = data.get("directories", []) file_entries = data.get("file_entries", []) file_versions = data.get("file_versions", []) blobs = data.get("blobs", []) blob_replicas = data.get("blob_replicas", []) backends = data.get("backends", []) auth_sessions = data.get("auth_sessions", []) upload_sessions = data.get("upload_sessions", []) upload_session_parts = data.get("upload_session_parts", []) jobs = data.get("jobs", []) cache_entries = data.get("cache_entries", []) preview_artifacts = data.get("preview_artifacts", []) if not users: issues.append("Snapshot must contain at least one user.") if not any(directory.get("id") == "dir_root" for directory in directories): issues.append("Snapshot must contain the root directory dir_root.") user_ids = {row["id"] for row in users if row.get("id")} directory_ids = {row["id"] for row in directories if row.get("id")} file_entry_ids = {row["id"] for row in file_entries if row.get("id")} version_ids = {row["id"] for row in file_versions if row.get("id")} blob_ids = {row["id"] for row in blobs if row.get("id")} backend_ids = {row["id"] for row in backends if row.get("id")} upload_ids = {row["id"] for row in upload_sessions if row.get("id")} for row in auth_sessions: if row.get("user_id") not in user_ids: issues.append(f"Auth session {row.get('id')} references missing user.") for row in file_entries: if row.get("directory_id") not in directory_ids: issues.append(f"File entry {row.get('id')} references missing directory.") current_version_id = row.get("current_version_id") if current_version_id is not None and current_version_id not in version_ids: issues.append(f"File entry {row.get('id')} references missing current version.") for row in file_versions: if row.get("file_entry_id") not in file_entry_ids: issues.append(f"File version {row.get('id')} references missing file entry.") for row in blobs: if row.get("file_version_id") not in version_ids: issues.append(f"Blob {row.get('id')} references missing file version.") for row in blob_replicas: if row.get("blob_id") not in blob_ids: issues.append(f"Blob replica {row.get('id')} references missing blob.") if row.get("backend_id") not in backend_ids: issues.append(f"Blob replica {row.get('id')} references missing backend.") for row in upload_sessions: directory_id = row.get("directory_id") if directory_id is not None and directory_id not in directory_ids: issues.append(f"Upload session {row.get('id')} references missing directory.") for row in upload_session_parts: if row.get("upload_session_id") not in upload_ids: issues.append(f"Upload session part {row.get('id')} references missing upload session.") for row in preview_artifacts: if row.get("file_version_id") not in version_ids: issues.append(f"Preview artifact {row.get('id')} references missing file version.") if row.get("blob_id") not in blob_ids: issues.append(f"Preview artifact {row.get('id')} references missing blob.") for row in cache_entries: if row.get("blob_id") not in blob_ids: issues.append(f"Cache entry {row.get('id')} references missing blob.") return {"ok": not issues, "issues": issues} async def verify_runtime_integrity(self) -> dict[str, Any]: snapshot = await self.export_metadata() validation = await self.validate_metadata_snapshot(snapshot) issues = list(validation["issues"]) backends = {row["id"]: row for row in snapshot["backends"]} for replica in snapshot["blob_replicas"]: if replica.get("status") != "ready": continue backend_row = backends.get(replica["backend_id"]) if backend_row is None or not backend_row.get("is_enabled", False): continue backend = await self.session.get(Backend, replica["backend_id"]) if backend is None: continue try: adapter = self.storage_service._build_adapter(backend) exists = await adapter.object_exists(replica["storage_key"]) except Exception as exc: issues.append(f"Replica {replica['id']} could not be verified: {exc}") continue if not exists: issues.append(f"Replica {replica['id']} is marked ready but object is missing.") return {"ok": not issues, "issues": issues} @staticmethod def _build_validation_token(data: dict[str, list[dict[str, Any]]]) -> str: canonical = json.dumps(data, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() async def _dump(self, model) -> list[dict]: result = await self.session.execute(select(model)) items = [] for row in result.scalars().all(): payload = {} for column in model.__table__.columns: value = getattr(row, column.name) if isinstance(value, datetime): payload[column.name] = value.isoformat() else: payload[column.name] = value items.append(payload) return items @staticmethod def _normalize_row(model, row: dict[str, Any]) -> dict[str, Any]: normalized = {} for column in model.__table__.columns: value = row.get(column.name) if value is not None and isinstance(value, str): python_type = getattr(column.type, "python_type", None) if python_type is datetime: normalized[column.name] = datetime.fromisoformat(value) continue normalized[column.name] = value return normalized