diff --git a/application/single_app/config.py b/application/single_app/config.py index 9f8af2cc3..e7b780d6e 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -95,7 +95,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.105" +VERSION = "0.250.106" IS_DEVELOPMENT = is_development_env_enabled() SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') diff --git a/application/single_app/functions_data_management.py b/application/single_app/functions_data_management.py index e46f16504..6c17aaf08 100644 --- a/application/single_app/functions_data_management.py +++ b/application/single_app/functions_data_management.py @@ -84,6 +84,19 @@ start_backup_resource, update_backup_resource, ) +from functions_data_management_restore_state import ( + RESTORE_RESOURCE_STATUS_COMPLETED, + RESTORE_RESOURCE_STATUS_FAILED, + build_restore_configuration_fingerprint, + complete_restore_attempt, + complete_restore_resource, + fail_restore_resource, + initialize_restore_state, + is_restore_resource_completed, + start_restore_attempt, + start_restore_resource, + update_restore_resource, +) from functions_data_management_search_write_fence import ( acquire_data_management_search_write_fence, acquire_data_management_target_migration_coordinator, @@ -122,6 +135,7 @@ DATA_MANAGEMENT_MIGRATION_MANIFEST_BATCH_TYPE = "data_management_migration_manifest_batch" DATA_MANAGEMENT_MIGRATION_REVIEW_TYPE = "data_management_migration_review" DATA_MANAGEMENT_MIGRATION_REVIEW_TTL_SECONDS = 900 +DATA_MANAGEMENT_RESTORE_REVIEW_TYPE = "data_management_restore_review" DATA_MANAGEMENT_MIRROR_DELETION_BATCH_TYPE = "data_management_mirror_deletion_batch" DATA_MANAGEMENT_MIGRATION_LOCK_TYPE = "data_management_migration_lock" DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_TYPE = "data_management_backup_manifest_batch" @@ -235,6 +249,16 @@ DATA_MANAGEMENT_MIGRATION_MODE_MIRROR, } DATA_MANAGEMENT_MIRROR_CONFIRMATION = "MIRROR WITH DELETIONS" +DATA_MANAGEMENT_RESTORE_POLICY_CREATE_ONLY = "create_only" +DATA_MANAGEMENT_RESTORE_POLICY_OVERWRITE = "overwrite_existing" +DATA_MANAGEMENT_RESTORE_POLICIES = { + DATA_MANAGEMENT_RESTORE_POLICY_CREATE_ONLY, + DATA_MANAGEMENT_RESTORE_POLICY_OVERWRITE, +} +DATA_MANAGEMENT_RESTORE_OVERWRITE_CONFIRMATION = "RESTORE WITH OVERWRITE" +DATA_MANAGEMENT_RESTORE_REVIEW_TTL_SECONDS = 900 +DATA_MANAGEMENT_RESTORE_MAX_MANIFEST_ENTRIES = 100000 +DATA_MANAGEMENT_RESTORE_BATCH_SIZE = 100 DATA_MANAGEMENT_SEARCH_WRITE_FREEZE_CONFIRMATION_ERROR = ( "Confirm that external destination AI Search writers are frozen before migrating AI Search documents." ) @@ -339,6 +363,14 @@ class DataManagementBackupCanceledError(RuntimeError): class DataManagementBackupOverlapError(RuntimeError): """Raised when another backup owns the same source scope.""" + +class DataManagementRestoreLeaseLostError(RuntimeError): + """Raised when a stale restore worker no longer owns its job lease.""" + + +class DataManagementRestoreCanceledError(RuntimeError): + """Raised when an administrator requests cooperative restore cancellation.""" + DATA_MANAGEMENT_FRONTEND_SECRET_FIELDS = { "backup_storage_connection_string", "encryption_key_reference", @@ -7416,6 +7448,41 @@ def _assert_data_management_job_lease(job, allow_cancel_requested=False): _assert_migration_job_lease(job, allow_cancel_requested=allow_cancel_requested) elif job.get("operation") == DATA_MANAGEMENT_OPERATION_BACKUP: _assert_backup_job_lease(job, allow_cancel_requested=allow_cancel_requested) + elif job.get("operation") == DATA_MANAGEMENT_OPERATION_RESTORE: + _assert_restore_job_lease(job, allow_cancel_requested=allow_cancel_requested) + + +def _assert_restore_job_lease(job, allow_cancel_requested=False): + """Stop a stale restore worker before it mutates another target resource.""" + if not isinstance(job, dict) or job.get("operation") != DATA_MANAGEMENT_OPERATION_RESTORE: + return + lease_holder_id = _safe_text(job.get("lease_holder_id")) + if not lease_holder_id: + raise DataManagementRestoreLeaseLostError("Restore worker has no durable job lease.") + try: + persisted_job = _read_job(job.get("id")) + except Exception as exc: + raise DataManagementRestoreLeaseLostError( + "Restore worker could not verify its durable job lease." + ) from exc + if persisted_job.get("cancel_requested_at") and not allow_cancel_requested: + raise DataManagementRestoreCanceledError( + "Restore cancellation was requested by an administrator." + ) + persisted_expiry = _parse_iso_datetime(persisted_job.get("lease_expires_at")) + if ( + persisted_job.get("status") != DATA_MANAGEMENT_STATUS_RUNNING or + _safe_text(persisted_job.get("lease_holder_id")) != lease_holder_id or + _safe_int(persisted_job.get("lease_generation"), default=0) != + _safe_int(job.get("lease_generation"), default=0) or + persisted_expiry is None or + persisted_expiry <= _now_utc() + ): + raise DataManagementRestoreLeaseLostError( + "Restore worker lease was superseded or expired." + ) + if persisted_job.get("_etag"): + job["_etag"] = persisted_job.get("_etag") def _run_backup_transfer_with_heartbeat(job, settings, message, transfer): @@ -7730,236 +7797,1052 @@ def _build_backup_lineage_id(backup_plan): }) -def queue_data_management_job(operation, backup_type=None, requested_by=None, requested_by_email=None, options=None, scheduled=False, occurrence_id=None): - normalized_operation = _safe_text(operation) - if normalized_operation not in DATA_MANAGEMENT_OPERATIONS: - raise ValueError("Unsupported data management operation.") - - normalized_backup_type = _safe_text(backup_type) - if normalized_operation == DATA_MANAGEMENT_OPERATION_BACKUP and normalized_backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: - raise ValueError("Backup jobs must be full or partial.") - if normalized_operation != DATA_MANAGEMENT_OPERATION_BACKUP and normalized_backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: - normalized_backup_type = None - - now = _now_iso() - job_id = occurrence_id or str(uuid.uuid4()) - backup_plan = None - backup_state = None - source_scope = None - source_cutoff_at = None - normalized_options = options if isinstance(options, dict) else {} - if normalized_operation == DATA_MANAGEMENT_OPERATION_BACKUP: - settings = get_data_management_settings() - if settings.get("encryption_enabled") and not settings.get("encryption_key_reference"): - generate_data_management_encryption_key() - settings = get_data_management_settings() - backup_plan = _normalize_data_management_backup_plan( - settings, - normalized_backup_type, - normalized_options, - source_cutoff_at=now, - ) - source_scope = backup_plan["source_scope"] - source_cutoff_at = backup_plan["source_cutoff_at"] - backup_state = initialize_backup_state( - None, - job_id, - backup_plan, - source_scope, - source_cutoff_at, - ) - job = { - "id": job_id, - "type": DATA_MANAGEMENT_JOB_TYPE, - "operation": normalized_operation, - "backup_type": normalized_backup_type, - "status": DATA_MANAGEMENT_STATUS_QUEUED, - "created_at": now, - "updated_at": now, - "started_at": None, - "completed_at": None, - "last_heartbeat_at": None, - "last_message": "Queued data management job", - "last_error": None, - "cancel_requested_at": None, - "cancel_requested_by": None, - "cancel_requested_by_email": None, - "cancel_reason": None, - "requested_by": _safe_text(requested_by), - "requested_by_email": _safe_text(requested_by_email), - "scheduled": bool(scheduled), - "occurrence_id": occurrence_id, - "options": normalized_options, - "progress": { - "total_steps": 0, - "completed_steps": 0, - "current_step": None, - "percent_complete": 0, - }, - "lease_holder_id": None, - "lease_expires_at": None, - "lease_generation": 0, - "warnings": [], - "source_scope": source_scope, - "source_cutoff_at": source_cutoff_at, - "backup_plan": backup_plan, - "backup_state": backup_state, - } - - try: - saved_job = cosmos_data_management_jobs_container.create_item(body=job) - _record_data_management_job_event( - saved_job.get("id"), - "queued", - saved_job, - status=DATA_MANAGEMENT_STATUS_QUEUED, - message="Queued data management job", - details={"operation": normalized_operation, "backup_type": normalized_backup_type, "scheduled": bool(scheduled)}, - ) - return saved_job - except Exception as exc: - if occurrence_id and getattr(exc, "status_code", None) == 409: - return _read_job(job_id) - raise - - -def _normalize_data_management_history_datetime(value, field_name, end_exclusive=False): - text = _safe_text(value) - if not text: - return None - date_only = bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", text)) - parsed = _parse_iso_datetime(text) - if parsed is None: - raise DataManagementHistoryPaginationError( - f"{field_name} must be an ISO-8601 date or timestamp." - ) - if date_only and end_exclusive: - parsed += timedelta(days=1) - return parsed.isoformat() - - -def normalize_data_management_history_filters(list_kind, filters=None): - normalized_kind = _safe_text(list_kind).lower() - if normalized_kind not in {"jobs", "backups"}: - raise DataManagementHistoryPaginationError("Unsupported Data Management history list.") - - raw_filters = filters if isinstance(filters, dict) else {} - status = _safe_text(raw_filters.get("status")).lower() - allowed_statuses = { - DATA_MANAGEMENT_STATUS_QUEUED, - DATA_MANAGEMENT_STATUS_RUNNING, +def _get_restore_backup_job(backup_id): + safe_backup_id = _safe_text(backup_id) + if not safe_backup_id: + raise DataManagementSettingsValidationError("Choose a backup before running restore review.") + job = get_data_management_job(safe_backup_id) + if not job or job.get("operation") != DATA_MANAGEMENT_OPERATION_BACKUP: + raise DataManagementSettingsValidationError("Selected backup was not found.") + if job.get("status") not in { DATA_MANAGEMENT_STATUS_COMPLETED, DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS, - DATA_MANAGEMENT_STATUS_FAILED, - DATA_MANAGEMENT_STATUS_CANCELED, - } - if normalized_kind == "backups": - allowed_statuses.add("available") - if status and status not in allowed_statuses: - raise DataManagementHistoryPaginationError("Unsupported Data Management status filter.") + }: + raise DataManagementSettingsValidationError("Only completed backups can be restored.") + return job - scheduled_value = _safe_text(raw_filters.get("scheduled")).lower() - if scheduled_value in {"", "all"}: - scheduled = None - elif scheduled_value == "scheduled": - scheduled = True - elif scheduled_value == "manual": - scheduled = False - else: - raise DataManagementHistoryPaginationError("Scheduled filter must be all, scheduled, or manual.") - created_from = _normalize_data_management_history_datetime( - raw_filters.get("created_from"), - "created_from", - ) - created_to = _normalize_data_management_history_datetime( - raw_filters.get("created_to"), - "created_to", - end_exclusive=True, - ) - if bool(created_from) != bool(created_to): - raise DataManagementHistoryPaginationError( - "created_from and created_to must be provided together." - ) - if created_from and created_to: - start = _parse_iso_datetime(created_from) - end = _parse_iso_datetime(created_to) - if end <= start: - raise DataManagementHistoryPaginationError("created_to must be after created_from.") - if end - start > timedelta(days=DATA_MANAGEMENT_HISTORY_MAX_DATE_RANGE_DAYS): - raise DataManagementHistoryPaginationError( - f"Date range cannot exceed {DATA_MANAGEMENT_HISTORY_MAX_DATE_RANGE_DAYS} days." +def _normalize_data_management_restore_plan(options=None, backup_job=None, require_confirmation=False): + """Build the secret-free immutable plan used by one restore job.""" + source = options if isinstance(options, dict) else {} + restore_plan = source.get("restore_plan") if isinstance(source.get("restore_plan"), dict) else source + backup_id = _safe_text( + restore_plan.get("backup_id") or + restore_plan.get("source_backup_id") or + source.get("backup_id") or + source.get("source_backup_id") + ) + selected_backup = backup_job if isinstance(backup_job, dict) else _get_restore_backup_job(backup_id) + backup_plan = selected_backup.get("backup_plan") if isinstance(selected_backup.get("backup_plan"), dict) else {} + backup_result = selected_backup.get("result") if isinstance(selected_backup.get("result"), dict) else {} + manifest_path = _safe_text(backup_result.get("manifest_path") or (selected_backup.get("backup_state") or {}).get("manifest", {}).get("path")) + if not manifest_path: + raise DataManagementSettingsValidationError("Selected backup does not have a durable manifest.") + + restore_policy = _safe_text( + restore_plan.get("restore_policy") or restore_plan.get("policy"), + DATA_MANAGEMENT_RESTORE_POLICY_CREATE_ONLY, + ) + if restore_policy not in DATA_MANAGEMENT_RESTORE_POLICIES: + restore_policy = DATA_MANAGEMENT_RESTORE_POLICY_CREATE_ONLY + overwrite_confirmed = _safe_bool(restore_plan.get("overwrite_confirmed"), False) + overwrite_phrase = _safe_text(restore_plan.get("overwrite_confirmation_phrase")) + if restore_policy == DATA_MANAGEMENT_RESTORE_POLICY_OVERWRITE: + phrase_matches = hmac.compare_digest( + overwrite_phrase, + DATA_MANAGEMENT_RESTORE_OVERWRITE_CONFIRMATION, + ) + if require_confirmation and (not overwrite_confirmed or not phrase_matches): + raise DataManagementSettingsValidationError( + "Overwrite restore requires a separate confirmation phrase." ) - - operation = None - backup_type = None - if normalized_kind == "jobs": - operation = _safe_text(raw_filters.get("operation")).lower() - if operation and operation not in DATA_MANAGEMENT_OPERATIONS: - raise DataManagementHistoryPaginationError("Unsupported Data Management operation filter.") else: - backup_type = _safe_text(raw_filters.get("backup_type")).lower() - if backup_type and backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: - raise DataManagementHistoryPaginationError("Backup type filter must be full or partial.") + overwrite_confirmed = False + overwrite_phrase = "" return { - "operation": operation or None, - "backup_type": backup_type or None, - "status": status or None, - "scheduled": scheduled, - "created_from": created_from, - "created_to": created_to, + "source_backup_id": selected_backup.get("id"), + "source_backup_type": selected_backup.get("backup_type"), + "source_backup_completed_at": selected_backup.get("completed_at"), + "source_backup_status": selected_backup.get("status"), + "source_backup_manifest_path": manifest_path, + "base_prefix": _safe_text(backup_result.get("base_prefix")), + "restore_policy": restore_policy, + "overwrite_confirmed": overwrite_confirmed, + "include_cosmos": _safe_bool( + restore_plan.get("include_cosmos"), + _safe_bool(backup_plan.get("include_cosmos"), True), + ), + "include_ai_search": _safe_bool( + restore_plan.get("include_ai_search"), + _safe_bool(backup_plan.get("include_ai_search"), True), + ), + "include_source_blobs": _safe_bool( + restore_plan.get("include_source_blobs"), + _safe_bool(backup_plan.get("include_source_blobs"), True), + ), + "backup_storage_container_name": _safe_text(backup_plan.get("backup_storage_container_name")), + "backup_storage_path_prefix": _safe_text(backup_plan.get("backup_storage_path_prefix")).strip("/"), + "backup_storage_identity": _safe_text(backup_plan.get("storage_identity")), + "backup_encryption_enabled": _safe_bool(backup_plan.get("encryption_enabled"), False), + "backup_encryption_key_storage": _safe_text(backup_plan.get("encryption_key_storage")), + "backup_encryption_key_reference": _safe_text(backup_plan.get("encryption_key_reference")), + "backup_encryption_key_fingerprint": _safe_text(backup_plan.get("encryption_key_fingerprint")), + "deletion_policy": _safe_text(backup_plan.get("deletion_policy") or (backup_plan.get("source_cutoff_semantics") or {}).get("deletion_policy")), + "source_cutoff_at": _safe_text(backup_plan.get("source_cutoff_at")), + "source_lower_bound_at": _safe_text(backup_plan.get("source_lower_bound_at")), + "differential_mode": _safe_text(backup_plan.get("differential_mode")), } -def _get_data_management_history_cipher(): - secret = _safe_text(getattr(app_config, "SECRET_KEY", "")) - derived_key = hashlib.sha256( - f"data-management-history:{secret}".encode("utf-8") - ).digest() - return Fernet(base64.urlsafe_b64encode(derived_key)) +def _restore_review_fingerprint(settings, restore_plan): + """Bind a restore review to target routing and the immutable restore plan.""" + plan = restore_plan if isinstance(restore_plan, dict) else {} + normalized_settings = normalize_data_management_settings(existing_settings=settings) + target_identity = { + "target_cosmos_authentication_type": _safe_text(normalized_settings.get("target_cosmos_authentication_type")), + "target_cosmos_endpoint_hash": hashlib.sha256( + _safe_text(normalized_settings.get("target_cosmos_endpoint")).lower().encode("utf-8") + ).hexdigest(), + "target_ai_search_authentication_type": _safe_text(normalized_settings.get("target_ai_search_authentication_type")), + "target_ai_search_endpoint_hash": hashlib.sha256( + _safe_text(normalized_settings.get("target_ai_search_endpoint")).lower().encode("utf-8") + ).hexdigest(), + "target_enhanced_citations_storage_authentication_type": _safe_text( + normalized_settings.get("target_enhanced_citations_storage_authentication_type") + ), + "target_enhanced_citations_storage_endpoint_hash": hashlib.sha256( + _safe_text(normalized_settings.get("target_enhanced_citations_storage_blob_endpoint")).lower().encode("utf-8") + ).hexdigest(), + "restore_policy": _safe_text(plan.get("restore_policy")), + } + return build_restore_configuration_fingerprint({ + "restore_plan": plan, + "target_identity": target_identity, + }) -def _encode_data_management_history_token(list_kind, filters, page_size, cursor): - if not isinstance(cursor, dict): - return None - cursor_created_at = _safe_text(cursor.get("created_at")) - cursor_id = _safe_text(cursor.get("id")) - if not cursor_created_at or not cursor_id: - return None - payload = { - "version": DATA_MANAGEMENT_HISTORY_TOKEN_VERSION, - "list_kind": list_kind, - "filters": filters, - "page_size": page_size, - "sort": DATA_MANAGEMENT_HISTORY_SORT, - "cursor": { - "created_at": cursor_created_at, - "id": cursor_id, - }, - } - token = _get_data_management_history_cipher().encrypt( - json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") +def get_data_management_restore_review_fingerprint(settings=None, restore_plan=None): + """Return the execution fingerprint for current restore settings and plan.""" + normalized_settings = _normalize_data_management_settings_from_payload(settings) + normalized_plan = _normalize_data_management_restore_plan( + restore_plan if isinstance(restore_plan, dict) else {}, ) - return token.decode("utf-8") + return _restore_review_fingerprint(normalized_settings, normalized_plan) -def _decode_data_management_history_token(token, list_kind, filters, page_size): - safe_token = _safe_text(token) - if not safe_token: - return None - try: - decoded_token = base64.b64decode( - safe_token.encode("utf-8"), - altchars=b"-_", - validate=True, +def _restore_review_admin_hash(admin_user_id): + normalized_user_id = _safe_text(admin_user_id) + if not normalized_user_id or normalized_user_id == "unknown": + raise DataManagementSettingsValidationError( + "An authenticated administrator is required for restore review." ) - canonical_token = base64.urlsafe_b64encode(decoded_token).decode("utf-8") - if not hmac.compare_digest(canonical_token, safe_token): - raise ValueError("Continuation token is not canonical.") - decrypted = _get_data_management_history_cipher().decrypt( - safe_token.encode("utf-8"), - ttl=DATA_MANAGEMENT_HISTORY_TOKEN_TTL_SECONDS, + return hashlib.sha256(normalized_user_id.encode("utf-8")).hexdigest() + + +def create_data_management_restore_review_authorization(admin_user_id, review_fingerprint): + """Persist one short-lived, admin-bound authorization for a ready restore review.""" + normalized_fingerprint = _safe_text(review_fingerprint) + if not normalized_fingerprint: + raise DataManagementSettingsValidationError("Restore review fingerprint is required.") + now = _now_utc() + authorization_id = str(uuid.uuid4()) + authorization = { + "id": authorization_id, + "type": DATA_MANAGEMENT_RESTORE_REVIEW_TYPE, + "status": "ready", + "admin_hash": _restore_review_admin_hash(admin_user_id), + "review_fingerprint": normalized_fingerprint, + "created_at": now.isoformat(), + "expires_at": (now + timedelta(seconds=DATA_MANAGEMENT_RESTORE_REVIEW_TTL_SECONDS)).isoformat(), + } + created = cosmos_data_management_jobs_container.create_item(body=authorization) + return { + "authorization_token": created.get("id", authorization_id), + "authorization_expires_at": created.get("expires_at", authorization["expires_at"]), + } + + +def _read_data_management_restore_review_authorization( + authorization_token, + admin_user_id, + review_fingerprint, + allow_expired=False, +): + normalized_token = _safe_text(authorization_token) + normalized_fingerprint = _safe_text(review_fingerprint) + if not normalized_token or not normalized_fingerprint: + raise DataManagementSettingsValidationError("A current restore review authorization is required.") + try: + authorization = cosmos_data_management_jobs_container.read_item( + item=normalized_token, + partition_key=normalized_token, + ) + except (CosmosResourceNotFoundError, KeyError, ResourceNotFoundError) as exc: + raise DataManagementSettingsValidationError( + "Restore review authorization is invalid or expired." + ) from exc + expires_at = _parse_iso_datetime(authorization.get("expires_at")) + if ( + authorization.get("type") != DATA_MANAGEMENT_RESTORE_REVIEW_TYPE or + (not allow_expired and (not expires_at or expires_at <= _now_utc())) or + not hmac.compare_digest( + _safe_text(authorization.get("admin_hash")), + _restore_review_admin_hash(admin_user_id), + ) or + not hmac.compare_digest( + _safe_text(authorization.get("review_fingerprint")), + normalized_fingerprint, + ) + ): + raise DataManagementSettingsValidationError( + "Restore review authorization is invalid or expired." + ) + return authorization + + +def reserve_data_management_restore_review_authorization( + authorization_token, + admin_user_id, + review_fingerprint, +): + """Reserve one ready restore authorization before durable job creation.""" + authorization = _read_data_management_restore_review_authorization( + authorization_token, + admin_user_id, + review_fingerprint, + ) + if authorization.get("status") != "ready": + raise DataManagementSettingsValidationError( + "Restore review authorization is already reserved or used." + ) + reservation_token = uuid.uuid4().hex + job_id = str(uuid.uuid5( + uuid.NAMESPACE_URL, + f"simplechat-restore-review:{authorization['id']}", + )) + replacement = copy.deepcopy(authorization) + replacement.update({ + "status": "reserved", + "reservation_token": reservation_token, + "reserved_at": _now_iso(), + "job_id": job_id, + }) + try: + cosmos_data_management_jobs_container.replace_item( + item=authorization["id"], + body=replacement, + etag=authorization.get("_etag"), + match_condition=MatchConditions.IfNotModified, + ) + except Exception as exc: + if getattr(exc, "status_code", None) in {409, 412}: + raise DataManagementSettingsValidationError( + "Restore review authorization was already reserved or used." + ) from exc + raise + return { + "authorization_token": authorization["id"], + "reservation_token": reservation_token, + "job_id": job_id, + } + + +def release_data_management_restore_review_reservation(authorization_token, reservation_token): + """Release an exact restore review reservation when durable job creation fails.""" + normalized_token = _safe_text(authorization_token) + normalized_reservation = _safe_text(reservation_token) + if not normalized_token or not normalized_reservation: + return False + try: + authorization = cosmos_data_management_jobs_container.read_item( + item=normalized_token, + partition_key=normalized_token, + ) + except (CosmosResourceNotFoundError, KeyError, ResourceNotFoundError): + return False + if ( + authorization.get("status") != "reserved" or + not hmac.compare_digest(_safe_text(authorization.get("reservation_token")), normalized_reservation) + ): + return False + replacement = copy.deepcopy(authorization) + replacement.update({ + "status": "ready", + "reservation_released_at": _now_iso(), + }) + replacement.pop("reservation_token", None) + replacement.pop("reserved_at", None) + replacement.pop("job_id", None) + try: + cosmos_data_management_jobs_container.replace_item( + item=normalized_token, + body=replacement, + etag=authorization.get("_etag"), + match_condition=MatchConditions.IfNotModified, + ) + except Exception as exc: + if getattr(exc, "status_code", None) in {409, 412}: + return False + log_event( + "[DataManagement] Failed to release restore review reservation.", + {"authorization_id": normalized_token, "error_type": type(exc).__name__}, + level=logging.WARNING, + exceptionTraceback=True, + ) + return False + return True + + +def consume_data_management_restore_review_authorization( + authorization_token, + admin_user_id, + review_fingerprint, + reservation_token, + job_id, +): + """Consume an exact restore review reservation before work starts.""" + authorization = _read_data_management_restore_review_authorization( + authorization_token, + admin_user_id, + review_fingerprint, + allow_expired=True, + ) + normalized_reservation = _safe_text(reservation_token) + normalized_job_id = _safe_text(job_id) + if ( + authorization.get("status") == "consumed" and + hmac.compare_digest(_safe_text(authorization.get("consumed_job_id")), normalized_job_id) + ): + return True + if ( + authorization.get("status") != "reserved" or + not normalized_reservation or + not normalized_job_id or + not hmac.compare_digest(_safe_text(authorization.get("reservation_token")), normalized_reservation) or + not hmac.compare_digest(_safe_text(authorization.get("job_id")), normalized_job_id) + ): + raise DataManagementSettingsValidationError( + "Restore review reservation is invalid or already used." + ) + replacement = copy.deepcopy(authorization) + replacement.update({ + "status": "consumed", + "consumed_at": _now_iso(), + "consumed_job_id": normalized_job_id, + }) + replacement.pop("reservation_token", None) + try: + cosmos_data_management_jobs_container.replace_item( + item=authorization["id"], + body=replacement, + etag=authorization.get("_etag"), + match_condition=MatchConditions.IfNotModified, + ) + except Exception as exc: + if getattr(exc, "status_code", None) in {409, 412}: + raise DataManagementSettingsValidationError( + "Restore review reservation changed before consumption." + ) from exc + raise + return True + + +def _get_restore_fernet(settings, backup_plan): + restore_settings = copy.deepcopy(settings if isinstance(settings, dict) else {}) + restore_settings["encryption_enabled"] = _safe_bool(backup_plan.get("encryption_enabled"), False) + return _get_backup_fernet( + restore_settings, + key_reference=_safe_text(backup_plan.get("encryption_key_reference")), + ) + + +def _download_backup_artifact_bytes(container_client, artifact_path): + safe_path = _safe_text(artifact_path) + if not safe_path: + raise DataManagementSettingsValidationError("Backup artifact path is missing.") + download = container_client.get_blob_client(safe_path).download_blob( + timeout=DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + ) + return download.readall() + + +def _read_backup_json_artifact(container_client, artifact_path, fernet=None): + data = _download_backup_artifact_bytes(container_client, artifact_path) + if fernet: + data = fernet.decrypt(data) + try: + payload = json.loads(data.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DataManagementSettingsValidationError("Backup JSON artifact is invalid.") from exc + if not isinstance(payload, dict): + raise DataManagementSettingsValidationError("Backup JSON artifact must be an object.") + return payload + + +def _iter_backup_jsonl_artifact(container_client, artifact_path, fernet=None): + data = _download_backup_artifact_bytes(container_client, artifact_path) + for raw_line in data.splitlines(): + if not raw_line: + continue + line_bytes = raw_line + if fernet: + line_bytes = fernet.decrypt(raw_line) + try: + yield json.loads(line_bytes.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise DataManagementSettingsValidationError("Backup JSONL artifact contains an invalid record.") from exc + + +def _iter_backup_source_blob_artifact_chunks( + container_client, + artifact_path, + transfer_format, + fernet=None, + source_version="", +): + data = _download_backup_artifact_bytes(container_client, artifact_path) + if transfer_format != DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT: + yield data + return + if fernet is None: + raise DataManagementSettingsValidationError("Encrypted source blob artifact requires the backup encryption key.") + if not data.startswith(DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC): + raise DataManagementSettingsValidationError("Encrypted source blob artifact has an invalid header.") + offset = len(DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_MAGIC) + expected_digest = hashlib.sha256(_safe_text(source_version).encode("utf-8")).digest() + expected_index = 0 + total_chunks = None + while offset < len(data): + if offset + 4 > len(data): + raise DataManagementSettingsValidationError("Encrypted source blob artifact is truncated.") + token_length = struct.unpack(">I", data[offset:offset + 4])[0] + offset += 4 + token = data[offset:offset + token_length] + offset += token_length + authenticated_chunk = fernet.decrypt(token) + if len(authenticated_chunk) < 49: + raise DataManagementSettingsValidationError("Encrypted source blob chunk is invalid.") + source_digest = authenticated_chunk[:32] + chunk_index, chunk_total, is_last = struct.unpack(">QQ?", authenticated_chunk[32:49]) + if source_digest != expected_digest: + raise DataManagementSettingsValidationError("Encrypted source blob source version did not match the manifest.") + if chunk_index != expected_index: + raise DataManagementSettingsValidationError("Encrypted source blob chunks are out of order.") + if total_chunks is None: + total_chunks = chunk_total + elif chunk_total != total_chunks: + raise DataManagementSettingsValidationError("Encrypted source blob chunk count changed.") + if bool(is_last) != (chunk_index == chunk_total - 1): + raise DataManagementSettingsValidationError("Encrypted source blob final chunk marker is invalid.") + expected_index += 1 + yield authenticated_chunk[49:] + if total_chunks is None or expected_index != total_chunks: + raise DataManagementSettingsValidationError("Encrypted source blob artifact ended before all chunks were read.") + + +def _iter_backup_manifest_entries(backup_job_id, service=None, resource_name=None, statuses=None): + query = ( + "SELECT * FROM c WHERE c.job_id = @job_id AND c.type = @type " + "ORDER BY c.created_at ASC" + ) + parameters = [ + {"name": "@job_id", "value": _safe_text(backup_job_id)}, + {"name": "@type", "value": DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_TYPE}, + ] + entries_seen = 0 + allowed_statuses = { + _safe_text(status).lower() + for status in (statuses or {"succeeded", "skipped"}) + if _safe_text(status) + } + batches = cosmos_data_management_job_items_container.query_items( + query=query, + parameters=parameters, + partition_key=_safe_text(backup_job_id), + max_item_count=DATA_MANAGEMENT_BACKUP_MANIFEST_BATCH_SIZE, + ) + for batch in batches: + if resource_name and _safe_text(batch.get("resource_name")) != _safe_text(resource_name): + continue + for entry in batch.get("entries") or []: + if not isinstance(entry, dict): + continue + if service and _safe_text(entry.get("service")) != _safe_text(service): + continue + if allowed_statuses and _safe_text(entry.get("status")).lower() not in allowed_statuses: + continue + if not _safe_text(entry.get("artifact_path")): + continue + entries_seen += 1 + if entries_seen > DATA_MANAGEMENT_RESTORE_MAX_MANIFEST_ENTRIES: + raise DataManagementSettingsValidationError( + "Backup manifest exceeds the supported restore entry limit." + ) + yield entry + + +def _get_restore_cosmos_artifact_by_resource(resource_name): + normalized_resource = _safe_text(resource_name) + for artifact in DATA_MANAGEMENT_COSMOS_ARTIFACTS: + if normalized_resource == f"cosmos:{artifact['name']}": + return artifact + return None + + +def _get_restore_search_artifact_by_resource(resource_name): + normalized_resource = _safe_text(resource_name) + for artifact in DATA_MANAGEMENT_SEARCH_ARTIFACTS: + if normalized_resource in { + f"ai_search:{artifact['index_name']}", + f"ai_search_schema:{artifact['index_name']}", + }: + return artifact + return None + + +def _build_restore_check(check_id, label, status, message, details=None): + return { + "id": check_id, + "label": label, + "status": status, + "message": _sanitize_data_management_backup_text(message), + "details": _sanitize_activity_value(details if isinstance(details, dict) else {}), + } + + +def _get_restore_backup_artifact_context(settings, backup_job): + backup_plan = backup_job.get("backup_plan") if isinstance(backup_job.get("backup_plan"), dict) else {} + _assert_backup_execution_settings(settings, backup_plan) + container_client = _get_backup_container_client(settings) + fernet = _get_restore_fernet(settings, backup_plan) + return container_client, fernet + + +def _summarize_restore_manifest(manifest, backup_job): + artifacts = manifest.get("artifacts") if isinstance(manifest.get("artifacts"), list) else [] + service_counts = {"cosmos": 0, "ai_search": 0, "source_blobs": 0} + for artifact in artifacts: + artifact_type = _safe_text((artifact or {}).get("type")) + if artifact_type == "cosmos_container": + service_counts["cosmos"] += 1 + elif artifact_type in {"ai_search_schema", "ai_search_documents"}: + service_counts["ai_search"] += 1 + elif artifact_type == "source_blob_container": + service_counts["source_blobs"] += 1 + return { + "backup_id": backup_job.get("id"), + "backup_type": backup_job.get("backup_type"), + "completed_at": backup_job.get("completed_at"), + "artifact_count": len(artifacts), + "service_counts": service_counts, + "warnings": len(manifest.get("warnings") or []), + "failed_resource_names": list(manifest.get("failed_resource_names") or []), + "differential_mode": _safe_text(manifest.get("differential_mode")), + "deletion_policy": _safe_text(manifest.get("deletion_policy")), + } + + +def _target_cosmos_container_has_records(container): + try: + result = next(iter(container.query_items( + query="SELECT TOP 1 VALUE c.id FROM c", + enable_cross_partition_query=True, + max_item_count=1, + )), None) + return result is not None + except Exception: + return True + + +def _target_search_index_has_documents(search_client): + try: + result = next(iter(search_client.search( + search_text="*", + top=1, + select=["id"], + include_total_count=False, + connection_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + read_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + retry_total=0, + )), None) + return result is not None + except ResourceNotFoundError: + return False + + +def _run_data_management_restore_preflight(settings, restore_plan, backup_job, manifest): + checks = [] + blockers = [] + warnings = [] + policy = _safe_text(restore_plan.get("restore_policy"), DATA_MANAGEMENT_RESTORE_POLICY_CREATE_ONLY) + overwrite = policy == DATA_MANAGEMENT_RESTORE_POLICY_OVERWRITE + failed_resources = list(manifest.get("failed_resource_names") or []) + + if failed_resources: + blockers.append("Backup has failed resources that must be retried before restore.") + checks.append(_build_restore_check( + "manifest_integrity", + "Backup manifest integrity", + "block", + "Backup has failed resources that are not restore-safe.", + {"failed_resource_count": len(failed_resources)}, + )) + else: + checks.append(_build_restore_check( + "manifest_integrity", + "Backup manifest integrity", + "pass", + "Backup manifest and resource outcomes are restore-safe.", + )) + + if restore_plan.get("differential_mode") == "latest_item_state": + warnings.append("Partial backups restore the latest captured changed/unchanged items but do not apply deletions.") + checks.append(_build_restore_check( + "partial_semantics", + "Partial backup semantics", + "warn", + "Partial backups are non-destructive and do not replay deletions.", + )) + else: + checks.append(_build_restore_check( + "partial_semantics", + "Partial backup semantics", + "pass", + "Full backup snapshot semantics are available.", + )) + + if restore_plan.get("include_cosmos"): + try: + database = _get_existing_target_cosmos_database(settings) + cosmos_collisions = [] + for artifact in DATA_MANAGEMENT_COSMOS_ARTIFACTS: + container_name = getattr(app_config, artifact["container_name_attr"], artifact["name"]) + container = database.get_container_client(container_name) + try: + container.read() + _validate_target_cosmos_container_partition_key( + container, + container_name, + artifact["partition_key_path"], + ) + if not overwrite and _target_cosmos_container_has_records(container): + cosmos_collisions.append(container_name) + except (CosmosResourceNotFoundError, ResourceNotFoundError): + continue + if cosmos_collisions: + blockers.append("Destination Cosmos contains records and restore policy is create-only.") + checks.append(_build_restore_check( + "target_cosmos", + "Target Cosmos DB", + "block", + "Existing destination Cosmos records require overwrite confirmation.", + {"collision_container_count": len(cosmos_collisions)}, + )) + else: + checks.append(_build_restore_check( + "target_cosmos", + "Target Cosmos DB", + "pass", + "Target Cosmos DB is reachable and partition keys are compatible.", + )) + except Exception as exc: + blockers.append("Target Cosmos DB could not be validated.") + checks.append(_build_restore_check( + "target_cosmos", + "Target Cosmos DB", + "block", + "Target Cosmos DB validation failed.", + {"error_type": type(exc).__name__}, + )) + + if restore_plan.get("include_ai_search"): + try: + endpoint = _safe_text(settings.get("target_ai_search_endpoint")) + if not endpoint: + raise DataManagementSettingsValidationError("Target Search endpoint is required.") + index_client = SearchIndexClient( + endpoint=endpoint, + credential=_get_target_ai_search_credential(settings), + connection_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + read_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + retry_total=0, + ) + search_collisions = [] + for artifact in DATA_MANAGEMENT_SEARCH_ARTIFACTS: + try: + index_client.get_index(artifact["index_name"]) + except ResourceNotFoundError: + continue + search_client = _get_target_search_client(settings, artifact["index_name"]) + if not overwrite and _target_search_index_has_documents(search_client): + search_collisions.append(artifact["index_name"]) + if search_collisions: + blockers.append("Destination Search contains documents and restore policy is create-only.") + checks.append(_build_restore_check( + "target_ai_search", + "Target AI Search", + "block", + "Existing destination Search documents require overwrite confirmation.", + {"collision_index_count": len(search_collisions)}, + )) + else: + checks.append(_build_restore_check( + "target_ai_search", + "Target AI Search", + "pass", + "Target AI Search is reachable and ready for restore.", + )) + except Exception as exc: + blockers.append("Target AI Search could not be validated.") + checks.append(_build_restore_check( + "target_ai_search", + "Target AI Search", + "block", + "Target AI Search validation failed.", + {"error_type": type(exc).__name__}, + )) + + if restore_plan.get("include_source_blobs"): + try: + target_client = _get_target_enhanced_citations_blob_client(settings) + container_names = [ + _safe_text(resource_name).split("source_blobs:", 1)[1] + for resource_name in { + _safe_text(entry.get("resource_name")) + for entry in _iter_backup_manifest_entries( + backup_job.get("id"), + service="source_blobs", + ) + } + if _safe_text(resource_name).startswith("source_blobs:") + ] + for container_name in sorted(set(container_names)): + container = target_client.get_container_client(container_name) + if container.exists(): + continue + checks.append(_build_restore_check( + "target_source_blobs", + "Target Enhanced Citation Storage", + "pass", + "Target Enhanced Citation Storage is reachable.", + {"container_count": len(set(container_names))}, + )) + except Exception as exc: + blockers.append("Target Enhanced Citation Storage could not be validated.") + checks.append(_build_restore_check( + "target_source_blobs", + "Target Enhanced Citation Storage", + "block", + "Target Enhanced Citation Storage validation failed.", + {"error_type": type(exc).__name__}, + )) + + if overwrite: + checks.append(_build_restore_check( + "destructive_policy", + "Restore overwrite policy", + "warn", + "Overwrite restore is explicitly confirmed and can replace existing target data.", + )) + else: + checks.append(_build_restore_check( + "destructive_policy", + "Restore overwrite policy", + "pass", + "Create-only restore is non-destructive and blocks existing target collisions.", + )) + + return { + "ready": not blockers, + "blocker_count": len(blockers), + "warning_count": len(warnings), + "blockers": blockers, + "warnings": warnings, + "checks": checks, + } + + +def review_data_management_restore(settings=None, restore_plan=None): + """Run a sanitized restore preflight without mutating restore targets.""" + normalized_settings = _normalize_data_management_settings_from_payload(settings) + normalized_plan = _normalize_data_management_restore_plan( + restore_plan if isinstance(restore_plan, dict) else {}, + ) + backup_job = _get_restore_backup_job(normalized_plan["source_backup_id"]) + container_client, fernet = _get_restore_backup_artifact_context( + normalized_settings, + backup_job, + ) + manifest = _read_backup_json_artifact( + container_client, + normalized_plan["source_backup_manifest_path"], + fernet=fernet, + ) + if manifest.get("schema_version") != 2 or manifest.get("app") != "SimpleChat": + raise DataManagementSettingsValidationError("Selected backup manifest is not supported for restore.") + if _safe_text(manifest.get("job_id")) != _safe_text(backup_job.get("id")): + raise DataManagementSettingsValidationError("Selected backup manifest does not match the backup job.") + + preflight = _run_data_management_restore_preflight( + normalized_settings, + normalized_plan, + backup_job, + manifest, + ) + fingerprint = _restore_review_fingerprint(normalized_settings, normalized_plan) + return { + **preflight, + "review_fingerprint": fingerprint, + "summary": _summarize_restore_manifest(manifest, backup_job), + "restore_plan": { + key: value + for key, value in normalized_plan.items() + if key != "backup_encryption_key_reference" + }, + } + + +def queue_data_management_job(operation, backup_type=None, requested_by=None, requested_by_email=None, options=None, scheduled=False, occurrence_id=None): + normalized_operation = _safe_text(operation) + if normalized_operation not in DATA_MANAGEMENT_OPERATIONS: + raise ValueError("Unsupported data management operation.") + + normalized_backup_type = _safe_text(backup_type) + if normalized_operation == DATA_MANAGEMENT_OPERATION_BACKUP and normalized_backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: + raise ValueError("Backup jobs must be full or partial.") + if normalized_operation != DATA_MANAGEMENT_OPERATION_BACKUP and normalized_backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: + normalized_backup_type = None + + now = _now_iso() + job_id = occurrence_id or str(uuid.uuid4()) + backup_plan = None + backup_state = None + restore_plan = None + restore_state = None + source_scope = None + source_cutoff_at = None + normalized_options = options if isinstance(options, dict) else {} + if normalized_operation == DATA_MANAGEMENT_OPERATION_BACKUP: + settings = get_data_management_settings() + if settings.get("encryption_enabled") and not settings.get("encryption_key_reference"): + generate_data_management_encryption_key() + settings = get_data_management_settings() + backup_plan = _normalize_data_management_backup_plan( + settings, + normalized_backup_type, + normalized_options, + source_cutoff_at=now, + ) + source_scope = backup_plan["source_scope"] + source_cutoff_at = backup_plan["source_cutoff_at"] + backup_state = initialize_backup_state( + None, + job_id, + backup_plan, + source_scope, + source_cutoff_at, + ) + elif normalized_operation == DATA_MANAGEMENT_OPERATION_RESTORE: + source_backup = _get_restore_backup_job( + (normalized_options.get("restore_plan") or {}).get("source_backup_id") + if isinstance(normalized_options.get("restore_plan"), dict) + else normalized_options.get("source_backup_id") + ) + restore_plan = _normalize_data_management_restore_plan( + normalized_options, + backup_job=source_backup, + require_confirmation=True, + ) + restore_state = initialize_restore_state( + None, + job_id, + restore_plan, + ) + job = { + "id": job_id, + "type": DATA_MANAGEMENT_JOB_TYPE, + "operation": normalized_operation, + "backup_type": normalized_backup_type, + "status": DATA_MANAGEMENT_STATUS_QUEUED, + "created_at": now, + "updated_at": now, + "started_at": None, + "completed_at": None, + "last_heartbeat_at": None, + "last_message": "Queued data management job", + "last_error": None, + "cancel_requested_at": None, + "cancel_requested_by": None, + "cancel_requested_by_email": None, + "cancel_reason": None, + "requested_by": _safe_text(requested_by), + "requested_by_email": _safe_text(requested_by_email), + "scheduled": bool(scheduled), + "occurrence_id": occurrence_id, + "options": normalized_options, + "progress": { + "total_steps": 0, + "completed_steps": 0, + "current_step": None, + "percent_complete": 0, + }, + "lease_holder_id": None, + "lease_expires_at": None, + "lease_generation": 0, + "warnings": [], + "source_scope": source_scope, + "source_cutoff_at": source_cutoff_at, + "backup_plan": backup_plan, + "backup_state": backup_state, + "restore_plan": restore_plan, + "restore_state": restore_state, + } + + try: + saved_job = cosmos_data_management_jobs_container.create_item(body=job) + _record_data_management_job_event( + saved_job.get("id"), + "queued", + saved_job, + status=DATA_MANAGEMENT_STATUS_QUEUED, + message="Queued data management job", + details={"operation": normalized_operation, "backup_type": normalized_backup_type, "scheduled": bool(scheduled)}, + ) + return saved_job + except Exception as exc: + if occurrence_id and getattr(exc, "status_code", None) == 409: + return _read_job(job_id) + raise + + +def _normalize_data_management_history_datetime(value, field_name, end_exclusive=False): + text = _safe_text(value) + if not text: + return None + date_only = bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", text)) + parsed = _parse_iso_datetime(text) + if parsed is None: + raise DataManagementHistoryPaginationError( + f"{field_name} must be an ISO-8601 date or timestamp." + ) + if date_only and end_exclusive: + parsed += timedelta(days=1) + return parsed.isoformat() + + +def normalize_data_management_history_filters(list_kind, filters=None): + normalized_kind = _safe_text(list_kind).lower() + if normalized_kind not in {"jobs", "backups"}: + raise DataManagementHistoryPaginationError("Unsupported Data Management history list.") + + raw_filters = filters if isinstance(filters, dict) else {} + status = _safe_text(raw_filters.get("status")).lower() + allowed_statuses = { + DATA_MANAGEMENT_STATUS_QUEUED, + DATA_MANAGEMENT_STATUS_RUNNING, + DATA_MANAGEMENT_STATUS_COMPLETED, + DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS, + DATA_MANAGEMENT_STATUS_FAILED, + DATA_MANAGEMENT_STATUS_CANCELED, + } + if normalized_kind == "backups": + allowed_statuses.add("available") + if status and status not in allowed_statuses: + raise DataManagementHistoryPaginationError("Unsupported Data Management status filter.") + + scheduled_value = _safe_text(raw_filters.get("scheduled")).lower() + if scheduled_value in {"", "all"}: + scheduled = None + elif scheduled_value == "scheduled": + scheduled = True + elif scheduled_value == "manual": + scheduled = False + else: + raise DataManagementHistoryPaginationError("Scheduled filter must be all, scheduled, or manual.") + + created_from = _normalize_data_management_history_datetime( + raw_filters.get("created_from"), + "created_from", + ) + created_to = _normalize_data_management_history_datetime( + raw_filters.get("created_to"), + "created_to", + end_exclusive=True, + ) + if bool(created_from) != bool(created_to): + raise DataManagementHistoryPaginationError( + "created_from and created_to must be provided together." + ) + if created_from and created_to: + start = _parse_iso_datetime(created_from) + end = _parse_iso_datetime(created_to) + if end <= start: + raise DataManagementHistoryPaginationError("created_to must be after created_from.") + if end - start > timedelta(days=DATA_MANAGEMENT_HISTORY_MAX_DATE_RANGE_DAYS): + raise DataManagementHistoryPaginationError( + f"Date range cannot exceed {DATA_MANAGEMENT_HISTORY_MAX_DATE_RANGE_DAYS} days." + ) + + operation = None + backup_type = None + if normalized_kind == "jobs": + operation = _safe_text(raw_filters.get("operation")).lower() + if operation and operation not in DATA_MANAGEMENT_OPERATIONS: + raise DataManagementHistoryPaginationError("Unsupported Data Management operation filter.") + else: + backup_type = _safe_text(raw_filters.get("backup_type")).lower() + if backup_type and backup_type not in DATA_MANAGEMENT_BACKUP_TYPES: + raise DataManagementHistoryPaginationError("Backup type filter must be full or partial.") + + return { + "operation": operation or None, + "backup_type": backup_type or None, + "status": status or None, + "scheduled": scheduled, + "created_from": created_from, + "created_to": created_to, + } + + +def _get_data_management_history_cipher(): + secret = _safe_text(getattr(app_config, "SECRET_KEY", "")) + derived_key = hashlib.sha256( + f"data-management-history:{secret}".encode("utf-8") + ).digest() + return Fernet(base64.urlsafe_b64encode(derived_key)) + + +def _encode_data_management_history_token(list_kind, filters, page_size, cursor): + if not isinstance(cursor, dict): + return None + cursor_created_at = _safe_text(cursor.get("created_at")) + cursor_id = _safe_text(cursor.get("id")) + if not cursor_created_at or not cursor_id: + return None + payload = { + "version": DATA_MANAGEMENT_HISTORY_TOKEN_VERSION, + "list_kind": list_kind, + "filters": filters, + "page_size": page_size, + "sort": DATA_MANAGEMENT_HISTORY_SORT, + "cursor": { + "created_at": cursor_created_at, + "id": cursor_id, + }, + } + token = _get_data_management_history_cipher().encrypt( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ) + return token.decode("utf-8") + + +def _decode_data_management_history_token(token, list_kind, filters, page_size): + safe_token = _safe_text(token) + if not safe_token: + return None + try: + decoded_token = base64.b64decode( + safe_token.encode("utf-8"), + altchars=b"-_", + validate=True, + ) + canonical_token = base64.urlsafe_b64encode(decoded_token).decode("utf-8") + if not hmac.compare_digest(canonical_token, safe_token): + raise ValueError("Continuation token is not canonical.") + decrypted = _get_data_management_history_cipher().decrypt( + safe_token.encode("utf-8"), + ttl=DATA_MANAGEMENT_HISTORY_TOKEN_TTL_SECONDS, ) payload = json.loads(decrypted.decode("utf-8")) except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError, TypeError, ValueError) as exc: @@ -8168,6 +9051,7 @@ def get_recoverable_data_management_jobs( for operation in (operations or { DATA_MANAGEMENT_OPERATION_BACKUP, DATA_MANAGEMENT_OPERATION_MIGRATION, + DATA_MANAGEMENT_OPERATION_RESTORE, }) } for job in candidates: @@ -8211,7 +9095,7 @@ def get_recoverable_data_management_migration_jobs( def recover_data_management_jobs(app=None, settings=None, current_time=None, operations=None): - """Resubmit recoverable backup and migration jobs through the executor, never inline.""" + """Resubmit recoverable backup, restore, and migration jobs through the executor, never inline.""" now = current_time if isinstance(current_time, datetime) else _now_utc() recovery_results = [] for recovery in get_recoverable_data_management_jobs( @@ -8221,7 +9105,11 @@ def recover_data_management_jobs(app=None, settings=None, current_time=None, ope job = recovery["job"] reason = recovery["reason"] operation = _safe_text(job.get("operation")) - operation_label = "backup" if operation == DATA_MANAGEMENT_OPERATION_BACKUP else "migration" + operation_label = ( + "backup" if operation == DATA_MANAGEMENT_OPERATION_BACKUP else + "restore" if operation == DATA_MANAGEMENT_OPERATION_RESTORE else + "migration" + ) job.update({ "last_recovery_submitted_at": now.isoformat(), "recovery_attempt_count": _safe_int(job.get("recovery_attempt_count"), default=0, minimum=0) + 1, @@ -8559,6 +9447,61 @@ def sanitize_item_summary(summary): } +def _sanitize_data_management_restore_state_for_admin(state): + """Expose durable restore progress without backup contents or credentials.""" + if not isinstance(state, dict): + return {} + resources = {} + for resource_name, resource in (state.get("resources") or {}).items(): + if not isinstance(resource, dict): + continue + progress = resource.get("progress") if isinstance(resource.get("progress"), dict) else {} + result = resource.get("result") if isinstance(resource.get("result"), dict) else {} + resources[_safe_text(resource_name)] = { + "status": resource.get("status"), + "phase": resource.get("phase"), + "attempts": _safe_int(resource.get("attempts"), default=0, minimum=0), + "started_at": resource.get("started_at"), + "attempt_started_at": resource.get("attempt_started_at"), + "updated_at": resource.get("updated_at"), + "completed_at": resource.get("completed_at"), + "last_error": _sanitize_data_management_backup_text(resource.get("last_error")), + "progress": _sanitize_activity_value(progress), + "result": _sanitize_activity_value(result), + } + totals = state.get("totals") if isinstance(state.get("totals"), dict) else {} + attempt_history = [] + for attempt in (state.get("attempt_history") or [])[-DATA_MANAGEMENT_BACKUP_MAX_RECENT_CHECKPOINTS:]: + if not isinstance(attempt, dict): + continue + attempt_history.append({ + "attempt_id": _safe_text(attempt.get("attempt_id")), + "lease_generation": _safe_int(attempt.get("lease_generation"), default=0, minimum=0), + "started_at": attempt.get("started_at"), + "completed_at": attempt.get("completed_at"), + "outcome": _safe_text(attempt.get("outcome")), + }) + return { + "schema_version": state.get("schema_version"), + "restore_id": state.get("restore_id"), + "status": state.get("status"), + "phase": _sanitize_data_management_backup_text(state.get("phase")), + "created_at": state.get("created_at"), + "updated_at": state.get("updated_at"), + "completed_at": state.get("completed_at"), + "last_progress_at": state.get("last_progress_at"), + "resume_count": _safe_int(state.get("resume_count"), default=0, minimum=0), + "preflight": _sanitize_activity_value(state.get("preflight") if isinstance(state.get("preflight"), dict) else {}), + "totals": _sanitize_activity_value(totals), + "warnings": [ + _sanitize_data_management_backup_text(warning) + for warning in (state.get("warnings") or [])[-DATA_MANAGEMENT_BACKUP_MAX_PUBLIC_ITEM_SUMMARIES:] + ], + "attempt_history": attempt_history, + "resources": resources, + } + + def sanitize_data_management_job_for_admin(job): if not isinstance(job, dict): return None @@ -8579,6 +9522,18 @@ def sanitize_data_management_job_for_admin(job): result["backup_state"] = _sanitize_data_management_backup_state_for_admin( job.get("backup_state") ) + if isinstance(result, dict) and job.get("operation") == DATA_MANAGEMENT_OPERATION_RESTORE: + if isinstance(result.get("artifacts"), list): + result["artifacts"] = summarize_backup_artifacts(result.get("artifacts")) + if isinstance(result.get("warnings"), list): + result["warnings"] = [ + _sanitize_data_management_backup_text(warning) + for warning in result.get("warnings", [])[-DATA_MANAGEMENT_BACKUP_MAX_PUBLIC_ITEM_SUMMARIES:] + ] + if "restore_state" in result: + result["restore_state"] = _sanitize_data_management_restore_state_for_admin( + job.get("restore_state") + ) return { "id": job.get("id"), "operation": job.get("operation"), @@ -8603,10 +9558,12 @@ def sanitize_data_management_job_for_admin(job): "result": result, "migration_state": _sanitize_data_management_migration_state_for_admin(job.get("migration_state")), "backup_state": _sanitize_data_management_backup_state_for_admin(job.get("backup_state")), + "restore_state": _sanitize_data_management_restore_state_for_admin(job.get("restore_state")), "can_retry": bool( job.get("operation") in { DATA_MANAGEMENT_OPERATION_MIGRATION, DATA_MANAGEMENT_OPERATION_BACKUP, + DATA_MANAGEMENT_OPERATION_RESTORE, } and ( job.get("status") in {DATA_MANAGEMENT_STATUS_FAILED, DATA_MANAGEMENT_STATUS_CANCELED} or @@ -8626,6 +9583,7 @@ def sanitize_data_management_job_for_admin(job): job.get("operation") in { DATA_MANAGEMENT_OPERATION_MIGRATION, DATA_MANAGEMENT_OPERATION_BACKUP, + DATA_MANAGEMENT_OPERATION_RESTORE, } and job.get("status") in {DATA_MANAGEMENT_STATUS_QUEUED, DATA_MANAGEMENT_STATUS_RUNNING} and not job.get("cancel_requested_at") @@ -8670,8 +9628,12 @@ def request_data_management_job_cancellation( if not job: raise DataManagementSettingsValidationError("Data Management job was not found.") operation = _safe_text(job.get("operation")) - if operation not in {DATA_MANAGEMENT_OPERATION_MIGRATION, DATA_MANAGEMENT_OPERATION_BACKUP}: - raise DataManagementSettingsValidationError("Only queued or running backup and migration jobs can be canceled.") + if operation not in { + DATA_MANAGEMENT_OPERATION_MIGRATION, + DATA_MANAGEMENT_OPERATION_BACKUP, + DATA_MANAGEMENT_OPERATION_RESTORE, + }: + raise DataManagementSettingsValidationError("Only queued or running backup, restore, and migration jobs can be canceled.") if job.get("status") == DATA_MANAGEMENT_STATUS_CANCELED: return job if job.get("status") in DATA_MANAGEMENT_TERMINAL_STATUSES: @@ -8695,6 +9657,14 @@ def request_data_management_job_cancellation( "cancel_reason": safe_reason, "updated_at": now, }) + restore_state = job.get("restore_state") + if isinstance(restore_state, dict): + restore_state.update({ + "cancel_requested_at": now, + "cancel_requested_by": _safe_text(requested_by), + "cancel_reason": safe_reason, + "updated_at": now, + }) job.update({ "cancel_requested_at": now, @@ -8702,17 +9672,22 @@ def request_data_management_job_cancellation( "cancel_requested_by_email": _safe_text(requested_by_email), "cancel_reason": safe_reason, "updated_at": now, - "last_message": "Migration cancellation requested. The worker will stop at its next durable checkpoint.", + "last_message": "Data Management job cancellation requested. The worker will stop at its next durable checkpoint.", "migration_state": migration_state, "backup_state": backup_state, + "restore_state": restore_state, }) - noun = "Backup" if operation == DATA_MANAGEMENT_OPERATION_BACKUP else "Migration" + noun = ( + "Backup" if operation == DATA_MANAGEMENT_OPERATION_BACKUP else + "Restore" if operation == DATA_MANAGEMENT_OPERATION_RESTORE else + "Migration" + ) if job.get("status") == DATA_MANAGEMENT_STATUS_QUEUED: job.update({ "status": DATA_MANAGEMENT_STATUS_CANCELED, "completed_at": now, "last_heartbeat_at": now, - "last_message": "Queued migration canceled before execution started.", + "last_message": f"Queued {operation} canceled before execution started.", "lease_holder_id": None, "lease_expires_at": None, }) @@ -8730,6 +9705,14 @@ def request_data_management_job_cancellation( "updated_at": now, }) complete_backup_attempt(backup_state, "canceled") + if isinstance(restore_state, dict): + restore_state.update({ + "status": DATA_MANAGEMENT_STATUS_CANCELED, + "phase": "canceled", + "completed_at": now, + "updated_at": now, + }) + complete_restore_attempt(restore_state, "canceled") event_name = f"{operation}-canceled" event_message = f"Queued {operation} canceled before execution started." else: @@ -8742,6 +9725,12 @@ def request_data_management_job_cancellation( if job.get("status") == DATA_MANAGEMENT_STATUS_CANCELED else "Backup cancellation requested. The worker will stop at its next durable checkpoint." ) + elif operation == DATA_MANAGEMENT_OPERATION_RESTORE: + job["last_message"] = ( + "Queued restore canceled before execution started." + if job.get("status") == DATA_MANAGEMENT_STATUS_CANCELED else + "Restore cancellation requested. The worker will stop at its next durable checkpoint." + ) saved_job = _save_data_management_job(job) _record_data_management_job_event( @@ -8887,31 +9876,112 @@ def retry_data_management_backup_job(job_id): retryable_statuses.add(DATA_MANAGEMENT_STATUS_RUNNING) if job.get("status") not in retryable_statuses: raise DataManagementSettingsValidationError( - "Only failed, canceled, or stale backup jobs can be retried." + "Only failed, canceled, or stale backup jobs can be retried." + ) + backup_plan = job.get("backup_plan") + canceled_before_start = ( + job.get("status") == DATA_MANAGEMENT_STATUS_CANCELED and + not isinstance(backup_state, dict) + ) + if not canceled_before_start and ( + not isinstance(backup_plan, dict) or + not isinstance(backup_state, dict) or + _safe_text(backup_state.get("backup_id")) != _safe_text(job.get("id")) + ): + raise DataManagementSettingsValidationError( + "This backup does not have durable checkpoint state to retry." + ) + + now = _now_iso() + if isinstance(backup_state, dict): + backup_state["last_cancellation"] = { + "requested_at": backup_state.get("cancel_requested_at"), + "requested_by": backup_state.get("cancel_requested_by"), + "reason": backup_state.get("cancel_reason"), + } + backup_state.update({ + "status": BACKUP_RESOURCE_STATUS_PENDING, + "phase": "queued", + "updated_at": now, + "completed_at": None, + "cancel_requested_at": None, + "cancel_requested_by": None, + "cancel_reason": None, + }) + job.update({ + "status": DATA_MANAGEMENT_STATUS_QUEUED, + "updated_at": now, + "completed_at": None, + "last_heartbeat_at": None, + "last_message": "Backup retry queued from durable checkpoints", + "last_error": None, + "last_cancellation": { + "requested_at": job.get("cancel_requested_at"), + "requested_by": job.get("cancel_requested_by"), + "requested_by_email": job.get("cancel_requested_by_email"), + "reason": job.get("cancel_reason"), + }, + "cancel_requested_at": None, + "cancel_requested_by": None, + "cancel_requested_by_email": None, + "cancel_reason": None, + "lease_holder_id": None, + "lease_expires_at": None, + "backup_source_lock": None, + "backup_state": backup_state, + "result": {}, + }) + saved_job = _save_data_management_job(job) + _record_data_management_job_event( + saved_job.get("id"), + "backup-retry-queued", + saved_job, + status=DATA_MANAGEMENT_STATUS_QUEUED, + message="Backup retry queued from durable checkpoints", + details={ + "source_cutoff_at": (backup_plan or {}).get("source_cutoff_at"), + "resume_count": (backup_state or {}).get("resume_count", 0), + }, + ) + return saved_job + + +def retry_data_management_restore_job(job_id): + """Requeue a restore from durable checkpoints without changing its immutable plan.""" + job = get_data_management_job(job_id) + if not job or job.get("operation") != DATA_MANAGEMENT_OPERATION_RESTORE: + raise DataManagementSettingsValidationError("Data Management restore job was not found.") + retryable_statuses = {DATA_MANAGEMENT_STATUS_FAILED, DATA_MANAGEMENT_STATUS_CANCELED} + if job.get("status") == DATA_MANAGEMENT_STATUS_RUNNING and _is_stale_job(job): + retryable_statuses.add(DATA_MANAGEMENT_STATUS_RUNNING) + if job.get("status") not in retryable_statuses: + raise DataManagementSettingsValidationError( + "Only failed, canceled, or stale restore jobs can be retried." ) - backup_plan = job.get("backup_plan") + restore_plan = job.get("restore_plan") + restore_state = job.get("restore_state") canceled_before_start = ( job.get("status") == DATA_MANAGEMENT_STATUS_CANCELED and - not isinstance(backup_state, dict) + not isinstance(restore_state, dict) ) if not canceled_before_start and ( - not isinstance(backup_plan, dict) or - not isinstance(backup_state, dict) or - _safe_text(backup_state.get("backup_id")) != _safe_text(job.get("id")) + not isinstance(restore_plan, dict) or + not isinstance(restore_state, dict) or + _safe_text(restore_state.get("restore_id")) != _safe_text(job.get("id")) ): raise DataManagementSettingsValidationError( - "This backup does not have durable checkpoint state to retry." + "This restore does not have durable checkpoint state to retry." ) now = _now_iso() - if isinstance(backup_state, dict): - backup_state["last_cancellation"] = { - "requested_at": backup_state.get("cancel_requested_at"), - "requested_by": backup_state.get("cancel_requested_by"), - "reason": backup_state.get("cancel_reason"), + if isinstance(restore_state, dict): + restore_state["last_cancellation"] = { + "requested_at": restore_state.get("cancel_requested_at"), + "requested_by": restore_state.get("cancel_requested_by"), + "reason": restore_state.get("cancel_reason"), } - backup_state.update({ - "status": BACKUP_RESOURCE_STATUS_PENDING, + restore_state.update({ + "status": DATA_MANAGEMENT_STATUS_QUEUED, "phase": "queued", "updated_at": now, "completed_at": None, @@ -8924,7 +9994,7 @@ def retry_data_management_backup_job(job_id): "updated_at": now, "completed_at": None, "last_heartbeat_at": None, - "last_message": "Backup retry queued from durable checkpoints", + "last_message": "Restore retry queued from durable checkpoints", "last_error": None, "last_cancellation": { "requested_at": job.get("cancel_requested_at"), @@ -8938,20 +10008,19 @@ def retry_data_management_backup_job(job_id): "cancel_reason": None, "lease_holder_id": None, "lease_expires_at": None, - "backup_source_lock": None, - "backup_state": backup_state, + "restore_state": restore_state, "result": {}, }) saved_job = _save_data_management_job(job) _record_data_management_job_event( saved_job.get("id"), - "backup-retry-queued", + "restore-retry-queued", saved_job, status=DATA_MANAGEMENT_STATUS_QUEUED, - message="Backup retry queued from durable checkpoints", + message="Restore retry queued from durable checkpoints", details={ - "source_cutoff_at": (backup_plan or {}).get("source_cutoff_at"), - "resume_count": (backup_state or {}).get("resume_count", 0), + "source_backup_id": (restore_plan or {}).get("source_backup_id"), + "resume_count": (restore_state or {}).get("resume_count", 0), }, ) return saved_job @@ -10496,6 +11565,8 @@ def _try_claim_data_management_job(job_id, settings=None): job["migration_attempt_id"] = str(uuid.uuid4()) elif job.get("operation") == DATA_MANAGEMENT_OPERATION_BACKUP: job["backup_attempt_id"] = str(uuid.uuid4()) + elif job.get("operation") == DATA_MANAGEMENT_OPERATION_RESTORE: + job["restore_attempt_id"] = str(uuid.uuid4()) try: claimed_job = _replace_job(job) if claimed_job.get("operation") == DATA_MANAGEMENT_OPERATION_BACKUP: @@ -15348,49 +16419,780 @@ def _execute_backup_cosmos_resources( _append_backup_warning(backup_state, warning) _fail_backup_resource_checkpoint( job, - backup_state, + backup_state, + settings, + resource_name, + "Cosmos backup resource failed.", + f"Recorded failed Cosmos backup resource {artifact['name']}", + ) + artifacts.append({ + "name": artifact["name"], + "type": "cosmos_container", + "status": "warning", + "warning": warning, + }) + finally: + source_capacity = ( + backup_state.get("source_capacity") + if isinstance(backup_state.get("source_capacity"), dict) else {} + ) + if capacity_attempted or source_capacity.get("restore_pending"): + try: + restore_warnings, _ = _restore_temporary_backup_source_capacity( + job, + backup_state, + settings, + allow_cancel_requested=True, + ) + except DataManagementBackupLeaseLostError: + raise + except Exception as exc: + restore_warnings = [ + f"Failed to restore temporary source Cosmos capacity: {str(exc)[:300]}" + ] + for warning in restore_warnings: + _append_backup_warning(backup_state, warning) + + source_capacity = ( + backup_state.get("source_capacity") + if isinstance(backup_state.get("source_capacity"), dict) else {} + ) + if source_capacity.get("restore_pending"): + raise RuntimeError( + "Source Cosmos capacity restoration remains pending. Retry the backup to restore the recorded capacity snapshot." + ) + return artifacts + + +def _append_restore_warning(state, warning): + """Keep restore warning text bounded in durable job state.""" + if not isinstance(state, dict): + return + warnings = state.setdefault("warnings", []) + if not isinstance(warnings, list): + warnings = [] + state["warnings"] = warnings + safe_warning = _sanitize_data_management_backup_text(warning) + if safe_warning: + warnings.append(safe_warning) + del warnings[:-DATA_MANAGEMENT_BACKUP_MAX_PUBLIC_ITEM_SUMMARIES] + + +def _update_restore_state_totals(state): + totals = { + "processed_count": 0, + "copied_count": 0, + "created_count": 0, + "updated_count": 0, + "skipped_count": 0, + "failed_count": 0, + "collision_count": 0, + "bytes": 0, + "checkpoint_count": 0, + } + for resource in (state.get("resources") or {}).values(): + if not isinstance(resource, dict): + continue + metrics = resource.get("progress") if isinstance(resource.get("progress"), dict) else {} + if isinstance(resource.get("result"), dict): + metrics = {**metrics, **resource["result"]} + totals["processed_count"] += _safe_int(metrics.get("processed_count"), default=0, minimum=0) + totals["copied_count"] += _safe_int(metrics.get("copied_count"), default=0, minimum=0) + totals["created_count"] += _safe_int(metrics.get("created_count"), default=0, minimum=0) + totals["updated_count"] += _safe_int(metrics.get("updated_count"), default=0, minimum=0) + totals["skipped_count"] += _safe_int(metrics.get("skipped_count"), default=0, minimum=0) + totals["failed_count"] += _safe_int(metrics.get("failed_count"), default=0, minimum=0) + totals["collision_count"] += _safe_int(metrics.get("collision_count"), default=0, minimum=0) + totals["bytes"] += _safe_int(metrics.get("bytes"), default=0, minimum=0) + totals["checkpoint_count"] += _safe_int(metrics.get("checkpoint_count"), default=0, minimum=0) + state["totals"] = totals + return totals + + +def _persist_restore_state(job, state, settings, message, allow_cancel_requested=False): + """Persist restore-level phase changes and renew the fenced job lease.""" + _assert_restore_job_lease(job, allow_cancel_requested=allow_cancel_requested) + _update_restore_state_totals(state) + now = _now_utc() + state["updated_at"] = now.isoformat() + job.update({ + "restore_state": state, + "updated_at": now.isoformat(), + "last_heartbeat_at": now.isoformat(), + "lease_expires_at": ( + now + timedelta(seconds=_get_data_management_job_lease_seconds(settings)) + ).isoformat(), + "last_message": _safe_text(message), + }) + _save_data_management_job(job) + saved_state = job.get("restore_state") if isinstance(job.get("restore_state"), dict) else state + if saved_state is not state: + state.clear() + state.update(copy.deepcopy(saved_state)) + return state + + +def _persist_restore_checkpoint(job, state, settings, resource_name, progress, checkpoint, message): + """Persist bounded restore resource progress.""" + _assert_restore_job_lease(job) + update_restore_resource(state, resource_name, progress, checkpoint=checkpoint) + _update_restore_state_totals(state) + now = _now_utc() + state["last_progress_at"] = now.isoformat() + job.update({ + "restore_state": state, + "updated_at": now.isoformat(), + "last_heartbeat_at": now.isoformat(), + "last_progress_at": now.isoformat(), + "lease_expires_at": ( + now + timedelta(seconds=_get_data_management_job_lease_seconds(settings)) + ).isoformat(), + "last_message": _safe_text(message), + }) + _save_data_management_job(job) + saved_state = job.get("restore_state") if isinstance(job.get("restore_state"), dict) else state + if saved_state is not state: + state.clear() + state.update(copy.deepcopy(saved_state)) + return state + + +def _complete_restore_resource_checkpoint(job, state, settings, resource_name, result, message): + _assert_restore_job_lease(job) + complete_restore_resource(state, resource_name, result=result) + _update_restore_state_totals(state) + now = _now_utc() + state["last_progress_at"] = now.isoformat() + job.update({ + "restore_state": state, + "updated_at": now.isoformat(), + "last_heartbeat_at": now.isoformat(), + "last_progress_at": now.isoformat(), + "lease_expires_at": ( + now + timedelta(seconds=_get_data_management_job_lease_seconds(settings)) + ).isoformat(), + "last_message": _safe_text(message), + }) + _save_data_management_job(job) + return state + + +def _fail_restore_resource_checkpoint(job, state, settings, resource_name, error_message, message, result=None): + _assert_restore_job_lease(job) + resource = fail_restore_resource(state, resource_name, _sanitize_data_management_backup_text(error_message)) + if isinstance(result, dict): + resource["result"] = copy.deepcopy(result) + _update_restore_state_totals(state) + now = _now_utc() + job.update({ + "restore_state": state, + "updated_at": now.isoformat(), + "last_heartbeat_at": now.isoformat(), + "last_progress_at": now.isoformat(), + "lease_expires_at": ( + now + timedelta(seconds=_get_data_management_job_lease_seconds(settings)) + ).isoformat(), + "last_message": _safe_text(message), + }) + _save_data_management_job(job) + return state + + +def _restore_policy_allows_overwrite(restore_plan): + return _safe_text(restore_plan.get("restore_policy")) == DATA_MANAGEMENT_RESTORE_POLICY_OVERWRITE + + +def _restore_cosmos_source_identity(document, artifact): + container_name = _safe_text(getattr(app_config, artifact["container_name_attr"], artifact["name"])) + return _build_backup_source_identity( + "cosmos", + container_name, + (document or {}).get("id"), + _get_document_path_value(document, artifact["partition_key_path"]), + ) + + +def _restore_search_source_identity(document, artifact): + return _build_backup_source_identity( + "ai_search", + artifact["index_name"], + (document or {}).get("id"), + ) + + +def _group_restore_entries_by_resource(backup_job_id, service): + grouped = {} + for entry in _iter_backup_manifest_entries(backup_job_id, service=service): + grouped.setdefault(_safe_text(entry.get("resource_name")), []).append(entry) + return grouped + + +def _iter_restore_records_from_entries( + container_client, + entries, + fernet, + identity_builder, +): + entries_by_artifact = {} + for entry in entries: + entries_by_artifact.setdefault(_safe_text(entry.get("artifact_path")), set()).add( + _safe_text(entry.get("source_identity")) + ) + yielded = set() + for artifact_path, expected_identities in entries_by_artifact.items(): + for record in _iter_backup_jsonl_artifact(container_client, artifact_path, fernet=fernet): + source_identity = _safe_text(identity_builder(record)) + if source_identity not in expected_identities or source_identity in yielded: + continue + yielded.add(source_identity) + yield record + + +def _execute_restore_cosmos_resources(job, state, settings, restore_plan, container_client, fernet): + if not restore_plan.get("include_cosmos"): + return [{ + "name": "cosmos", + "type": "cosmos_restore", + "status": "skipped", + "warning": "Cosmos DB restore is disabled for this job.", + }] + target_database = _get_target_cosmos_database(settings) + artifacts = [] + overwrite = _restore_policy_allows_overwrite(restore_plan) + grouped_entries = _group_restore_entries_by_resource(restore_plan["source_backup_id"], "cosmos") + for resource_name, entries in grouped_entries.items(): + artifact = _get_restore_cosmos_artifact_by_resource(resource_name) + if not artifact: + continue + if is_restore_resource_completed(state, resource_name): + resource = state.get("resources", {}).get(resource_name) or {} + artifacts.append(copy.deepcopy(resource.get("result") or {})) + continue + start_restore_resource(state, resource_name, "cosmos_restore") + target_container_name = getattr(app_config, artifact["container_name_attr"], artifact["name"]) + target_container = _get_target_cosmos_container( + target_database, + target_container_name, + artifact["partition_key_path"], + ) + result = { + "name": artifact["name"], + "type": "cosmos_restore", + "container_name": target_container_name, + "partition_key_path": artifact["partition_key_path"], + "status": "completed", + "processed_count": 0, + "copied_count": 0, + "created_count": 0, + "updated_count": 0, + "skipped_count": 0, + "failed_count": 0, + "collision_count": 0, + "bytes": 0, + "checkpoint_count": 0, + } + try: + for record in _iter_restore_records_from_entries( + container_client, + entries, + fernet, + lambda document, artifact=artifact: _restore_cosmos_source_identity(document, artifact), + ): + _assert_restore_job_lease(job) + result["processed_count"] += 1 + result["bytes"] += len(json.dumps(record, default=_json_default).encode("utf-8")) + document_id = _safe_text((record or {}).get("id")) + partition_key = _get_document_path_value(record, artifact["partition_key_path"]) + if not document_id or partition_key is None: + result["failed_count"] += 1 + continue + if overwrite: + target_container.upsert_item(record) + result["updated_count"] += 1 + else: + try: + target_container.create_item(record) + result["created_count"] += 1 + except Exception as exc: + if getattr(exc, "status_code", None) == 409: + result["collision_count"] += 1 + result["skipped_count"] += 1 + continue + raise + result["copied_count"] += 1 + if result["processed_count"] % DATA_MANAGEMENT_RESTORE_BATCH_SIZE == 0: + result["checkpoint_count"] += 1 + _persist_restore_checkpoint( + job, + state, + settings, + resource_name, + result, + {"processed_count": result["processed_count"]}, + f"Restored Cosmos records for {target_container_name}", + ) + result["checkpoint_count"] += 1 + if result["failed_count"] or result["collision_count"]: + result["status"] = "warning" + _complete_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + result, + f"Completed Cosmos restore for {target_container_name}", + ) + except (DataManagementRestoreCanceledError, DataManagementRestoreLeaseLostError): + raise + except Exception as exc: + result["status"] = "warning" + result["failed_count"] += 1 + _append_restore_warning(state, f"Cosmos restore failed for {target_container_name}: {str(exc)[:300]}") + _fail_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + str(exc), + f"Recorded failed Cosmos restore for {target_container_name}", + result=result, + ) + artifacts.append(result) + return artifacts + + +def _execute_restore_search_resources(job, state, settings, restore_plan, container_client, fernet): + if not restore_plan.get("include_ai_search"): + return [{ + "name": "ai_search", + "type": "ai_search_restore", + "status": "skipped", + "warning": "AI Search restore is disabled for this job.", + }] + artifacts = [] + overwrite = _restore_policy_allows_overwrite(restore_plan) + grouped_entries = _group_restore_entries_by_resource(restore_plan["source_backup_id"], "ai_search") + for resource_name, entries in grouped_entries.items(): + artifact = _get_restore_search_artifact_by_resource(resource_name) + if not artifact: + continue + if is_restore_resource_completed(state, resource_name): + resource = state.get("resources", {}).get(resource_name) or {} + artifacts.append(copy.deepcopy(resource.get("result") or {})) + continue + start_restore_resource(state, resource_name, "ai_search_restore") + result = { + "name": artifact["name"], + "type": "ai_search_restore", + "index_name": artifact["index_name"], + "status": "completed", + "processed_count": 0, + "copied_count": 0, + "created_count": 0, + "updated_count": 0, + "skipped_count": 0, + "failed_count": 0, + "collision_count": 0, + "bytes": 0, + "checkpoint_count": 0, + } + try: + _ensure_target_search_index(settings, artifact["index_name"], artifact["schema_file"]) + target_client = _get_target_search_client(settings, artifact["index_name"]) + batch = [] + + def flush_batch(): + if not batch: + return + document_ids = [_safe_text(document.get("id")) for document in batch] + existing = _get_target_search_documents_by_ids(target_client, document_ids) + writable = [] + for document in list(batch): + document_id = _safe_text(document.get("id")) + if not document_id: + result["failed_count"] += 1 + continue + if not overwrite and document_id in existing: + result["collision_count"] += 1 + result["skipped_count"] += 1 + continue + writable.append(document) + if document_id in existing: + result["updated_count"] += 1 + else: + result["created_count"] += 1 + if writable: + upload_results = list(target_client.upload_documents( + documents=writable, + connection_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + read_timeout=DATA_MANAGEMENT_MIGRATION_REMOTE_REQUEST_TIMEOUT_SECONDS, + retry_total=0, + )) + accepted = sum(1 for upload_result in upload_results if _get_search_result_succeeded(upload_result)) + failed = max(0, len(writable) - accepted) + result["copied_count"] += accepted + result["failed_count"] += failed + batch.clear() + result["checkpoint_count"] += 1 + _persist_restore_checkpoint( + job, + state, settings, resource_name, - "Cosmos backup resource failed.", - f"Recorded failed Cosmos backup resource {artifact['name']}", + result, + {"processed_count": result["processed_count"]}, + f"Restored AI Search documents for {artifact['index_name']}", ) - artifacts.append({ - "name": artifact["name"], - "type": "cosmos_container", - "status": "warning", - "warning": warning, - }) - finally: - source_capacity = ( - backup_state.get("source_capacity") - if isinstance(backup_state.get("source_capacity"), dict) else {} - ) - if capacity_attempted or source_capacity.get("restore_pending"): - try: - restore_warnings, _ = _restore_temporary_backup_source_capacity( - job, - backup_state, - settings, - allow_cancel_requested=True, + + for record in _iter_restore_records_from_entries( + container_client, + entries, + fernet, + lambda document, artifact=artifact: _restore_search_source_identity(document, artifact), + ): + _assert_restore_job_lease(job) + result["processed_count"] += 1 + result["bytes"] += len(json.dumps(record, default=_json_default).encode("utf-8")) + batch.append(record) + if len(batch) >= DATA_MANAGEMENT_RESTORE_BATCH_SIZE: + flush_batch() + flush_batch() + if result["failed_count"] or result["collision_count"]: + result["status"] = "warning" + _complete_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + result, + f"Completed AI Search restore for {artifact['index_name']}", + ) + except (DataManagementRestoreCanceledError, DataManagementRestoreLeaseLostError): + raise + except Exception as exc: + result["status"] = "warning" + result["failed_count"] += 1 + _append_restore_warning(state, f"AI Search restore failed for {artifact['index_name']}: {str(exc)[:300]}") + _fail_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + str(exc), + f"Recorded failed AI Search restore for {artifact['index_name']}", + result=result, + ) + artifacts.append(result) + return artifacts + + +def _restore_blob_name_from_artifact_path(artifact_path, base_prefix, container_name, transfer_format): + prefix = f"{_safe_text(base_prefix).strip('/')}/source_blobs/{container_name}/" + artifact = _safe_text(artifact_path) + if artifact.startswith(prefix): + blob_name = artifact[len(prefix):] + else: + marker = f"/source_blobs/{container_name}/" + blob_name = artifact.split(marker, 1)[1] if marker in artifact else "" + if transfer_format == DATA_MANAGEMENT_BLOB_BACKUP_ENCRYPTED_FORMAT and blob_name.endswith(".fernet"): + blob_name = blob_name[:-7] + return blob_name + + +def _execute_restore_source_blob_resources(job, state, settings, restore_plan, container_client, fernet): + if not restore_plan.get("include_source_blobs"): + return [{ + "name": "source_blobs", + "type": "source_blob_restore", + "status": "skipped", + "warning": "Source blob restore is disabled for this job.", + }] + target_service = _get_target_enhanced_citations_blob_client(settings) + artifacts = [] + overwrite = _restore_policy_allows_overwrite(restore_plan) + grouped_entries = _group_restore_entries_by_resource(restore_plan["source_backup_id"], "source_blobs") + for resource_name, entries in grouped_entries.items(): + if not resource_name.startswith("source_blobs:"): + continue + if is_restore_resource_completed(state, resource_name): + resource = state.get("resources", {}).get(resource_name) or {} + artifacts.append(copy.deepcopy(resource.get("result") or {})) + continue + container_name = resource_name.split("source_blobs:", 1)[1] + start_restore_resource(state, resource_name, "source_blob_restore") + target_container = target_service.get_container_client(container_name) + try: + if not target_container.exists(): + target_container.create_container() + except ResourceExistsError: + pass + result = { + "name": container_name, + "type": "source_blob_restore", + "container_name": container_name, + "status": "completed", + "processed_count": 0, + "copied_count": 0, + "created_count": 0, + "updated_count": 0, + "skipped_count": 0, + "failed_count": 0, + "collision_count": 0, + "bytes": 0, + "checkpoint_count": 0, + } + try: + seen_artifacts = set() + for entry in entries: + artifact_path = _safe_text(entry.get("artifact_path")) + if not artifact_path or artifact_path in seen_artifacts: + continue + seen_artifacts.add(artifact_path) + _assert_restore_job_lease(job) + result["processed_count"] += 1 + transfer_format = _safe_text(entry.get("transfer_format"), DATA_MANAGEMENT_BLOB_BACKUP_RAW_FORMAT) + blob_name = _restore_blob_name_from_artifact_path( + artifact_path, + restore_plan.get("base_prefix"), + container_name, + transfer_format, ) - except DataManagementBackupLeaseLostError: - raise - except Exception as exc: - restore_warnings = [ - f"Failed to restore temporary source Cosmos capacity: {str(exc)[:300]}" - ] - for warning in restore_warnings: - _append_backup_warning(backup_state, warning) + if not blob_name: + result["failed_count"] += 1 + continue + target_blob = target_container.get_blob_client(blob_name) + existing = _get_blob_properties_or_none(target_blob) + if existing is not None and not overwrite: + result["collision_count"] += 1 + result["skipped_count"] += 1 + continue + source_bytes = _safe_int(entry.get("source_bytes") or entry.get("bytes"), default=0, minimum=0) + target_blob.upload_blob( + data=_iter_backup_source_blob_artifact_chunks( + container_client, + artifact_path, + transfer_format, + fernet=fernet, + source_version=entry.get("source_version"), + ), + length=source_bytes if source_bytes else None, + overwrite=overwrite, + metadata={ + "simplechatrestorejob": _safe_text(job.get("id")), + "simplechatrestorebackup": _safe_text(restore_plan.get("source_backup_id")), + }, + timeout=DATA_MANAGEMENT_BLOB_SERVICE_TIMEOUT_SECONDS, + ) + if existing is not None: + result["updated_count"] += 1 + else: + result["created_count"] += 1 + result["copied_count"] += 1 + result["bytes"] += source_bytes + result["checkpoint_count"] += 1 + if result["processed_count"] % DATA_MANAGEMENT_RESTORE_BATCH_SIZE == 0: + _persist_restore_checkpoint( + job, + state, + settings, + resource_name, + result, + {"processed_count": result["processed_count"]}, + f"Restored source blobs for {container_name}", + ) + if result["failed_count"] or result["collision_count"]: + result["status"] = "warning" + _complete_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + result, + f"Completed source blob restore for {container_name}", + ) + except (DataManagementRestoreCanceledError, DataManagementRestoreLeaseLostError): + raise + except Exception as exc: + result["status"] = "warning" + result["failed_count"] += 1 + _append_restore_warning(state, f"Source blob restore failed for {container_name}: {str(exc)[:300]}") + _fail_restore_resource_checkpoint( + job, + state, + settings, + resource_name, + str(exc), + f"Recorded failed source blob restore for {container_name}", + result=result, + ) + artifacts.append(result) + return artifacts - source_capacity = ( - backup_state.get("source_capacity") - if isinstance(backup_state.get("source_capacity"), dict) else {} + +def execute_restore_job(job, settings): + """Restore supported backup artifacts into the configured Data Management targets.""" + restore_plan = _normalize_data_management_restore_plan( + job.get("options") if isinstance(job.get("options"), dict) else {}, + backup_job=_get_restore_backup_job((job.get("restore_plan") or {}).get("source_backup_id")), + require_confirmation=True, ) - if source_capacity.get("restore_pending"): - raise RuntimeError( - "Source Cosmos capacity restoration remains pending. Retry the backup to restore the recorded capacity snapshot." - ) - return artifacts + job["restore_plan"] = restore_plan + reviewed_fingerprint = _safe_text((job.get("options") or {}).get("review_fingerprint")) + if reviewed_fingerprint: + current_fingerprint = _restore_review_fingerprint(settings, restore_plan) + if not hmac.compare_digest(reviewed_fingerprint, current_fingerprint): + raise DataManagementSettingsValidationError( + "Restore settings changed after review. Run preflight review again." + ) + review_authorization_token = _safe_text((job.get("options") or {}).get("review_authorization_token")) + review_reservation_token = _safe_text((job.get("options") or {}).get("review_reservation_token")) + if review_authorization_token and review_reservation_token: + consume_data_management_restore_review_authorization( + review_authorization_token, + job.get("requested_by"), + reviewed_fingerprint, + review_reservation_token, + job.get("id"), + ) + + restore_state = initialize_restore_state( + job.get("restore_state"), + job.get("id"), + restore_plan, + ) + start_restore_attempt( + restore_state, + job.get("restore_attempt_id"), + job.get("lease_generation"), + ) + job["restore_state"] = restore_state + restore_state = _persist_restore_state( + job, + restore_state, + settings, + "Initialized durable restore plan and checkpoint state", + ) + + backup_job = _get_restore_backup_job(restore_plan["source_backup_id"]) + container_client, fernet = _get_restore_backup_artifact_context(settings, backup_job) + manifest = _read_backup_json_artifact( + container_client, + restore_plan["source_backup_manifest_path"], + fernet=fernet, + ) + preflight = _run_data_management_restore_preflight( + settings, + restore_plan, + backup_job, + manifest, + ) + restore_state["preflight"] = { + **preflight, + "completed_at": _now_iso(), + } + if not preflight.get("ready"): + raise DataManagementSettingsValidationError("Restore preflight has blockers.") + + total_steps = 5 + _set_job_progress(job, "Starting restore", 0, total_steps, current_step="start") + artifacts = [] + warnings = [] + + artifacts.extend(_execute_restore_cosmos_resources( + job, + restore_state, + settings, + restore_plan, + container_client, + fernet, + )) + _set_job_progress(job, "Cosmos restore step completed", 1, total_steps, current_step="cosmos") + + artifacts.extend(_execute_restore_search_resources( + job, + restore_state, + settings, + restore_plan, + container_client, + fernet, + )) + _set_job_progress(job, "AI Search restore step completed", 2, total_steps, current_step="ai_search") + + artifacts.extend(_execute_restore_source_blob_resources( + job, + restore_state, + settings, + restore_plan, + container_client, + fernet, + )) + _set_job_progress(job, "Source blob restore step completed", 3, total_steps, current_step="source_blobs") + + warnings = list(restore_state.get("warnings") or []) + failed_resource_names = [ + resource_name + for resource_name, resource in (restore_state.get("resources") or {}).items() + if isinstance(resource, dict) and resource.get("status") == RESTORE_RESOURCE_STATUS_FAILED + ] + if failed_resource_names: + warnings.append("One or more restore resources failed and remain eligible for retry.") + restore_state.update({ + "phase": "completed", + "status": RESTORE_RESOURCE_STATUS_COMPLETED, + "completed_at": _now_iso(), + "failed_resource_names": failed_resource_names, + }) + complete_restore_attempt( + restore_state, + "completed_with_warnings" if warnings or failed_resource_names else "completed", + ) + _persist_restore_state( + job, + restore_state, + settings, + "Restore completed", + allow_cancel_requested=True, + ) + artifact_summaries = summarize_backup_artifacts(artifacts) + artifact_totals = _backup_artifact_totals(artifact_summaries) + _record_data_management_job_event( + job.get("id"), + "restore", + job, + status=( + DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS + if warnings or failed_resource_names else DATA_MANAGEMENT_STATUS_COMPLETED + ), + message="Restore target resources written", + details={ + "source_backup_id": restore_plan.get("source_backup_id"), + "restore_policy": restore_plan.get("restore_policy"), + "artifact_totals": artifact_totals, + "warnings": warnings, + "failed_resource_count": len(failed_resource_names), + }, + ) + _set_job_progress( + job, + "Restore completed", + total_steps, + total_steps, + current_step="completed", + status=( + DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS + if warnings or failed_resource_names else DATA_MANAGEMENT_STATUS_COMPLETED + ), + ) + return { + "source_backup_id": restore_plan.get("source_backup_id"), + "restore_policy": restore_plan.get("restore_policy"), + "artifact_count": len(artifacts), + "artifact_totals": artifact_totals, + "artifacts": artifact_summaries, + "warnings": warnings, + "restore_state": restore_state, + "failed_resource_names": failed_resource_names, + } def execute_backup_job(job, settings): @@ -16154,22 +17956,32 @@ def process_data_management_job(job_id): warnings = list(job.get("warnings") or []) + result.get("warnings", []) status = DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS if warnings else DATA_MANAGEMENT_STATUS_COMPLETED message = "Migration completed with warnings" if warnings else "Migration completed successfully" + elif job.get("operation") == DATA_MANAGEMENT_OPERATION_RESTORE: + result = execute_restore_job(job, settings) + warnings = list(job.get("warnings") or []) + result.get("warnings", []) + failed_resource_names = list(result.get("failed_resource_names") or []) + if failed_resource_names: + warnings.append( + "One or more restore resources completed with retryable failures." + ) + status = DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS if warnings else DATA_MANAGEMENT_STATUS_COMPLETED + message = "Restore completed with warnings" if warnings else "Restore completed successfully" else: warnings = list(job.get("warnings") or []) warnings.append( - "Restore and migration apply logic has not run in this job. The durable job record, settings, and admin workflow are ready for the restore and migration implementation layer." + "This Data Management operation does not have an execution handler." ) result = {"warnings": warnings} _record_data_management_job_event( job.get("id"), - "orchestration-foundation", + "unsupported-operation", job, status=DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS, - message="Restore and migration apply logic has not run in this job.", + message="Data Management operation has no execution handler.", details={"message": warnings[-1]}, ) status = DATA_MANAGEMENT_STATUS_COMPLETED_WITH_WARNINGS - message = "Data Management job foundation completed with implementation warnings" + message = "Data Management job completed with implementation warnings" now = _now_iso() completed_progress = ( @@ -16300,6 +18112,52 @@ def process_data_management_job(job_id): }, ) return saved_job + except DataManagementRestoreCanceledError: + now = _now_iso() + try: + persisted_job = _read_job(job_id) + except Exception: + persisted_job = job + restore_state = persisted_job.get("restore_state") + if isinstance(restore_state, dict): + restore_state.update({ + "status": DATA_MANAGEMENT_STATUS_CANCELED, + "phase": "canceled", + "completed_at": now, + "updated_at": now, + "last_error": None, + }) + complete_restore_attempt(restore_state, "canceled") + persisted_job.update({ + "status": DATA_MANAGEMENT_STATUS_CANCELED, + "updated_at": now, + "completed_at": now, + "last_heartbeat_at": now, + "last_message": "Restore canceled at a durable checkpoint.", + "last_error": None, + "lease_holder_id": None, + "lease_expires_at": None, + "restore_state": restore_state, + "progress": { + "total_steps": persisted_job.get("progress", {}).get("total_steps", 0), + "completed_steps": persisted_job.get("progress", {}).get("completed_steps", 0), + "current_step": "canceled", + "percent_complete": persisted_job.get("progress", {}).get("percent_complete", 0), + }, + }) + saved_job = _save_data_management_job(persisted_job) + _record_data_management_job_event( + saved_job.get("id"), + "restore-canceled", + saved_job, + status=DATA_MANAGEMENT_STATUS_CANCELED, + message="Restore canceled at a durable checkpoint.", + details={ + "cancel_requested_at": saved_job.get("cancel_requested_at"), + "reason_present": bool(saved_job.get("cancel_reason")), + }, + ) + return saved_job except DataManagementBackupLeaseLostError as exc: log_event( "[DataManagement] Backup worker stopped after losing its job or source lease.", @@ -16314,6 +18172,13 @@ def process_data_management_job(job_id): level=logging.WARNING, ) return None + except DataManagementRestoreLeaseLostError as exc: + log_event( + "[DataManagement] Restore worker stopped after losing its job lease.", + {"job_id": job_id, "operation": job.get("operation"), "error": str(exc)}, + level=logging.WARNING, + ) + return None except Exception as exc: now = _now_iso() job.update({ diff --git a/application/single_app/functions_data_management_restore_state.py b/application/single_app/functions_data_management_restore_state.py new file mode 100644 index 000000000..e462efa60 --- /dev/null +++ b/application/single_app/functions_data_management_restore_state.py @@ -0,0 +1,213 @@ +# functions_data_management_restore_state.py +"""Durable, secret-free checkpoint state for Data Management restores.""" + +import copy +import hashlib +import json +from datetime import datetime, timezone + + +RESTORE_STATE_SCHEMA_VERSION = 1 +RESTORE_RESOURCE_STATUS_PENDING = "pending" +RESTORE_RESOURCE_STATUS_IN_PROGRESS = "in_progress" +RESTORE_RESOURCE_STATUS_COMPLETED = "completed" +RESTORE_RESOURCE_STATUS_FAILED = "failed" +RESTORE_STATE_MAX_ATTEMPT_HISTORY = 20 + + +def _utc_now_iso(current_time=None): + """Return a normalized UTC timestamp without accepting local wall-clock time.""" + timestamp = current_time if isinstance(current_time, datetime) else datetime.now(timezone.utc) + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + return timestamp.astimezone(timezone.utc).isoformat() + + +def build_restore_configuration_fingerprint(configuration): + """Hash a caller-supplied, secret-free immutable restore plan.""" + encoded = json.dumps( + configuration if isinstance(configuration, dict) else {}, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + default=str, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def initialize_restore_state( + existing_state, + restore_id, + normalized_plan, + current_time=None, +): + """Create or resume restore state while protecting immutable plan values.""" + plan = copy.deepcopy(normalized_plan if isinstance(normalized_plan, dict) else {}) + fingerprint = build_restore_configuration_fingerprint(plan) + timestamp = _utc_now_iso(current_time) + state = copy.deepcopy(existing_state) if isinstance(existing_state, dict) else {} + + if state: + if state.get("schema_version") != RESTORE_STATE_SCHEMA_VERSION: + raise ValueError("Restore state uses an unsupported schema version.") + if state.get("restore_id") != restore_id: + raise ValueError("Restore state belongs to a different restore job.") + if state.get("configuration_fingerprint") != fingerprint: + raise ValueError("Restore plan changed after the job was checkpointed. Queue a new restore job.") + state["status"] = RESTORE_RESOURCE_STATUS_IN_PROGRESS + state["updated_at"] = timestamp + state["resume_count"] = int(state.get("resume_count") or 0) + 1 + state.setdefault("attempt_history", []) + state.setdefault("resources", {}) + state.setdefault("preflight", {}) + state.setdefault("totals", {}) + state.setdefault("warnings", []) + return state + + return { + "schema_version": RESTORE_STATE_SCHEMA_VERSION, + "restore_id": restore_id, + "normalized_plan": plan, + "configuration_fingerprint": fingerprint, + "status": RESTORE_RESOURCE_STATUS_PENDING, + "phase": "queued", + "created_at": timestamp, + "updated_at": timestamp, + "completed_at": None, + "resume_count": 0, + "attempt_history": [], + "resources": {}, + "preflight": {}, + "totals": {}, + "warnings": [], + } + + +def start_restore_attempt(state, attempt_id, lease_generation, current_time=None): + """Record one fenced execution attempt while retaining a bounded history.""" + if not isinstance(state, dict): + raise ValueError("Restore state must be a dictionary.") + timestamp = _utc_now_iso(current_time) + history = state.setdefault("attempt_history", []) + if history: + state["resume_count"] = int(state.get("resume_count") or 0) + 1 + history.append({ + "attempt_id": str(attempt_id or ""), + "lease_generation": int(lease_generation or 0), + "started_at": timestamp, + "completed_at": None, + "outcome": "running", + }) + del history[:-RESTORE_STATE_MAX_ATTEMPT_HISTORY] + state.update({ + "status": RESTORE_RESOURCE_STATUS_IN_PROGRESS, + "phase": "initializing", + "updated_at": timestamp, + "completed_at": None, + }) + return state + + +def complete_restore_attempt(state, outcome, current_time=None): + """Close the current bounded attempt record at a durable terminal boundary.""" + if not isinstance(state, dict): + raise ValueError("Restore state must be a dictionary.") + timestamp = _utc_now_iso(current_time) + history = state.setdefault("attempt_history", []) + if history and isinstance(history[-1], dict): + history[-1]["completed_at"] = timestamp + history[-1]["outcome"] = str(outcome or "completed") + state["updated_at"] = timestamp + return state + + +def get_restore_resource(state, resource_name): + """Return a resource checkpoint without allocating a missing record.""" + resources = state.get("resources") if isinstance(state, dict) else None + return resources.get(resource_name) if isinstance(resources, dict) else None + + +def is_restore_resource_completed(state, resource_name): + """Return whether one restore resource has a verified durable result.""" + resource = get_restore_resource(state, resource_name) + return isinstance(resource, dict) and resource.get("status") == RESTORE_RESOURCE_STATUS_COMPLETED + + +def start_restore_resource(state, resource_name, phase, current_time=None): + """Start or resume one restore resource while preserving durable progress.""" + if not isinstance(state, dict): + raise ValueError("Restore state must be a dictionary.") + timestamp = _utc_now_iso(current_time) + resources = state.setdefault("resources", {}) + existing = resources.get(resource_name) if isinstance(resources, dict) else None + if isinstance(existing, dict) and existing.get("status") == RESTORE_RESOURCE_STATUS_COMPLETED: + return existing + + progress = copy.deepcopy(existing.get("progress") if isinstance(existing, dict) else {}) + checkpoint = copy.deepcopy(existing.get("checkpoint") if isinstance(existing, dict) else {}) + attempts = int(existing.get("attempts") or 0) + 1 if isinstance(existing, dict) else 1 + resource = { + "status": RESTORE_RESOURCE_STATUS_IN_PROGRESS, + "phase": str(phase or "restore"), + "attempts": attempts, + "started_at": existing.get("started_at") if isinstance(existing, dict) and existing.get("started_at") else timestamp, + "attempt_started_at": timestamp, + "updated_at": timestamp, + "completed_at": None, + "last_error": None, + "progress": progress, + "checkpoint": checkpoint, + "result": {}, + } + resources[resource_name] = resource + state["phase"] = resource["phase"] + state["updated_at"] = timestamp + return resource + + +def update_restore_resource(state, resource_name, progress, checkpoint=None, current_time=None): + """Persist bounded per-resource restore progress.""" + resource = get_restore_resource(state, resource_name) + if not isinstance(resource, dict): + raise ValueError(f"Restore resource '{resource_name}' was not started.") + if resource.get("status") == RESTORE_RESOURCE_STATUS_COMPLETED: + raise ValueError(f"Restore resource '{resource_name}' is already completed.") + timestamp = _utc_now_iso(current_time) + resource["progress"] = copy.deepcopy(progress if isinstance(progress, dict) else {}) + if checkpoint is not None: + resource["checkpoint"] = copy.deepcopy(checkpoint if isinstance(checkpoint, dict) else {}) + resource["updated_at"] = timestamp + state["updated_at"] = timestamp + return resource + + +def complete_restore_resource(state, resource_name, result=None, current_time=None): + """Mark a restore resource complete only after final target writes are known.""" + resource = get_restore_resource(state, resource_name) + if not isinstance(resource, dict): + raise ValueError(f"Restore resource '{resource_name}' was not started.") + timestamp = _utc_now_iso(current_time) + resource.update({ + "status": RESTORE_RESOURCE_STATUS_COMPLETED, + "updated_at": timestamp, + "completed_at": timestamp, + "last_error": None, + "result": copy.deepcopy(result if isinstance(result, dict) else {}), + }) + state["updated_at"] = timestamp + return resource + + +def fail_restore_resource(state, resource_name, error_message, current_time=None): + """Persist a failed restore resource so the same job can resume later.""" + resource = get_restore_resource(state, resource_name) + if not isinstance(resource, dict): + resource = start_restore_resource(state, resource_name, "restore", current_time=current_time) + timestamp = _utc_now_iso(current_time) + resource.update({ + "status": RESTORE_RESOURCE_STATUS_FAILED, + "updated_at": timestamp, + "last_error": str(error_message or "Restore resource failed."), + }) + state["updated_at"] = timestamp + return resource diff --git a/application/single_app/route_backend_data_management.py b/application/single_app/route_backend_data_management.py index 6e8aacb86..7adb27318 100644 --- a/application/single_app/route_backend_data_management.py +++ b/application/single_app/route_backend_data_management.py @@ -18,6 +18,7 @@ DataManagementHistoryPaginationError, DataManagementSettingsValidationError, create_data_management_migration_review_authorization, + create_data_management_restore_review_authorization, export_data_management_migration_manifest, generate_data_management_encryption_key, get_data_management_cosmos_editor_containers, @@ -28,6 +29,7 @@ get_data_management_jobs_page, get_data_management_migration_catalog, get_data_management_migration_review_fingerprint, + get_data_management_restore_review_fingerprint, get_data_management_settings, log_data_management_cosmos_editor_activity, preview_data_management_migration_plan, @@ -35,11 +37,15 @@ query_data_management_cosmos_editor_documents, request_data_management_job_cancellation, release_data_management_migration_review_reservation, + release_data_management_restore_review_reservation, reserve_data_management_migration_review_authorization, + reserve_data_management_restore_review_authorization, review_data_management_migration, + review_data_management_restore, retry_data_management_backup_job, resolve_data_management_migration_manifest_item, retry_data_management_migration_job, + retry_data_management_restore_job, sanitize_data_management_job_for_admin, sanitize_data_management_settings_for_admin, save_data_management_cosmos_editor_document, @@ -522,6 +528,8 @@ def retry_admin_data_management_migration_job(job_id): ) if operation == DATA_MANAGEMENT_OPERATION_BACKUP: job = retry_data_management_backup_job(job_id) + elif operation == DATA_MANAGEMENT_OPERATION_RESTORE: + job = retry_data_management_restore_job(job_id) else: job = retry_data_management_migration_job(job_id) submitted = submit_data_management_job(current_app._get_current_object(), job.get("id")) @@ -697,6 +705,59 @@ def review_admin_data_management_migration(): }), 400 return jsonify({"success": True, "review": review}), 200 + @bp.route("/api/admin/data-management/restore/review", methods=["POST"]) + @swagger_route(security=get_auth_security()) + @login_required + @admin_required + def review_admin_data_management_restore(): + payload = request.get_json(silent=True) or {} + try: + review = review_data_management_restore( + settings=( + payload.get("settings") + if isinstance(payload.get("settings"), dict) + else None + ), + restore_plan=( + payload.get("restore_plan") + if isinstance(payload.get("restore_plan"), dict) + else None + ), + ) + if review.get("ready") is True: + admin_user_id, _admin_email = _get_admin_context() + review.update( + create_data_management_restore_review_authorization( + admin_user_id, + review.get("review_fingerprint"), + ) + ) + except DataManagementSettingsValidationError as exc: + log_event( + "[DataManagement] Restore review validation failed.", + {"error_type": type(exc).__name__}, + level=logging.WARNING, + exceptionTraceback=True, + ) + return jsonify({ + "success": False, + "error": "Restore review input is invalid.", + "workflow_step": "review", + }), 400 + except Exception as exc: + log_event( + "[DataManagement] Restore review failed.", + {"error_type": type(exc).__name__}, + level=logging.ERROR, + exceptionTraceback=True, + ) + return jsonify({ + "success": False, + "error": "Restore review could not be completed.", + "workflow_step": "review", + }), 400 + return jsonify({"success": True, "review": review}), 200 + @bp.route("/api/admin/data-management/jobs", methods=["POST"]) @swagger_route(security=get_auth_security()) @login_required @@ -706,6 +767,8 @@ def create_admin_data_management_job(): operation = str(payload.get("operation") or DATA_MANAGEMENT_OPERATION_DRY_RUN).strip() backup_type = payload.get("backup_type") review_reservation = None + review_authorization_token = "" + review_reservation_release = None if operation not in { DATA_MANAGEMENT_OPERATION_BACKUP, DATA_MANAGEMENT_OPERATION_RESTORE, @@ -766,6 +829,7 @@ def create_admin_data_management_job(): job_options["review_reservation_token"] = ( review_reservation["reservation_token"] ) + review_reservation_release = release_data_management_migration_review_reservation except DataManagementSettingsValidationError as exc: log_event( "[DataManagement] Migration review reservation validation failed.", @@ -777,6 +841,61 @@ def create_admin_data_management_job(): "error": "Migration review authorization is invalid or expired. Run review again before execution.", "workflow_step": "review", }), 409 + elif operation == DATA_MANAGEMENT_OPERATION_RESTORE: + provided_review_fingerprint = str( + job_options.get("review_fingerprint") or "" + ).strip() + review_authorization_token = str( + job_options.get("review_authorization_token") or "" + ).strip() + restore_plan = ( + job_options.get("restore_plan") + if isinstance(job_options.get("restore_plan"), dict) + else None + ) + expected_review_fingerprint = ( + get_data_management_restore_review_fingerprint( + restore_plan=restore_plan, + ) + ) + if ( + not provided_review_fingerprint or + not hmac.compare_digest( + provided_review_fingerprint, + expected_review_fingerprint, + ) + ): + return jsonify({ + "success": False, + "error": ( + "Restore inputs changed after preflight review. " + "Run review again before execution." + ), + "workflow_step": "review", + }), 409 + try: + review_reservation = ( + reserve_data_management_restore_review_authorization( + review_authorization_token, + admin_user_id, + expected_review_fingerprint, + ) + ) + job_options["review_reservation_token"] = ( + review_reservation["reservation_token"] + ) + review_reservation_release = release_data_management_restore_review_reservation + except DataManagementSettingsValidationError as exc: + log_event( + "[DataManagement] Restore review reservation validation failed.", + {"operation": operation, "error": str(exc)}, + level=logging.WARNING, + ) + return jsonify({ + "success": False, + "error": "Restore review authorization is invalid or expired. Run review again before execution.", + "workflow_step": "review", + }), 409 try: job = queue_data_management_job( operation, @@ -791,8 +910,8 @@ def create_admin_data_management_job(): ), ) except Exception: - if review_reservation: - release_data_management_migration_review_reservation( + if review_reservation and review_reservation_release: + review_reservation_release( review_authorization_token, review_reservation["reservation_token"], ) @@ -808,7 +927,7 @@ def create_admin_data_management_job(): "success": False, "error": "Data Management job request is invalid.", } - if operation == DATA_MANAGEMENT_OPERATION_MIGRATION: + if operation in {DATA_MANAGEMENT_OPERATION_MIGRATION, DATA_MANAGEMENT_OPERATION_RESTORE}: response["workflow_step"] = "confirm" return jsonify(response), 400 except Exception as exc: diff --git a/application/single_app/static/js/admin/admin_data_management.js b/application/single_app/static/js/admin/admin_data_management.js index f28b18178..04d81bf27 100644 --- a/application/single_app/static/js/admin/admin_data_management.js +++ b/application/single_app/static/js/admin/admin_data_management.js @@ -8,6 +8,7 @@ const backupStorageAuthConnectionString = "connection_string"; const targetCosmosDatabaseName = "SimpleChat"; const cosmosEditorConfirmationPhrase = "I understand this can damage system data"; const migrationMirrorConfirmationPhrase = "MIRROR WITH DELETIONS"; +const restoreOverwriteConfirmationPhrase = "RESTORE WITH OVERWRITE"; const elements = {}; let dataManagementModified = false; let storedBackupConnectionStringAvailable = false; @@ -23,6 +24,8 @@ let cosmosEditorResultCount = 0; let cosmosEditorSelectedDocument = null; let cosmosEditorPendingDocument = null; let pendingDataManagementCancellationJob = null; +let pendingRestoreBackup = null; +let restoreReview = null; let migrationWorkflowRefreshTimer = null; let migrationWorkflowRefreshInFlight = false; @@ -220,6 +223,20 @@ function bindElements() { "data-management-backup-pagination-status", "data-management-backup-previous-page-btn", "data-management-backup-next-page-btn", + "data-management-restore-modal", + "data-management-restore-title", + "data-management-restore-backup-summary", + "data_management_restore_policy", + "data_management_restore_include_cosmos", + "data_management_restore_include_ai_search", + "data_management_restore_include_source_blobs", + "data-management-restore-confirmation-section", + "data_management_restore_overwrite_confirmation_phrase", + "data-management-restore-review-btn", + "data-management-restore-review-status", + "data-management-restore-review-checks", + "data_management_restore_final_confirmation", + "data-management-restore-queue-btn", "data-management-refresh-jobs-btn", "data-management-jobs-tbody", "data_management_job_operation_filter", @@ -329,6 +346,19 @@ function bindEvents() { ].forEach((filter) => filter?.addEventListener("change", () => resetAndLoadHistory("backups"))); elements.dataManagementBackupPreviousPageBtn?.addEventListener("click", () => loadPreviousHistoryPage("backups")); elements.dataManagementBackupNextPageBtn?.addEventListener("click", () => loadNextHistoryPage("backups")); + elements.datamanagementrestorepolicy?.addEventListener("change", () => { + invalidateRestoreReview(); + updateRestoreConfirmationVisibility(); + }); + [ + elements.datamanagementrestoreincludecosmos, + elements.datamanagementrestoreincludeaisearch, + elements.datamanagementrestoreincludesourceblobs, + ].forEach((control) => control?.addEventListener("change", invalidateRestoreReview)); + elements.datamanagementrestoreoverwriteconfirmationphrase?.addEventListener("input", updateRestoreQueueButtonState); + elements.datamanagementrestorefinalconfirmation?.addEventListener("change", updateRestoreQueueButtonState); + elements.dataManagementRestoreReviewBtn?.addEventListener("click", runRestoreReview); + elements.dataManagementRestoreQueueBtn?.addEventListener("click", queueRestore); elements.dataManagementRefreshJobsBtn?.addEventListener("click", loadDataManagementJobs); [ elements.datamanagementjoboperationfilter, @@ -1662,6 +1692,9 @@ function buttonForOperation(operation, backupType) { if (operation === "migration") { return elements.dataManagementExecuteMigrationBtn; } + if (operation === "restore") { + return elements.dataManagementRestoreQueueBtn; + } return null; } @@ -2399,6 +2432,9 @@ function formatOperation(operation, backupType) { if (operation === "backup") { return `${backupType || "manual"} backup`; } + if (operation === "restore") { + return "restore"; + } return operation.replace(/_/g, " "); } @@ -2916,7 +2952,7 @@ function createBackupRow(backup) { row.appendChild(createBackupStorageCell(backup)); row.appendChild(createBackupProtectionCell(backup)); row.appendChild(createBackupWarningCell(backup)); - row.appendChild(createJobActionCell(backup.id)); + row.appendChild(createBackupActionCell(backup)); return row; } @@ -2977,6 +3013,36 @@ function createBackupWarningCell(backup) { cell.appendChild(createBadge(`${formatNumber(warningCount)} warning${warningCount === 1 ? "" : "s"}`, "bg-warning text-dark")); return cell; } + + function createBackupActionCell(backup) { + const cell = document.createElement("td"); + const viewButton = document.createElement("button"); + viewButton.type = "button"; + viewButton.className = "btn btn-outline-primary btn-sm"; + viewButton.disabled = !backup?.id; + const viewIcon = document.createElement("i"); + viewIcon.className = "bi bi-list-check me-1"; + viewButton.append(viewIcon, document.createTextNode("View Log")); + viewButton.addEventListener("click", () => loadDataManagementJobDetail(backup.id)); + cell.appendChild(viewButton); + + const canRestore = Boolean( + backup?.id && + backup?.manifest_path && + ["completed", "completed_with_warnings"].includes(String(backup.status || "")) + ); + if (canRestore) { + const restoreButton = document.createElement("button"); + restoreButton.type = "button"; + restoreButton.className = "btn btn-outline-danger btn-sm ms-1"; + const restoreIcon = document.createElement("i"); + restoreIcon.className = "bi bi-arrow-counterclockwise me-1"; + restoreButton.append(restoreIcon, document.createTextNode("Restore")); + restoreButton.addEventListener("click", () => openRestoreModal(backup)); + cell.appendChild(restoreButton); + } + return cell; + } const noWarnings = document.createElement("span"); noWarnings.className = "text-muted"; noWarnings.textContent = "None"; @@ -2984,6 +3050,189 @@ function createBackupWarningCell(backup) { return cell; } +function openRestoreModal(backup) { + pendingRestoreBackup = backup || null; + restoreReview = null; + setValue(elements.datamanagementrestorepolicy, "create_only"); + setChecked(elements.datamanagementrestoreincludecosmos, true); + setChecked(elements.datamanagementrestoreincludeaisearch, true); + setChecked(elements.datamanagementrestoreincludesourceblobs, true); + setChecked(elements.datamanagementrestorefinalconfirmation, false); + setValue(elements.datamanagementrestoreoverwriteconfirmationphrase, ""); + renderRestoreBackupSummary(backup); + renderRestoreReviewEmpty(); + updateRestoreConfirmationVisibility(); + updateRestoreQueueButtonState(); + if (elements.dataManagementRestoreModal && window.bootstrap?.Modal) { + window.bootstrap.Modal.getOrCreateInstance(elements.dataManagementRestoreModal).show(); + } +} + +function renderRestoreBackupSummary(backup) { + const container = elements.dataManagementRestoreBackupSummary; + if (!container) { + return; + } + container.replaceChildren( + createRestoreSummaryTile("Backup ID", backup?.id || "N/A"), + createRestoreSummaryTile("Type", formatBackupType(backup?.backup_type || "")), + createRestoreSummaryTile("Completed", formatDate(backup?.completed_at || backup?.created_at)), + createRestoreSummaryTile("Contents", `${formatNumber(backup?.record_count || 0)} records, ${formatNumber(backup?.blob_count || 0)} blobs`), + createRestoreSummaryTile("Manifest", backup?.manifest_path ? "Recorded" : "Missing"), + createRestoreSummaryTile("Protection", backup?.encrypted ? "Encrypted" : "Not encrypted") + ); +} + +function createRestoreSummaryTile(label, value) { + const tile = document.createElement("div"); + tile.className = "col-sm-6 col-xl-4"; + const labelElement = document.createElement("div"); + labelElement.className = "small text-muted"; + labelElement.textContent = label; + const valueElement = document.createElement("div"); + valueElement.className = "fw-semibold text-break"; + valueElement.textContent = value || "N/A"; + tile.append(labelElement, valueElement); + return tile; +} + +function buildRestorePlan() { + return { + source_backup_id: pendingRestoreBackup?.id || "", + restore_policy: getValue(elements.datamanagementrestorepolicy) || "create_only", + include_cosmos: Boolean(elements.datamanagementrestoreincludecosmos?.checked), + include_ai_search: Boolean(elements.datamanagementrestoreincludeaisearch?.checked), + include_source_blobs: Boolean(elements.datamanagementrestoreincludesourceblobs?.checked), + overwrite_confirmed: getValue(elements.datamanagementrestorepolicy) === "overwrite_existing", + overwrite_confirmation_phrase: getValue(elements.datamanagementrestoreoverwriteconfirmationphrase), + }; +} + +function invalidateRestoreReview() { + restoreReview = null; + renderRestoreReviewEmpty("Restore settings changed. Run review again before queueing."); + updateRestoreQueueButtonState(); +} + +function updateRestoreConfirmationVisibility() { + const requiresOverwrite = getValue(elements.datamanagementrestorepolicy) === "overwrite_existing"; + setElementVisible(elements.dataManagementRestoreConfirmationSection, requiresOverwrite); + updateRestoreQueueButtonState(); +} + +function renderRestoreReviewEmpty(message = "Run review to validate the manifest, target access, and restore policy.") { + setText(elements.dataManagementRestoreReviewStatus, message); + const container = elements.dataManagementRestoreReviewChecks; + if (!container) { + return; + } + const item = document.createElement("div"); + item.className = "list-group-item text-muted"; + item.textContent = "No current restore review is available."; + container.replaceChildren(item); +} + +function renderRestoreReview(review) { + const checks = Array.isArray(review?.checks) ? review.checks : []; + const container = elements.dataManagementRestoreReviewChecks; + setText( + elements.dataManagementRestoreReviewStatus, + review?.ready + ? "Restore preflight passed. Confirm the plan to queue the job." + : "Restore preflight found blockers. Resolve them and run review again." + ); + if (!container) { + return; + } + if (!checks.length) { + renderRestoreReviewEmpty("Restore review returned no checks."); + return; + } + container.replaceChildren(); + checks.forEach((check) => { + const item = document.createElement("div"); + item.className = "list-group-item"; + const header = document.createElement("div"); + header.className = "d-flex flex-wrap justify-content-between gap-2"; + const title = document.createElement("strong"); + title.textContent = check.label || check.id || "Restore check"; + const badge = createBadge( + formatActivityLabel(check.status || "unknown"), + check.status === "pass" ? "bg-success" : check.status === "warn" ? "bg-warning text-dark" : "bg-danger" + ); + header.append(title, badge); + const message = document.createElement("div"); + message.className = "small text-muted mt-1"; + message.textContent = check.message || ""; + item.append(header, message); + container.appendChild(item); + }); +} + +async function runRestoreReview() { + if (!pendingRestoreBackup?.id) { + showToast("Choose a backup to restore first.", "warning"); + return; + } + setBusy(elements.dataManagementRestoreReviewBtn, true, "Reviewing..."); + try { + await saveDataManagementSettings(true); + const data = await requestJson("/api/admin/data-management/restore/review", { + method: "POST", + body: JSON.stringify({ + settings: collectSettings(), + restore_plan: buildRestorePlan(), + }), + }); + restoreReview = data.review || null; + renderRestoreReview(restoreReview); + updateRestoreQueueButtonState(); + showToast( + restoreReview?.ready ? "Restore preflight passed." : "Restore preflight found blockers.", + restoreReview?.ready ? "success" : "warning" + ); + } catch (error) { + restoreReview = null; + renderRestoreReviewEmpty(error.message || "Restore review failed."); + showToast(error.message || "Restore review failed.", "danger"); + updateRestoreQueueButtonState(); + } finally { + setBusy(elements.dataManagementRestoreReviewBtn, false); + } +} + +function updateRestoreQueueButtonState() { + const overwriteRequired = getValue(elements.datamanagementrestorepolicy) === "overwrite_existing"; + const overwriteMatches = !overwriteRequired || getValue(elements.datamanagementrestoreoverwriteconfirmationphrase) === restoreOverwriteConfirmationPhrase; + const canQueue = Boolean( + pendingRestoreBackup?.id && + restoreReview?.ready === true && + elements.datamanagementrestorefinalconfirmation?.checked && + overwriteMatches + ); + setButtonDisabled(elements.dataManagementRestoreQueueBtn, !canQueue); +} + +async function queueRestore() { + if (!restoreReview?.ready || !pendingRestoreBackup?.id) { + showToast("Run a passing restore review before queueing.", "warning"); + return; + } + const restorePlan = buildRestorePlan(); + const data = await queueOperation("restore", null, { + restore_plan: restorePlan, + review_fingerprint: restoreReview.review_fingerprint || "", + review_authorization_token: restoreReview.authorization_token || "", + }, elements.dataManagementRestoreQueueBtn); + if (!data?.job) { + return; + } + if (elements.dataManagementRestoreModal && window.bootstrap?.Modal) { + window.bootstrap.Modal.getOrCreateInstance(elements.dataManagementRestoreModal).hide(); + } + loadDataManagementJobDetail(data.job.id); +} + function createLabeledValue(label, value, allowBreak = false) { const wrapper = document.createElement("div"); const labelElement = document.createElement("span"); @@ -3388,6 +3637,16 @@ function renderJobDetailProgress(job) { }); body.appendChild(metricGrid); } + } else if (job.operation === "restore") { + const restoreMetrics = getRestoreLiveMetrics(job); + if (restoreMetrics.length) { + const metricGrid = document.createElement("div"); + metricGrid.className = "row g-2 small mt-3"; + restoreMetrics.forEach((metric) => { + metricGrid.appendChild(createDetailMetric(metric.label, metric.value)); + }); + body.appendChild(metricGrid); + } } card.appendChild(body); container.replaceChildren(card); @@ -3481,6 +3740,33 @@ function getBackupLiveMetrics(job) { return metrics; } +function getRestoreLiveMetrics(job) { + const restoreState = job?.restore_state; + if (!restoreState || typeof restoreState !== "object") { + return []; + } + const totals = restoreState.totals && typeof restoreState.totals === "object" ? restoreState.totals : {}; + const resources = restoreState.resources && typeof restoreState.resources === "object" ? Object.values(restoreState.resources) : []; + const activeResource = resources.find((resource) => resource?.status === "in_progress") || {}; + const activeMetrics = activeResource.progress && typeof activeResource.progress === "object" ? activeResource.progress : {}; + const metrics = [ + { label: "Processed", value: formatNumber(totals.processed_count || 0) }, + { label: "Copied", value: formatNumber(totals.copied_count || 0) }, + { label: "Created / updated", value: `${formatNumber(totals.created_count || 0)} / ${formatNumber(totals.updated_count || 0)}` }, + { label: "Skipped / failed", value: `${formatNumber(totals.skipped_count || 0)} / ${formatNumber(totals.failed_count || 0)}` }, + { label: "Collisions", value: formatNumber(totals.collision_count || 0) }, + { label: "Transferred", value: formatBytes(totals.bytes || 0) }, + ]; + if (activeMetrics.container_name || activeMetrics.index_name) { + metrics.push({ label: "Current target", value: activeMetrics.container_name || activeMetrics.index_name }); + } + const progressAge = formatTimestampAge(job?.last_progress_at || restoreState.last_progress_at); + if (progressAge) { + metrics.push({ label: "Last progress", value: progressAge }); + } + return metrics; +} + function timestampAgeSeconds(value) { const timestamp = Date.parse(value || ""); if (!Number.isFinite(timestamp)) { diff --git a/application/single_app/templates/admin_settings.html b/application/single_app/templates/admin_settings.html index 9bd8fb62a..85f2e4342 100644 --- a/application/single_app/templates/admin_settings.html +++ b/application/single_app/templates/admin_settings.html @@ -6551,6 +6551,96 @@

Backup Inventory

+ +

Job History

diff --git a/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md b/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md index b342e1ee9..423ea2ebe 100644 --- a/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md +++ b/docs/explanation/features/DATA_MANAGEMENT_BACKUP_MIGRATION.md @@ -1,7 +1,7 @@ # Data Management Backup and Migration Implemented in version: **0.241.211** -Updated in version: **0.250.105** +Updated in version: **0.250.106** ## Overview @@ -112,6 +112,18 @@ Migration execution currently copies selected SimpleChat Cosmos records, matchin For durable provenance, destination access probes, collision protection, checkpoint retries, transfer telemetry, and temporary destination Cosmos capacity controls, see [Data Management Migration Resilience](DATA_MANAGEMENT_MIGRATION_RESILIENCE.md). +### Restore Workflow + +The Backup Inventory exposes a **Restore** action for completed backup jobs with a durable manifest. Restore uses a reviewed workflow before queueing any target writes: + +1. Select a completed backup from Backup Inventory. +2. Choose create-only or explicitly confirmed overwrite restore policy. +3. Select included restore surfaces: Cosmos DB, AI Search, and Enhanced Citation blobs. +4. Run preflight review to validate the manifest, target access, partition-key compatibility, and destination collision policy. +5. Confirm the reviewed plan and queue a durable restore job. + +Create-only restore is the default and blocks existing destination collisions. Overwrite restore requires the exact `RESTORE WITH OVERWRITE` phrase. Running restores use durable checkpoints, sanitized progress, cancellation, retry, and Activity Log auditing. For implementation and recovery guidance, see [Data Management Restore](DATA_MANAGEMENT_RESTORE.md). + ### Security - All Data Management routes are admin-only. diff --git a/docs/explanation/features/DATA_MANAGEMENT_RESTORE.md b/docs/explanation/features/DATA_MANAGEMENT_RESTORE.md new file mode 100644 index 000000000..4661ff9b1 --- /dev/null +++ b/docs/explanation/features/DATA_MANAGEMENT_RESTORE.md @@ -0,0 +1,80 @@ +# Data Management Restore + +Implemented in version: **0.250.106** + +## Overview + +Data Management Restore adds an admin-only recovery workflow for supported SimpleChat backups. Administrators can select a completed Data Management backup, run a preflight review, choose a non-destructive or explicitly confirmed overwrite policy, queue a durable restore job, and monitor progress through the existing Data Management job history. + +Restore is designed for deliberate recovery into configured target services. It does not expose backup contents, credentials, SAS URLs, storage keys, Cosmos account keys, Search keys, or internal endpoints to the browser. + +## Technical Specifications + +### Architecture + +- Admin API routes live in `route_backend_data_management.py` under the existing `backend_data_management` Blueprint and require `@swagger_route(security=get_auth_security())`, `@login_required`, and `@admin_required`. +- Restore planning, review, artifact reading, target writes, durable checkpoints, and retry/cancel support live in `functions_data_management.py`. +- Restore checkpoint primitives live in `functions_data_management_restore_state.py`. +- Restore jobs are stored in the existing `data_management_jobs` container with operation `restore`. +- Restore uses the existing `data_management_job_items` timeline and backup-manifest batch records to locate durable artifact paths. +- Restore activity is recorded in Activity Logs with the same Data Management audit path used by backup and migration jobs. + +### Restore Policies + +- **Create only** is the default and recommended policy. It creates missing target records and blocks or skips existing target collisions instead of overwriting them. +- **Overwrite existing target records** requires a separate confirmation phrase: `RESTORE WITH OVERWRITE`. +- Destructive behavior is never implicit. The UI disables queueing until a restore review is current, the final review acknowledgement is checked, and the overwrite phrase matches when overwrite mode is selected. + +### Preflight Review + +Restore review validates the selected backup and target before a job is queued: + +- Backup job is completed or completed with warnings. +- Manifest exists, has the expected SimpleChat schema, and matches the selected backup job. +- Failed backup resources block restore until the backup is retried or a different backup is selected. +- Partial backups warn that deletion replay is not supported because partial backup deletion policy is non-destructive. +- Target Cosmos DB is reachable and existing container partition keys match SimpleChat's expected contract. +- Target AI Search is reachable and existing indexes can be checked for create-only collisions. +- Target Enhanced Citation Storage is reachable when source blob restore is included. +- Review results are sanitized before returning to the browser. + +Ready reviews issue a short-lived, administrator-bound authorization. Job creation reserves that authorization before durable job creation and consumes it before restore execution starts, preventing replay or changed-plan execution. + +### Restore Execution + +Restore execution runs as a durable Data Management worker: + +1. Initializes a secret-free restore plan and checkpoint state. +2. Revalidates the selected backup manifest and target preflight. +3. Restores supported Cosmos DB JSONL artifact batches into configured target Cosmos containers. +4. Restores supported AI Search JSONL artifact pages after ensuring target indexes exist. +5. Restores supported Enhanced Citation source blob artifacts into target storage containers. +6. Persists per-resource progress, warning, collision, skip, failure, and retry counters. + +Queued restores can be canceled before execution. Running restores stop cooperatively at durable checkpoints. Failed, canceled, or stale restore jobs can be retried from their recorded restore state without changing the immutable plan. + +## Usage Instructions + +1. Open **Admin Settings > Data Management**. +2. Configure the target Cosmos DB, AI Search, and Enhanced Citation Storage settings. +3. In **Backup Inventory**, choose **Restore** on a completed backup row. +4. Select the restore policy and included surfaces. +5. Run **Restore Review** and resolve any blockers. +6. Confirm the reviewed plan. If overwrite mode is selected, type `RESTORE WITH OVERWRITE`. +7. Queue the restore job and monitor it in **Job History**. + +## Testing and Validation + +Coverage includes: + +- Restore state immutability and retry safety. +- Destructive overwrite confirmation. +- Manifest preflight blockers and partial-backup warnings. +- Admin-bound restore review authorizations. +- Admin UI restore controls and client-side route wiring. + +Known limitations: + +- Partial backups remain non-destructive and do not replay deletions. +- Restore requires backup storage and encryption-key access to remain available. +- Create-only restore blocks/skips existing target collisions; use overwrite only during an approved recovery window. diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index b902b880e..1227e160e 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -2,6 +2,15 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/). +### **(v0.250.106)** + +#### New Features + +* **Data Management Backup Restore Workflow** + * Added an admin-only restore workflow for completed Data Management backups, with manifest preflight, create-only default policy, explicit overwrite confirmation, durable restore jobs, cancellation/retry support, and sanitized progress in Job History. + * Restore supports configured target Cosmos DB, AI Search, and Enhanced Citation blob targets while preserving secret-safe review and job responses. + * (Ref: Closes #1091, `functions_data_management.py`, `functions_data_management_restore_state.py`, `route_backend_data_management.py`, `admin_settings.html`, `admin_data_management.js`, `DATA_MANAGEMENT_RESTORE.md`) + ### **(v0.250.105)** #### User Interface Enhancements diff --git a/functional_tests/test_data_management_history_pagination.py b/functional_tests/test_data_management_history_pagination.py index 4a0f220cc..454910add 100644 --- a/functional_tests/test_data_management_history_pagination.py +++ b/functional_tests/test_data_management_history_pagination.py @@ -2,10 +2,11 @@ #!/usr/bin/env python3 """ Functional test for Data Management history pagination. -Version: 0.250.105 +Version: 0.250.106 Implemented in: 0.250.103 Updated in: 0.250.104 Updated in: 0.250.105 +Updated in: 0.250.106 This test ensures job history and backup inventory use deterministic, filtered, sanitized Cosmos pages with opaque continuation state and global summaries. @@ -207,6 +208,7 @@ class FakeHistoryPaginationError(ValueError): imported_function_names = [ "create_data_management_migration_review_authorization", + "create_data_management_restore_review_authorization", "export_data_management_migration_manifest", "generate_data_management_encryption_key", "get_data_management_cosmos_editor_containers", @@ -215,6 +217,7 @@ class FakeHistoryPaginationError(ValueError): "get_data_management_job_progress", "get_data_management_migration_catalog", "get_data_management_migration_review_fingerprint", + "get_data_management_restore_review_fingerprint", "get_data_management_settings", "log_data_management_cosmos_editor_activity", "preview_data_management_migration_plan", @@ -222,11 +225,15 @@ class FakeHistoryPaginationError(ValueError): "query_data_management_cosmos_editor_documents", "request_data_management_job_cancellation", "release_data_management_migration_review_reservation", + "release_data_management_restore_review_reservation", "reserve_data_management_migration_review_authorization", + "reserve_data_management_restore_review_authorization", "review_data_management_migration", + "review_data_management_restore", "retry_data_management_backup_job", "resolve_data_management_migration_manifest_item", "retry_data_management_migration_job", + "retry_data_management_restore_job", "sanitize_data_management_job_for_admin", "sanitize_data_management_settings_for_admin", "save_data_management_cosmos_editor_document", diff --git a/functional_tests/test_data_management_restore_workflow.py b/functional_tests/test_data_management_restore_workflow.py new file mode 100644 index 000000000..0199b697c --- /dev/null +++ b/functional_tests/test_data_management_restore_workflow.py @@ -0,0 +1,294 @@ +#!/usr/bin/env python3 +# test_data_management_restore_workflow.py +""" +Functional tests for Data Management restore workflow safety. +Version: 0.250.106 +Implemented in: 0.250.106 + +This test ensures restore plans are immutable, destructive restore policies +require explicit confirmation, preflight blocks unsafe backups, and restore +review authorizations are admin-bound and single-use. +""" + +import copy +import importlib.util +from pathlib import Path +import sys +import types + +import pytest + + +REPO_ROOT = Path(__file__).resolve().parents[1] +APP_ROOT = REPO_ROOT / "application" / "single_app" +MODULE_PATH = APP_ROOT / "functions_data_management.py" +sys.path.insert(0, str(APP_ROOT)) + +from functions_data_management_restore_state import initialize_restore_state + + +class FakeJobContainer: + """Store Data Management jobs and review authorizations in memory.""" + + def __init__(self, documents=None): + self.documents = { + document["id"]: copy.deepcopy(document) + for document in (documents or []) + } + + def create_item(self, body): + if body["id"] in self.documents: + error = RuntimeError("conflict") + error.status_code = 409 + raise error + saved = copy.deepcopy(body) + saved["_etag"] = f"etag-{len(self.documents) + 1}" + self.documents[saved["id"]] = saved + return copy.deepcopy(saved) + + def read_item(self, item, partition_key): + del partition_key + if item not in self.documents: + raise KeyError(item) + return copy.deepcopy(self.documents[item]) + + def replace_item(self, item, body, etag=None, match_condition=None): + del match_condition + existing = self.documents[item] + if etag and existing.get("_etag") and etag != existing.get("_etag"): + error = RuntimeError("etag mismatch") + error.status_code = 412 + raise error + saved = copy.deepcopy(body) + saved["_etag"] = f"etag-{len(self.documents) + 1}" + self.documents[item] = saved + return copy.deepcopy(saved) + + def upsert_item(self, body): + saved = copy.deepcopy(body) + saved["_etag"] = f"etag-{len(self.documents) + 1}" + self.documents[saved["id"]] = saved + return copy.deepcopy(saved) + + def query_items(self, **_kwargs): + return iter(copy.deepcopy(list(self.documents.values()))) + + +def build_backup_job(status="completed"): + """Build a completed backup job with a restore-ready manifest pointer.""" + return { + "id": "backup-restore-source", + "type": "data_management_job", + "operation": "backup", + "backup_type": "full", + "status": status, + "created_at": "2026-07-31T12:00:00+00:00", + "completed_at": "2026-07-31T12:30:00+00:00", + "backup_plan": { + "backup_type": "full", + "source_scope": "simplechat-primary", + "source_cutoff_at": "2026-07-31T12:00:00+00:00", + "differential_mode": "full_snapshot", + "source_cutoff_semantics": {"deletion_policy": "none"}, + "include_cosmos": True, + "include_ai_search": True, + "include_source_blobs": True, + "backup_storage_container_name": "simplechat-backups", + "backup_storage_path_prefix": "simplechat-backups", + "storage_identity": "storage-fingerprint", + "encryption_enabled": False, + "encryption_key_storage": "not_configured", + "encryption_key_reference": "", + "encryption_key_fingerprint": "", + }, + "backup_state": { + "manifest": {"path": "simplechat-backups/full/manifest.json"}, + }, + "result": { + "manifest_path": "simplechat-backups/full/manifest.json", + "base_prefix": "simplechat-backups/full", + }, + } + + +def load_data_management_module(monkeypatch, job_container): + """Load production restore helpers with in-memory Cosmos dependencies.""" + config_module = types.ModuleType("config") + config_module.CLIENTS = {} + config_module.VERSION = "0.250.106" + config_module.cosmos_data_management_jobs_container = job_container + config_module.cosmos_data_management_job_items_container = job_container + config_module.cosmos_settings_container = job_container + config_module.SECRET_KEY = "unit-test-secret" + config_module.storage_account_user_documents_container_name = "user-documents" + config_module.storage_account_group_documents_container_name = "group-documents" + config_module.storage_account_public_documents_container_name = "public-documents" + config_module.storage_account_personal_chat_container_name = "personal-chat" + config_module.storage_account_group_chat_container_name = "group-chat" + monkeypatch.setitem(sys.modules, "config", config_module) + + appinsights_module = types.ModuleType("functions_appinsights") + appinsights_module.log_event = lambda *_args, **_kwargs: None + monkeypatch.setitem(sys.modules, "functions_appinsights", appinsights_module) + + throughput_module = types.ModuleType("functions_cosmos_throughput") + + class FakeCosmosThroughputError(Exception): + pass + + throughput_module.CosmosThroughputError = FakeCosmosThroughputError + throughput_module.get_container_throughput = lambda *_args, **_kwargs: {} + throughput_module.get_database_throughput = lambda *_args, **_kwargs: {} + throughput_module.set_database_throughput = lambda *_args, **_kwargs: {} + monkeypatch.setitem(sys.modules, "functions_cosmos_throughput", throughput_module) + + module_name = "data_management_restore_workflow_test_module" + spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH) + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + assert spec and spec.loader + spec.loader.exec_module(module) + monkeypatch.delitem(sys.modules, module_name, raising=False) + return module + + +def test_restore_state_rejects_changed_plan(): + """Verify restore checkpoints cannot resume with changed policy.""" + plan = { + "source_backup_id": "backup-restore-source", + "restore_policy": "create_only", + } + state = initialize_restore_state( + None, + "restore-job", + plan, + ) + + with pytest.raises(ValueError, match="Restore plan changed"): + initialize_restore_state( + state, + "restore-job", + {**plan, "restore_policy": "overwrite_existing"}, + ) + + +def test_restore_plan_requires_completed_backup_and_overwrite_confirmation(monkeypatch): + """Validate backup status and destructive confirmation gates.""" + job_container = FakeJobContainer([build_backup_job()]) + module = load_data_management_module(monkeypatch, job_container) + + create_only = module._normalize_data_management_restore_plan({ + "source_backup_id": "backup-restore-source", + }) + assert create_only["restore_policy"] == "create_only" + assert create_only["source_backup_manifest_path"].endswith("manifest.json") + + with pytest.raises( + module.DataManagementSettingsValidationError, + match="Overwrite restore requires", + ): + module._normalize_data_management_restore_plan( + { + "source_backup_id": "backup-restore-source", + "restore_policy": "overwrite_existing", + }, + require_confirmation=True, + ) + + overwrite = module._normalize_data_management_restore_plan( + { + "source_backup_id": "backup-restore-source", + "restore_policy": "overwrite_existing", + "overwrite_confirmed": True, + "overwrite_confirmation_phrase": "RESTORE WITH OVERWRITE", + }, + require_confirmation=True, + ) + assert overwrite["restore_policy"] == "overwrite_existing" + + failed_container = FakeJobContainer([build_backup_job(status="failed")]) + failed_module = load_data_management_module(monkeypatch, failed_container) + with pytest.raises( + failed_module.DataManagementSettingsValidationError, + match="Only completed backups", + ): + failed_module._normalize_data_management_restore_plan({ + "source_backup_id": "backup-restore-source", + }) + + +def test_restore_preflight_blocks_failed_manifest_and_warns_partial(monkeypatch): + """Validate restore review blocks incomplete backups and documents partial semantics.""" + job_container = FakeJobContainer([build_backup_job()]) + module = load_data_management_module(monkeypatch, job_container) + restore_plan = module._normalize_data_management_restore_plan({ + "source_backup_id": "backup-restore-source", + "include_cosmos": False, + "include_ai_search": False, + "include_source_blobs": False, + }) + restore_plan["differential_mode"] = "latest_item_state" + manifest = { + "schema_version": 2, + "app": "SimpleChat", + "failed_resource_names": ["cosmos:users"], + "warnings": [], + "artifacts": [], + } + + review = module._run_data_management_restore_preflight( + {}, + restore_plan, + build_backup_job(), + manifest, + ) + + assert review["ready"] is False + assert review["blocker_count"] == 1 + assert any(check["id"] == "manifest_integrity" and check["status"] == "block" for check in review["checks"]) + assert any(check["id"] == "partial_semantics" and check["status"] == "warn" for check in review["checks"]) + + +def test_restore_review_authorization_is_admin_bound_and_single_use(monkeypatch): + """Verify ready restore review authorizations cannot be replayed by another admin.""" + job_container = FakeJobContainer([build_backup_job()]) + module = load_data_management_module(monkeypatch, job_container) + authorization = module.create_data_management_restore_review_authorization( + "admin-user", + "restore-fingerprint", + ) + reservation = module.reserve_data_management_restore_review_authorization( + authorization["authorization_token"], + "admin-user", + "restore-fingerprint", + ) + + with pytest.raises( + module.DataManagementSettingsValidationError, + match="reserved or used", + ): + module.reserve_data_management_restore_review_authorization( + authorization["authorization_token"], + "admin-user", + "restore-fingerprint", + ) + + with pytest.raises( + module.DataManagementSettingsValidationError, + match="invalid or expired", + ): + module.consume_data_management_restore_review_authorization( + authorization["authorization_token"], + "different-admin", + "restore-fingerprint", + reservation["reservation_token"], + reservation["job_id"], + ) + + assert module.consume_data_management_restore_review_authorization( + authorization["authorization_token"], + "admin-user", + "restore-fingerprint", + reservation["reservation_token"], + reservation["job_id"], + ) is True diff --git a/ui_tests/test_admin_data_management_settings_ui.py b/ui_tests/test_admin_data_management_settings_ui.py index a6fa14313..f7cda7e1c 100644 --- a/ui_tests/test_admin_data_management_settings_ui.py +++ b/ui_tests/test_admin_data_management_settings_ui.py @@ -1,13 +1,14 @@ # test_admin_data_management_settings_ui.py """ UI test for Admin Settings Data Management controls. -Version: 0.250.105 +Version: 0.250.106 Implemented in: 0.241.211 Updated in: 0.241.221 Updated in: 0.250.102 Updated in: 0.250.103 Updated in: 0.250.104 Updated in: 0.250.105 +Updated in: 0.250.106 This test ensures admins can discover the Data Management tab, see the operational-business-hours warning, and access the backup, encryption, @@ -201,6 +202,20 @@ def test_admin_data_management_controls_render_from_template(): "data-management-backup-pagination-status", "data-management-backup-previous-page-btn", "data-management-backup-next-page-btn", + "data-management-restore-modal", + "data-management-restore-title", + "data-management-restore-backup-summary", + "data_management_restore_policy", + "data_management_restore_include_cosmos", + "data_management_restore_include_ai_search", + "data_management_restore_include_source_blobs", + "data-management-restore-confirmation-section", + "data_management_restore_overwrite_confirmation_phrase", + "data-management-restore-review-btn", + "data-management-restore-review-status", + "data-management-restore-review-checks", + "data_management_restore_final_confirmation", + "data-management-restore-queue-btn", "data-management-jobs-tbody", "data_management_job_operation_filter", "data_management_job_status_filter", @@ -245,6 +260,13 @@ def test_admin_data_management_controls_render_from_template(): assert "I understand this editor can damage overall system health." in template assert "Required phrase: I understand this can damage system data" in template assert '

Backup Inventory

' in template + assert "Restore Backup" in template + assert "Run Restore Review" in template + assert "Queue Restore Job" in template + assert "Required phrase: RESTORE WITH OVERWRITE" in template + assert "openRestoreModal" in js_source + assert 'requestJson("/api/admin/data-management/restore/review"' in js_source + assert 'queueOperation("restore"' in js_source assert 'aria-label="Backup inventory filters"' in template assert 'Available backups' in template assert 'Backup' in template