from __future__ import annotations import hashlib from collections import defaultdict from collections.abc import Sequence from typing import Any from sqlalchemy import or_ from sqlalchemy.orm import Session from govoplan_calendar.backend.db.models import ( CalendarCollection, CalendarEvent, CalendarMigrationBatch, CalendarMigrationResource, CalendarOutboxOperation, CalendarSyncSource, ) from govoplan_calendar.backend.ical import events_to_ics from govoplan_core.audit.logging import audit_event from govoplan_core.db.base import utcnow REMOTE_MOVE_CONFIRMATION = "MOVE REMOTE EVENTS" ACTIVE_MIGRATION_STATUSES = {"active", "blocked", "cancel_requested"} BLOCKING_OUTBOX_STATUSES = {"conflict", "dead", "cancelled"} class CalendarMigrationError(ValueError): pass def active_tenant_migration( session: Session, *, tenant_id: str, ) -> CalendarMigrationBatch | None: return ( session.query(CalendarMigrationBatch) .filter( CalendarMigrationBatch.tenant_id == tenant_id, CalendarMigrationBatch.status.in_(sorted(ACTIVE_MIGRATION_STATUSES)), ) .order_by(CalendarMigrationBatch.created_at.asc()) .first() ) def active_calendar_migration( session: Session, *, tenant_id: str, calendar_id: str, ) -> CalendarMigrationBatch | None: return ( session.query(CalendarMigrationBatch) .filter( CalendarMigrationBatch.tenant_id == tenant_id, CalendarMigrationBatch.status.in_(sorted(ACTIVE_MIGRATION_STATUSES)), or_( CalendarMigrationBatch.source_calendar_id == calendar_id, CalendarMigrationBatch.target_calendar_id == calendar_id, ), ) .order_by(CalendarMigrationBatch.created_at.asc()) .first() ) def active_source_migration( session: Session, *, tenant_id: str, source_id: str, ) -> CalendarMigrationBatch | None: return ( session.query(CalendarMigrationBatch) .filter( CalendarMigrationBatch.tenant_id == tenant_id, CalendarMigrationBatch.status.in_(sorted(ACTIVE_MIGRATION_STATUSES)), or_( CalendarMigrationBatch.source_sync_source_id == source_id, CalendarMigrationBatch.target_sync_source_id == source_id, ), ) .order_by(CalendarMigrationBatch.created_at.asc()) .first() ) def assert_calendar_not_migrating( session: Session, *, tenant_id: str, calendar_id: str, ) -> None: batch = active_calendar_migration( session, tenant_id=tenant_id, calendar_id=calendar_id, ) if batch is not None: raise CalendarMigrationError( "Calendar changes are blocked while remote move " f"{batch.id} is {batch.phase}." ) def assert_source_not_migrating( session: Session, *, tenant_id: str, source_id: str, ) -> None: batch = active_source_migration( session, tenant_id=tenant_id, source_id=source_id, ) if batch is not None: raise CalendarMigrationError( "Calendar source changes and synchronization are blocked while " f"remote move {batch.id} is {batch.phase}." ) def event_migration_batch_id(event: CalendarEvent) -> str | None: metadata = event.metadata_ if isinstance(event.metadata_, dict) else {} migration = metadata.get("calendar_migration") if not isinstance(migration, dict) or not migration.get("locked"): return None value = str(migration.get("batch_id") or "").strip() return value or None def assert_event_not_migrating(event: CalendarEvent) -> None: batch_id = event_migration_batch_id(event) if batch_id: raise CalendarMigrationError( "Event changes are blocked while destructive Calendar move " f"{batch_id} is being reconciled." ) def _validate_remote_move_sources( source: CalendarSyncSource, target: CalendarSyncSource, ) -> None: for label, item in (("source", source), ("target", target)): if ( item.source_kind != "caldav" or item.sync_direction != "two_way" or not item.sync_enabled or item.deleted_at is not None ): raise CalendarMigrationError( f"Remote move {label} must be an active two-way CalDAV source." ) def _validate_remote_move_authorization( *, confirmation: str | None, evidence_note: str | None, ) -> str: if confirmation != REMOTE_MOVE_CONFIRMATION: raise CalendarMigrationError( f"Type {REMOTE_MOVE_CONFIRMATION!r} to authorize the destructive remote move." ) evidence = str(evidence_note or "").strip() if len(evidence) < 10: raise CalendarMigrationError( "Destructive remote moves require an evidence note of at least 10 characters." ) return evidence def _group_source_resources( events: Sequence[CalendarEvent], ) -> dict[str, list[CalendarEvent]]: grouped: dict[str, list[CalendarEvent]] = defaultdict(list) for event in events: if event.source_kind != "caldav" or not event.source_href or not event.etag: raise CalendarMigrationError( "Every moved event must have a synchronized CalDAV href and ETag; " "synchronize and reconcile the source before moving it." ) grouped[event.source_href].append(event) if not grouped: raise CalendarMigrationError( "Remote move requires at least one synchronized Calendar resource." ) for href, resource_events in grouped.items(): etags = {event.etag for event in resource_events} if len(etags) != 1: raise CalendarMigrationError( f"CalDAV resource {href!r} has inconsistent ETags; synchronize before moving it." ) return dict(grouped) def _assert_no_uid_collisions( session: Session, *, tenant_id: str, target_calendar_id: str, events: Sequence[CalendarEvent], ) -> None: incoming = {(event.uid, event.recurrence_id) for event in events} existing = { (uid, recurrence_id) for uid, recurrence_id in ( session.query(CalendarEvent.uid, CalendarEvent.recurrence_id) .filter( CalendarEvent.tenant_id == tenant_id, CalendarEvent.calendar_id == target_calendar_id, CalendarEvent.deleted_at.is_(None), ) .all() ) } collisions = sorted(incoming & existing, key=lambda item: (item[0], item[1] or "")) if collisions: uid, recurrence_id = collisions[0] suffix = f" recurrence {recurrence_id}" if recurrence_id else "" raise CalendarMigrationError( f"Target calendar already contains UID {uid!r}{suffix}; UIDs are " "preserved and the collision must be resolved first." ) def _destination_href( *, source: CalendarSyncSource, target: CalendarSyncSource, source_href: str, ) -> str: from govoplan_calendar.backend.service import normalize_caldav_href digest = hashlib.sha256( f"{source.id}\0{target.id}\0{source_href}".encode("utf-8") ).hexdigest()[:40] return normalize_caldav_href( target.collection_url, f"govoplan-move-{digest}.ics", ) def _set_calendar_migration_metadata( calendar: CalendarCollection, *, batch: CalendarMigrationBatch, role: str, ) -> None: metadata = dict(calendar.metadata_ or {}) metadata["remote_move"] = { "batch_id": batch.id, "role": role, "status": batch.status, "phase": batch.phase, "source_calendar_id": batch.source_calendar_id, "target_calendar_id": batch.target_calendar_id, } calendar.metadata_ = metadata def _set_event_migration_metadata( event: CalendarEvent, *, batch: CalendarMigrationBatch, resource: CalendarMigrationResource, ) -> None: metadata = dict(event.metadata_ or {}) metadata.pop("caldav", None) metadata["calendar_migration"] = { "batch_id": batch.id, "resource_id": resource.id, "locked": True, "source_href": resource.source_href, "destination_href": resource.destination_href, } event.metadata_ = metadata def start_remote_move( session: Session, *, tenant_id: str, source_calendar: CalendarCollection, target_calendar: CalendarCollection, source: CalendarSyncSource, target_source: CalendarSyncSource, events: Sequence[CalendarEvent], previous_event_states: dict[str, dict[str, Any]], make_target_default: bool, confirmation: str | None, evidence_note: str | None, user_id: str | None, api_key_id: str | None, ) -> CalendarMigrationBatch: from govoplan_calendar.backend.outbox import ( calendar_outbox_has_live_lease, calendar_outbox_has_unresolved_desired_state, enqueue_caldav_desired_state, ) from govoplan_calendar.backend.service import ( clear_default_calendar, record_calendar_event_change, ) evidence = _validate_remote_move_authorization( confirmation=confirmation, evidence_note=evidence_note, ) _validate_remote_move_sources(source, target_source) if active_calendar_migration( session, tenant_id=tenant_id, calendar_id=source_calendar.id, ) or active_calendar_migration( session, tenant_id=tenant_id, calendar_id=target_calendar.id, ): raise CalendarMigrationError( "Source or target calendar already participates in an active remote move." ) for item in (source, target_source): if calendar_outbox_has_live_lease(session, source_id=item.id): raise CalendarMigrationError( "Remote move cannot start while either CalDAV source has an active delivery lease." ) if calendar_outbox_has_unresolved_desired_state(session, source_id=item.id): raise CalendarMigrationError( "Remote move requires both CalDAV sources to have no unresolved outbound desired state." ) grouped = _group_source_resources(events) _assert_no_uid_collisions( session, tenant_id=tenant_id, target_calendar_id=target_calendar.id, events=events, ) batch = CalendarMigrationBatch( tenant_id=tenant_id, source_calendar_id=source_calendar.id, target_calendar_id=target_calendar.id, source_sync_source_id=source.id, target_sync_source_id=target_source.id, total_resources=len(grouped), total_events=len(events), created_by_user_id=user_id, created_by_api_key_id=api_key_id, authorization_evidence={ "confirmation": confirmation, "note": evidence, "authorized_at": utcnow().isoformat(), "source_sync_enabled": bool(source.sync_enabled), "source_was_default": bool(source_calendar.is_default), "target_was_default": bool(target_calendar.is_default), }, ) session.add(batch) session.flush() _set_calendar_migration_metadata(source_calendar, batch=batch, role="source") _set_calendar_migration_metadata(target_calendar, batch=batch, role="target") for source_href, resource_events in sorted(grouped.items()): destination_href = _destination_href( source=source, target=target_source, source_href=source_href, ) resource = CalendarMigrationResource( tenant_id=tenant_id, batch_id=batch.id, source_href=source_href, source_expected_etag=str(resource_events[0].etag), destination_href=destination_href, event_ids=[event.id for event in resource_events], ) session.add(resource) session.flush() for event in resource_events: event.calendar_id = target_calendar.id event.source_kind = "local" event.source_href = None event.etag = None _set_event_migration_metadata( event, batch=batch, resource=resource, ) session.flush() destination_operation = enqueue_caldav_desired_state( session, source=target_source, trigger_event=resource_events[0], href=destination_href, resource_events=list(resource_events), payload_ics=events_to_ics(list(resource_events)), expected_etag=None, idempotency_context=( f"calendar-migration:{batch.id}:{resource.id}:destination" ), ) destination_metadata = dict(destination_operation.metadata_ or {}) destination_metadata.update( { "calendar_migration_batch_id": batch.id, "calendar_migration_resource_id": resource.id, "calendar_migration_role": "destination_put", "overwrite": False, } ) destination_operation.metadata_ = destination_metadata source_delete_operation = enqueue_caldav_desired_state( session, source=source, trigger_event=None, href=source_href, resource_events=[], payload_ics=None, expected_etag=resource.source_expected_etag, idempotency_context=( f"calendar-migration:{batch.id}:{resource.id}:source-delete" ), ) source_delete_metadata = dict(source_delete_operation.metadata_ or {}) source_delete_metadata.update( { "calendar_migration_batch_id": batch.id, "calendar_migration_resource_id": resource.id, "calendar_migration_role": "source_delete", "depends_on_operation_id": destination_operation.id, "overwrite": False, } ) source_delete_operation.metadata_ = source_delete_metadata resource.destination_operation_id = destination_operation.id resource.source_delete_operation_id = source_delete_operation.id for event in resource_events: record_calendar_event_change( session, event=event, operation="updated", user_id=user_id, previous=previous_event_states[event.id], ) source.sync_enabled = False source.next_sync_at = None source_metadata = dict(source.metadata_ or {}) source_metadata["remote_move_batch_id"] = batch.id source.metadata_ = source_metadata if make_target_default: clear_default_calendar(session, tenant_id=tenant_id) target_calendar.is_default = True audit_event( session, tenant_id=tenant_id, user_id=user_id, api_key_id=api_key_id, action="calendar.remote_move.started", object_type="calendar_migration_batch", object_id=batch.id, details={ "source_calendar_id": source_calendar.id, "target_calendar_id": target_calendar.id, "source_sync_source_id": source.id, "target_sync_source_id": target_source.id, "event_count": len(events), "resource_count": len(grouped), "evidence_note": evidence, }, ) session.flush() return batch def migration_operation_dependency_ready( session: Session, operation: CalendarOutboxOperation, ) -> bool: metadata = operation.metadata_ if isinstance(operation.metadata_, dict) else {} if metadata.get("calendar_migration_role") != "source_delete": return True batch_id = str(metadata.get("calendar_migration_batch_id") or "") batch = session.get(CalendarMigrationBatch, batch_id) if batch_id else None if batch is None or batch.status in {"cancel_requested", "cancelled"}: return False destination_ids = [ value for (value,) in session.query( CalendarMigrationResource.destination_operation_id ) .filter(CalendarMigrationResource.batch_id == batch.id) .all() if value ] if not destination_ids: return False succeeded = int( session.query(CalendarOutboxOperation) .filter( CalendarOutboxOperation.id.in_(destination_ids), CalendarOutboxOperation.status == "succeeded", ) .count() ) return succeeded == len(destination_ids) def _operation_by_id( session: Session, operation_id: str | None, ) -> CalendarOutboxOperation | None: return session.get(CalendarOutboxOperation, operation_id) if operation_id else None def _update_resource_state( resource: CalendarMigrationResource, destination: CalendarOutboxOperation | None, source_delete: CalendarOutboxOperation | None, ) -> None: if destination is None or source_delete is None: resource.status = "invalid" resource.last_error = "Migration outbox operation is missing." return if destination.status in BLOCKING_OUTBOX_STATUSES: resource.status = "destination_blocked" resource.last_error = destination.last_error or destination.status return if destination.status != "succeeded": resource.status = "copy_pending" resource.last_error = destination.last_error return if source_delete.status in BLOCKING_OUTBOX_STATUSES: resource.status = ( "source_retained" if source_delete.status == "cancelled" else "source_blocked" ) resource.last_error = source_delete.last_error return if source_delete.status == "succeeded": resource.status = "completed" resource.last_error = None return resource.status = ( "delete_in_progress" if source_delete.attempt_count else "copy_succeeded" ) resource.last_error = source_delete.last_error def _clear_migration_metadata( calendar: CalendarCollection | None, ) -> None: if calendar is None: return metadata = dict(calendar.metadata_ or {}) metadata.pop("remote_move", None) calendar.metadata_ = metadata def _unlock_batch_events(session: Session, batch: CalendarMigrationBatch) -> None: event_ids = { str(event_id) for resource in batch.resources for event_id in resource.event_ids or [] } if not event_ids: return for event in ( session.query(CalendarEvent) .filter( CalendarEvent.tenant_id == batch.tenant_id, CalendarEvent.id.in_(sorted(event_ids)), ) .all() ): metadata = dict(event.metadata_ or {}) metadata.pop("calendar_migration", None) event.metadata_ = metadata def _finalize_completed_batch( session: Session, batch: CalendarMigrationBatch, ) -> None: from govoplan_calendar.backend.service import retire_sync_source now = utcnow() source_calendar = session.get(CalendarCollection, batch.source_calendar_id) target_calendar = session.get(CalendarCollection, batch.target_calendar_id) source = session.get(CalendarSyncSource, batch.source_sync_source_id) _unlock_batch_events(session, batch) _clear_migration_metadata(source_calendar) _clear_migration_metadata(target_calendar) if source_calendar is not None: source_calendar.is_default = False source_calendar.deleted_at = now if source is not None and source.deleted_at is None: retire_sync_source( session, tenant_id=batch.tenant_id, source=source, deleted_at=now, deletion_reason="remote_move_completed", user_id=batch.created_by_user_id, api_key_id=batch.created_by_api_key_id, ) batch.status = "completed" batch.phase = "completed" batch.completed_at = now batch.last_error = None audit_event( session, tenant_id=batch.tenant_id, user_id=None, action="calendar.remote_move.completed", object_type="calendar_migration_batch", object_id=batch.id, details={ "source_calendar_id": batch.source_calendar_id, "target_calendar_id": batch.target_calendar_id, "event_count": batch.total_events, "resource_count": batch.total_resources, }, ) def _finalize_cancelled_batch( session: Session, batch: CalendarMigrationBatch, ) -> None: now = utcnow() source_calendar = session.get(CalendarCollection, batch.source_calendar_id) target_calendar = session.get(CalendarCollection, batch.target_calendar_id) source = session.get(CalendarSyncSource, batch.source_sync_source_id) _unlock_batch_events(session, batch) _clear_migration_metadata(source_calendar) _clear_migration_metadata(target_calendar) if source_calendar is not None: source_calendar.is_default = bool( (batch.authorization_evidence or {}).get("source_was_default", False) ) if target_calendar is not None: target_calendar.is_default = bool( (batch.authorization_evidence or {}).get("target_was_default", False) ) if source is not None and source.deleted_at is None: source.sync_enabled = bool( (batch.authorization_evidence or {}).get("source_sync_enabled", True) ) source.next_sync_at = now if source.sync_enabled else None metadata = dict(source.metadata_ or {}) metadata.pop("remote_move_batch_id", None) source.metadata_ = metadata batch.status = "cancelled" batch.phase = "cancelled_source_retained" batch.completed_at = now def refresh_calendar_migration_batch( session: Session, *, batch_id: str, ) -> CalendarMigrationBatch: batch = ( session.query(CalendarMigrationBatch) .filter(CalendarMigrationBatch.id == batch_id) .populate_existing() .with_for_update() .one_or_none() ) if batch is None: raise CalendarMigrationError("Calendar migration batch not found.") if batch.status in {"completed", "cancelled"}: return batch destination_states: list[str] = [] delete_states: list[str] = [] errors: list[str] = [] for resource in batch.resources: destination = _operation_by_id(session, resource.destination_operation_id) source_delete = _operation_by_id(session, resource.source_delete_operation_id) _update_resource_state(resource, destination, source_delete) destination_states.append(destination.status if destination else "missing") delete_states.append(source_delete.status if source_delete else "missing") if resource.last_error: errors.append(resource.last_error) all_destinations_succeeded = bool(destination_states) and all( value == "succeeded" for value in destination_states ) all_deletes_succeeded = bool(delete_states) and all( value == "succeeded" for value in delete_states ) if batch.status == "cancel_requested": if all( value in {"succeeded", "conflict", "dead", "cancelled"} for value in destination_states ): _finalize_cancelled_batch(session, batch) else: batch.phase = "finishing_destination_copies" elif all_deletes_succeeded: _finalize_completed_batch(session, batch) elif any(value in BLOCKING_OUTBOX_STATUSES for value in destination_states): batch.status = "blocked" batch.phase = "destination_conflict" elif all_destinations_succeeded and any( value in {"conflict", "dead"} for value in delete_states ): batch.status = "blocked" batch.phase = "source_delete_conflict" elif all_destinations_succeeded: batch.status = "active" batch.phase = "deleting_source" else: batch.status = "active" batch.phase = "copying_destination" batch.last_error = errors[0][:4000] if errors else None source_calendar = session.get(CalendarCollection, batch.source_calendar_id) target_calendar = session.get(CalendarCollection, batch.target_calendar_id) if batch.status in ACTIVE_MIGRATION_STATUSES: if source_calendar is not None: _set_calendar_migration_metadata( source_calendar, batch=batch, role="source" ) if target_calendar is not None: _set_calendar_migration_metadata( target_calendar, batch=batch, role="target" ) session.flush() return batch def refresh_migration_for_operation( session: Session, operation: CalendarOutboxOperation, ) -> None: metadata = operation.metadata_ if isinstance(operation.metadata_, dict) else {} batch_id = str(metadata.get("calendar_migration_batch_id") or "").strip() if batch_id: refresh_calendar_migration_batch(session, batch_id=batch_id) def get_calendar_migration_batch( session: Session, *, tenant_id: str, batch_id: str, refresh: bool = True, ) -> CalendarMigrationBatch: batch = ( session.query(CalendarMigrationBatch) .filter( CalendarMigrationBatch.tenant_id == tenant_id, CalendarMigrationBatch.id == batch_id, ) .one_or_none() ) if batch is None: raise CalendarMigrationError("Calendar migration batch not found.") return ( refresh_calendar_migration_batch(session, batch_id=batch.id) if refresh and batch.status in ACTIVE_MIGRATION_STATUSES else batch ) def list_calendar_migration_batches( session: Session, *, tenant_id: str, calendar_id: str | None = None, limit: int = 100, ) -> list[CalendarMigrationBatch]: query = session.query(CalendarMigrationBatch).filter( CalendarMigrationBatch.tenant_id == tenant_id ) if calendar_id: query = query.filter( or_( CalendarMigrationBatch.source_calendar_id == calendar_id, CalendarMigrationBatch.target_calendar_id == calendar_id, ) ) batches = ( query.order_by( CalendarMigrationBatch.created_at.desc(), CalendarMigrationBatch.id.desc(), ) .limit(max(1, min(limit, 500))) .all() ) return [ refresh_calendar_migration_batch(session, batch_id=batch.id) if batch.status in ACTIVE_MIGRATION_STATUSES else batch for batch in batches ] def cancel_calendar_migration_batch( session: Session, *, tenant_id: str, batch_id: str, evidence_note: str, user_id: str | None, api_key_id: str | None, ) -> CalendarMigrationBatch: note = evidence_note.strip() if len(note) < 10: raise CalendarMigrationError( "Cancelling a remote move requires an evidence note of at least " "10 characters." ) batch = get_calendar_migration_batch( session, tenant_id=tenant_id, batch_id=batch_id, refresh=False, ) if batch.status not in {"active", "blocked"}: raise CalendarMigrationError( f"Calendar migration cannot be cancelled while it is {batch.status}." ) delete_ids = [ resource.source_delete_operation_id for resource in batch.resources if resource.source_delete_operation_id ] session.query(CalendarSyncSource).filter( CalendarSyncSource.id == batch.source_sync_source_id, CalendarSyncSource.tenant_id == tenant_id, ).with_for_update().one() delete_operations = ( session.query(CalendarOutboxOperation) .filter(CalendarOutboxOperation.id.in_(delete_ids)) .order_by(CalendarOutboxOperation.id.asc()) .with_for_update() .all() ) batch = refresh_calendar_migration_batch(session, batch_id=batch.id) if batch.status not in {"active", "blocked"}: raise CalendarMigrationError( f"Calendar migration cannot be cancelled while it is {batch.status}." ) if any( operation.attempt_count > 0 or operation.status in {"in_progress", "succeeded"} for operation in delete_operations ): raise CalendarMigrationError( "Cancellation is no longer safe because conditional source deletion " "has started; reconcile the batch to completion." ) now = utcnow() for operation in delete_operations: operation.status = "cancelled" operation.completed_at = now operation.last_error = "Remote move cancelled before source deletion." operation.lease_token = None operation.lease_expires_at = None batch.status = "cancel_requested" batch.phase = "finishing_destination_copies" batch.cancellation_evidence = { "note": note, "cancelled_at": now.isoformat(), "cancelled_by_user_id": user_id, "cancelled_by_api_key_id": api_key_id, } audit_event( session, tenant_id=tenant_id, user_id=user_id, api_key_id=api_key_id, action="calendar.remote_move.cancel_requested", object_type="calendar_migration_batch", object_id=batch.id, details={"evidence_note": note}, ) return refresh_calendar_migration_batch(session, batch_id=batch.id) def calendar_migration_response( session: Session, batch: CalendarMigrationBatch, ) -> dict[str, Any]: resources: list[dict[str, Any]] = [] copied = 0 deleted = 0 conflicts = 0 source_delete_started = False for resource in batch.resources: destination = _operation_by_id(session, resource.destination_operation_id) source_delete = _operation_by_id(session, resource.source_delete_operation_id) if destination and destination.status == "succeeded": copied += 1 if source_delete and source_delete.status == "succeeded": deleted += 1 if source_delete and ( source_delete.attempt_count > 0 or source_delete.status in {"in_progress", "succeeded"} ): source_delete_started = True if resource.status in {"destination_blocked", "source_blocked", "invalid"}: conflicts += 1 resources.append( { "id": resource.id, "source_href": resource.source_href, "destination_href": resource.destination_href, "event_ids": list(resource.event_ids or []), "status": resource.status, "destination_operation_id": resource.destination_operation_id, "destination_operation_status": destination.status if destination else None, "source_delete_operation_id": resource.source_delete_operation_id, "source_delete_operation_status": source_delete.status if source_delete else None, "last_error": resource.last_error, } ) return { "id": batch.id, "migration_kind": batch.migration_kind, "status": batch.status, "phase": batch.phase, "source_calendar_id": batch.source_calendar_id, "target_calendar_id": batch.target_calendar_id, "source_sync_source_id": batch.source_sync_source_id, "target_sync_source_id": batch.target_sync_source_id, "total_resources": batch.total_resources, "copied_resources": copied, "deleted_source_resources": deleted, "conflict_count": conflicts, "total_events": batch.total_events, "last_error": batch.last_error, "can_cancel": batch.status in {"active", "blocked"} and not source_delete_started, "authorization_evidence": dict(batch.authorization_evidence or {}), "cancellation_evidence": ( dict(batch.cancellation_evidence) if batch.cancellation_evidence else None ), "created_by_user_id": batch.created_by_user_id, "created_by_api_key_id": batch.created_by_api_key_id, "created_at": batch.created_at, "updated_at": batch.updated_at, "completed_at": batch.completed_at, "resources": resources, } def migration_source_ids_in_progress( session: Session, *, tenant_id: str | None = None, ) -> set[str]: query = session.query( CalendarMigrationBatch.source_sync_source_id, CalendarMigrationBatch.target_sync_source_id, ).filter(CalendarMigrationBatch.status.in_(sorted(ACTIVE_MIGRATION_STATUSES))) if tenant_id is not None: query = query.filter(CalendarMigrationBatch.tenant_id == tenant_id) return {source_id for row in query.all() for source_id in row if source_id} __all__ = [ "ACTIVE_MIGRATION_STATUSES", "CalendarMigrationError", "REMOTE_MOVE_CONFIRMATION", "active_calendar_migration", "active_source_migration", "active_tenant_migration", "assert_calendar_not_migrating", "assert_event_not_migrating", "assert_source_not_migrating", "calendar_migration_response", "cancel_calendar_migration_batch", "get_calendar_migration_batch", "list_calendar_migration_batches", "migration_operation_dependency_ready", "migration_source_ids_in_progress", "refresh_calendar_migration_batch", "refresh_migration_for_operation", "start_remote_move", ]