diff --git a/docs/CALENDAR_INTEGRATION_CONCEPT.md b/docs/CALENDAR_INTEGRATION_CONCEPT.md index 587dd79..08f80a9 100644 --- a/docs/CALENDAR_INTEGRATION_CONCEPT.md +++ b/docs/CALENDAR_INTEGRATION_CONCEPT.md @@ -159,12 +159,28 @@ Collection deletion accepts two explicit, non-destructive external actions for rolling back or hiding the local move. Local-to-local moves retain their existing behavior and do not accept an -`external_action`. Implicit or mismatched actions, inbound-only destinations, -and synchronized-to-synchronized moves are rejected. `remote_move` is a known -but deliberately unsupported action: implementing it requires a separately -approved saga for destination reconciliation, conditional source deletion, -collision handling, and concurrent-edit policy. The WebUI does not offer that -destructive mode. +`external_action`. Implicit or mismatched actions and inbound-only destinations +are rejected. `remote_move` applies only between active, enabled, two-way +CalDAV sources. It is an administrator-authorized durable migration saga: + +- the request requires the exact `MOVE REMOTE EVENTS` confirmation and a + retained authorization-evidence note; +- all destination resources must complete or reconcile their conditional PUTs + before any source DELETE can be leased; +- every source DELETE uses the ETag captured when the batch started, so a + concurrent remote edit becomes an explicit conflict rather than data loss; +- UIDs are preserved and target UID collisions stop the batch before mutation; +- both calendars, their sources, and moved events reject ordinary edits and + synchronization while the batch is active; +- progress, resource states, conflicts, authorization evidence, and actor + provenance remain queryable; and +- cancellation is available only before the first source DELETE attempt. It + finishes safe destination copies, retains the source resources, and restores + source synchronization. Once deletion starts, the batch must be reconciled. + +The source collection is retired only after every source resource is confirmed +absent. A crash after a destination write is reconciled by semantic ICS content +before retry, preserving the outbox's no-blind-repeat guarantee. Resolved terminal rows (`succeeded`, `superseded`, and `cancelled`) are removed in bounded batches after `CALENDAR_OUTBOX_TERMINAL_RETENTION_DAYS` (90 days by diff --git a/src/govoplan_calendar/backend/db/models.py b/src/govoplan_calendar/backend/db/models.py index 7741abe..5ac77b9 100644 --- a/src/govoplan_calendar/backend/db/models.py +++ b/src/govoplan_calendar/backend/db/models.py @@ -256,9 +256,63 @@ class CalendarOutboxOperation(Base, TimestampMixin): event: Mapped[CalendarEvent | None] = relationship() +class CalendarMigrationBatch(Base, TimestampMixin): + """Durable destructive remote-move saga and its authorization evidence.""" + + __tablename__ = "calendar_migration_batches" + __table_args__ = ( + Index("ix_calendar_migration_batches_tenant_status", "tenant_id", "status", "created_at"), + Index("ix_calendar_migration_batches_calendars", "tenant_id", "source_calendar_id", "target_calendar_id"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + migration_kind: Mapped[str] = mapped_column(String(30), default="remote_move", nullable=False) + status: Mapped[str] = mapped_column(String(30), default="active", nullable=False, index=True) + phase: Mapped[str] = mapped_column(String(40), default="copying_destination", nullable=False, index=True) + source_calendar_id: Mapped[str] = mapped_column(ForeignKey("calendar_collections.id", ondelete="RESTRICT"), nullable=False, index=True) + target_calendar_id: Mapped[str] = mapped_column(ForeignKey("calendar_collections.id", ondelete="RESTRICT"), nullable=False, index=True) + source_sync_source_id: Mapped[str] = mapped_column(ForeignKey("calendar_sync_sources.id", ondelete="RESTRICT"), nullable=False, index=True) + target_sync_source_id: Mapped[str] = mapped_column(ForeignKey("calendar_sync_sources.id", ondelete="RESTRICT"), nullable=False, index=True) + total_resources: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + total_events: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + created_by_user_id: Mapped[str | None] = mapped_column(ForeignKey("access_users.id", ondelete="SET NULL"), nullable=True, index=True) + created_by_api_key_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + authorization_evidence: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + cancellation_evidence: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) + last_error: Mapped[str | None] = mapped_column(Text) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + + resources: Mapped[list["CalendarMigrationResource"]] = relationship(back_populates="batch", cascade="all, delete-orphan") + + +class CalendarMigrationResource(Base, TimestampMixin): + __tablename__ = "calendar_migration_resources" + __table_args__ = ( + UniqueConstraint("batch_id", "source_href", name="uq_calendar_migration_resources_batch_href"), + Index("ix_calendar_migration_resources_batch_status", "batch_id", "status"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + batch_id: Mapped[str] = mapped_column(ForeignKey("calendar_migration_batches.id", ondelete="CASCADE"), nullable=False, index=True) + source_href: Mapped[str] = mapped_column(String(1000), nullable=False) + source_expected_etag: Mapped[str] = mapped_column(String(255), nullable=False) + destination_href: Mapped[str] = mapped_column(String(1000), nullable=False) + event_ids: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False) + status: Mapped[str] = mapped_column(String(30), default="copy_pending", nullable=False, index=True) + destination_operation_id: Mapped[str | None] = mapped_column(ForeignKey("calendar_outbox_operations.id", ondelete="SET NULL"), nullable=True, unique=True) + source_delete_operation_id: Mapped[str | None] = mapped_column(ForeignKey("calendar_outbox_operations.id", ondelete="SET NULL"), nullable=True, unique=True) + last_error: Mapped[str | None] = mapped_column(Text) + + batch: Mapped[CalendarMigrationBatch] = relationship(back_populates="resources") + + __all__ = [ "CalendarCollection", "CalendarEvent", + "CalendarMigrationBatch", + "CalendarMigrationResource", "CalendarOutboxOperation", "CalendarSyncCredential", "CalendarSyncSource", diff --git a/src/govoplan_calendar/backend/manifest.py b/src/govoplan_calendar/backend/manifest.py index 55d50b5..401a430 100644 --- a/src/govoplan_calendar/backend/manifest.py +++ b/src/govoplan_calendar/backend/manifest.py @@ -300,6 +300,8 @@ CALENDAR_EXTERNAL_PROVIDERS = ( _calendar_table_retirement_provider = drop_table_retirement_provider( calendar_models.CalendarCollection, calendar_models.CalendarEvent, + calendar_models.CalendarMigrationBatch, + calendar_models.CalendarMigrationResource, calendar_models.CalendarOutboxOperation, calendar_models.CalendarSyncCredential, calendar_models.CalendarSyncSource, @@ -549,7 +551,7 @@ manifest = ModuleManifest( id="calendar.external-sources-and-sync", title="Connect and synchronize external calendars", summary="Calendar supports local collections, two-way CalDAV and Open-Xchange profiles, and read-only ICS/webcal, Microsoft Graph, and Exchange Web Services sources.", - body="Each external source keeps its URL, synchronization direction, status, and credential reference with the Calendar collection. Open-Xchange uses the proven CalDAV transport while retaining connector-profile, identity/group-mapping, and resource-calendar references. Manual or scheduled synchronization records bounded outcomes. CalDAV writes use conditional requests and durable outbox state; conflicts and unknown outcomes require synchronization or explicit reconciliation instead of blind repetition. Removing an external source removes the connection, while deleting a local calendar deletes its owned events after confirmation or transfer.", + body="Each external source keeps its URL, synchronization direction, status, and credential reference with the Calendar collection. Open-Xchange uses the proven CalDAV transport while retaining connector-profile, identity/group-mapping, and resource-calendar references. Manual or scheduled synchronization records bounded outcomes. CalDAV writes use conditional requests and durable outbox state; conflicts and unknown outcomes require synchronization or explicit reconciliation instead of blind repetition. Moving between two-way CalDAV calendars is an administrator-authorized migration batch: all destination resources must be copied before any source resource is conditionally deleted with its recorded ETag. Calendar and event changes remain locked while progress, conflicts, cancellation eligibility, and evidence are visible. Removing an external source removes the connection, while deleting a local calendar deletes its owned events after confirmation or transfer.", documentation_types=("admin", "user"), audience=("user", "calendar_manager", "operator"), related_modules=("connectors", "audit", "ops"), @@ -642,6 +644,8 @@ manifest = ModuleManifest( persistent_table_uninstall_guard( calendar_models.CalendarCollection, calendar_models.CalendarEvent, + calendar_models.CalendarMigrationBatch, + calendar_models.CalendarMigrationResource, calendar_models.CalendarOutboxOperation, calendar_models.CalendarSyncCredential, calendar_models.CalendarSyncSource, diff --git a/src/govoplan_calendar/backend/migrations/versions/d24e5f607182_calendar_remote_move_saga.py b/src/govoplan_calendar/backend/migrations/versions/d24e5f607182_calendar_remote_move_saga.py new file mode 100644 index 0000000..2423a29 --- /dev/null +++ b/src/govoplan_calendar/backend/migrations/versions/d24e5f607182_calendar_remote_move_saga.py @@ -0,0 +1,150 @@ +"""calendar remote move saga + +Revision ID: d24e5f607182 +Revises: c13d4e5f6071 +Create Date: 2026-08-02 12:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "d24e5f607182" +down_revision = "c13d4e5f6071" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "calendar_migration_batches", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("migration_kind", sa.String(length=30), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("phase", sa.String(length=40), nullable=False), + sa.Column("source_calendar_id", sa.String(length=36), nullable=False), + sa.Column("target_calendar_id", sa.String(length=36), nullable=False), + sa.Column("source_sync_source_id", sa.String(length=36), nullable=False), + sa.Column("target_sync_source_id", sa.String(length=36), nullable=False), + sa.Column("total_resources", sa.Integer(), nullable=False), + sa.Column("total_events", sa.Integer(), nullable=False), + sa.Column("created_by_user_id", sa.String(length=36), nullable=True), + sa.Column("created_by_api_key_id", sa.String(length=36), nullable=True), + sa.Column("authorization_evidence", sa.JSON(), nullable=False), + sa.Column("cancellation_evidence", sa.JSON(), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["created_by_user_id"], + ["access_users.id"], + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["source_calendar_id"], + ["calendar_collections.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["source_sync_source_id"], + ["calendar_sync_sources.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["target_calendar_id"], + ["calendar_collections.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["target_sync_source_id"], + ["calendar_sync_sources.id"], + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + ) + for name, columns in ( + ("ix_calendar_migration_batches_tenant_id", ("tenant_id",)), + ("ix_calendar_migration_batches_status", ("status",)), + ("ix_calendar_migration_batches_phase", ("phase",)), + ("ix_calendar_migration_batches_source_calendar_id", ("source_calendar_id",)), + ("ix_calendar_migration_batches_target_calendar_id", ("target_calendar_id",)), + ( + "ix_calendar_migration_batches_source_sync_source_id", + ("source_sync_source_id",), + ), + ( + "ix_calendar_migration_batches_target_sync_source_id", + ("target_sync_source_id",), + ), + ("ix_calendar_migration_batches_created_by_user_id", ("created_by_user_id",)), + ( + "ix_calendar_migration_batches_created_by_api_key_id", + ("created_by_api_key_id",), + ), + ("ix_calendar_migration_batches_completed_at", ("completed_at",)), + ( + "ix_calendar_migration_batches_tenant_status", + ("tenant_id", "status", "created_at"), + ), + ( + "ix_calendar_migration_batches_calendars", + ("tenant_id", "source_calendar_id", "target_calendar_id"), + ), + ): + op.create_index(name, "calendar_migration_batches", columns, unique=False) + + op.create_table( + "calendar_migration_resources", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("batch_id", sa.String(length=36), nullable=False), + sa.Column("source_href", sa.String(length=1000), nullable=False), + sa.Column("source_expected_etag", sa.String(length=255), nullable=False), + sa.Column("destination_href", sa.String(length=1000), nullable=False), + sa.Column("event_ids", sa.JSON(), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("destination_operation_id", sa.String(length=36), nullable=True), + sa.Column("source_delete_operation_id", sa.String(length=36), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["batch_id"], + ["calendar_migration_batches.id"], + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["destination_operation_id"], + ["calendar_outbox_operations.id"], + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["source_delete_operation_id"], + ["calendar_outbox_operations.id"], + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "batch_id", + "source_href", + name="uq_calendar_migration_resources_batch_href", + ), + sa.UniqueConstraint("destination_operation_id"), + sa.UniqueConstraint("source_delete_operation_id"), + ) + for name, columns in ( + ("ix_calendar_migration_resources_tenant_id", ("tenant_id",)), + ("ix_calendar_migration_resources_batch_id", ("batch_id",)), + ("ix_calendar_migration_resources_status", ("status",)), + ("ix_calendar_migration_resources_batch_status", ("batch_id", "status")), + ): + op.create_index(name, "calendar_migration_resources", columns, unique=False) + + +def downgrade() -> None: + op.drop_table("calendar_migration_resources") + op.drop_table("calendar_migration_batches") diff --git a/src/govoplan_calendar/backend/migrations_saga.py b/src/govoplan_calendar/backend/migrations_saga.py new file mode 100644 index 0000000..5147fc4 --- /dev/null +++ b/src/govoplan_calendar/backend/migrations_saga.py @@ -0,0 +1,993 @@ +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", +] diff --git a/src/govoplan_calendar/backend/outbox.py b/src/govoplan_calendar/backend/outbox.py index 30cfdef..e656463 100644 --- a/src/govoplan_calendar/backend/outbox.py +++ b/src/govoplan_calendar/backend/outbox.py @@ -48,17 +48,18 @@ def _operation_source_unavailable_reason( operation: CalendarOutboxOperation, source: CalendarSyncSource | None, ) -> str | None: + metadata = operation.metadata_ if isinstance(operation.metadata_, dict) else {} + is_migration_source_delete = metadata.get("calendar_migration_role") == "source_delete" if source is None: return "CalDAV source no longer exists" if source.tenant_id != operation.tenant_id: return "Calendar outbox source tenant does not match the operation tenant" if source.deleted_at is not None: return "CalDAV source was retired before this operation was delivered" - if not source.sync_enabled: + if not source.sync_enabled and not is_migration_source_delete: return "CalDAV source was disabled before this operation was delivered" if source.sync_direction != "two_way": return "CalDAV source was made inbound-only before this operation was delivered" - metadata = operation.metadata_ or {} frozen_configuration = ( metadata.get("source_configuration") if isinstance(metadata, dict) else None ) @@ -308,17 +309,21 @@ def enqueue_caldav_desired_state( resource_events: list[CalendarEvent], payload_ics: str | None, expected_etag: str | None, + idempotency_context: str | None = None, ) -> CalendarOutboxOperation: now = utcnow() operation_kind = "put" if payload_ics is not None else "delete" fingerprint = calendar_payload_fingerprint(payload_ics) if payload_ics is not None else None + generation = _event_generation(resource_events, trigger_event=trigger_event) + if idempotency_context: + generation = f"{generation}|context:{idempotency_context}" idempotency_key = _operation_idempotency_key( source=source, href=href, operation_kind=operation_kind, payload_fingerprint=fingerprint, expected_etag=expected_etag, - generation=_event_generation(resource_events, trigger_event=trigger_event), + generation=generation, ) existing = ( session.query(CalendarOutboxOperation) @@ -346,6 +351,7 @@ def enqueue_caldav_desired_state( "event_ids": [item.id for item in resource_events], "overwrite": source.conflict_policy == "overwrite", "source_configuration": _source_configuration_snapshot(source), + "idempotency_context": idempotency_context, }, ) session.add(operation) @@ -504,6 +510,12 @@ def _calendar_outbox_action_state( "allowed": discard_unavailable is None, "reason": discard_unavailable, } + metadata = operation.metadata_ if isinstance(operation.metadata_, dict) else {} + if metadata.get("calendar_migration_batch_id"): + actions["discard"] = { + "allowed": False, + "reason": "Migration operations are controlled by their destructive remote-move batch.", + } return actions @@ -641,6 +653,10 @@ def claim_next_calendar_outbox_operation( CalendarOutboxOperation.id.asc(), ).with_for_update(skip_locked=True).limit(50).all() for operation in candidates: + from govoplan_calendar.backend.migrations_saga import migration_operation_dependency_ready + + if not migration_operation_dependency_ready(session, operation): + continue source_delivery_in_progress = ( session.query(CalendarOutboxOperation.id) .filter( @@ -918,6 +934,9 @@ def _complete_operation( remote_etag=remote_etag, ) session.flush() + from govoplan_calendar.backend.migrations_saga import refresh_migration_for_operation + + refresh_migration_for_operation(session, operation) def _retry_delay_seconds(attempt_count: int) -> int: @@ -969,6 +988,9 @@ def _fail_operation( raw_ics=operation.payload_ics, ) session.flush() + from govoplan_calendar.backend.migrations_saga import refresh_migration_for_operation + + refresh_migration_for_operation(session, operation) def _lock_leased_outbox_operation( diff --git a/src/govoplan_calendar/backend/router.py b/src/govoplan_calendar/backend/router.py index c45b456..155a777 100644 --- a/src/govoplan_calendar/backend/router.py +++ b/src/govoplan_calendar/backend/router.py @@ -40,6 +40,9 @@ from govoplan_calendar.backend.schemas import ( CalendarFreeBusyRequest, CalendarFreeBusyResponse, CalendarIcsImportRequest, + CalendarMigrationBatchListResponse, + CalendarMigrationBatchResponse, + CalendarMigrationCancelRequest, CalendarOutboxDispatchResponse, CalendarOutboxOperationListResponse, CalendarOutboxOperationResponse, @@ -54,6 +57,13 @@ from govoplan_calendar.backend.schemas import ( CalendarViewPreferencesResponse, CalendarViewPreferencesUpdateRequest, ) +from govoplan_calendar.backend.migrations_saga import ( + CalendarMigrationError, + calendar_migration_response, + cancel_calendar_migration_batch, + get_calendar_migration_batch, + list_calendar_migration_batches, +) from govoplan_calendar.backend.outbox import ( calendar_outbox_operation_action_states, calendar_outbox_operation_response, @@ -293,6 +303,73 @@ def _outbox_operation_response( ) +def _migration_response(session: Session, batch) -> CalendarMigrationBatchResponse: + return CalendarMigrationBatchResponse.model_validate(calendar_migration_response(session, batch)) + + +@router.get("/migrations", response_model=CalendarMigrationBatchListResponse) +def api_list_calendar_migrations( + calendar_id: str | None = None, + limit: int = Query(default=100, ge=1, le=500), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "calendar:calendar:admin") + migrations = list_calendar_migration_batches( + session, + tenant_id=principal.tenant_id, + calendar_id=calendar_id, + limit=limit, + ) + response = CalendarMigrationBatchListResponse( + migrations=[_migration_response(session, item) for item in migrations] + ) + session.commit() + return response + + +@router.get("/migrations/{batch_id}", response_model=CalendarMigrationBatchResponse) +def api_get_calendar_migration( + batch_id: str, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "calendar:calendar:admin") + try: + batch = get_calendar_migration_batch(session, tenant_id=principal.tenant_id, batch_id=batch_id) + response = _migration_response(session, batch) + session.commit() + return response + except CalendarMigrationError as exc: + session.rollback() + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + + +@router.post("/migrations/{batch_id}/cancel", response_model=CalendarMigrationBatchResponse) +def api_cancel_calendar_migration( + batch_id: str, + payload: CalendarMigrationCancelRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "calendar:calendar:admin") + try: + batch = cancel_calendar_migration_batch( + session, + tenant_id=principal.tenant_id, + batch_id=batch_id, + evidence_note=payload.evidence_note, + user_id=principal.user.id, + api_key_id=principal.api_key_id, + ) + response = _migration_response(session, batch) + session.commit() + return response + except CalendarMigrationError as exc: + session.rollback() + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc + + @router.get("/calendars", response_model=CalendarCollectionListResponse) def api_list_calendars( principal: ApiPrincipal = Depends(get_api_principal), diff --git a/src/govoplan_calendar/backend/schemas.py b/src/govoplan_calendar/backend/schemas.py index d92eaab..46f2456 100644 --- a/src/govoplan_calendar/backend/schemas.py +++ b/src/govoplan_calendar/backend/schemas.py @@ -56,6 +56,8 @@ class CalendarCollectionDeleteRequest(BaseModel): target_calendar_id: str | None = None make_target_default: bool = False external_action: CalendarBulkMoveExternalAction | None = None + destructive_confirmation: str | None = Field(default=None, max_length=100) + evidence_note: str | None = Field(default=None, max_length=2000) class CalendarCollectionResponse(BaseModel): @@ -79,6 +81,55 @@ class CalendarCollectionListResponse(BaseModel): calendars: list[CalendarCollectionResponse] = Field(default_factory=list) +class CalendarMigrationResourceResponse(BaseModel): + id: str + source_href: str + destination_href: str + event_ids: list[str] = Field(default_factory=list) + status: str + destination_operation_id: str | None = None + destination_operation_status: str | None = None + source_delete_operation_id: str | None = None + source_delete_operation_status: str | None = None + last_error: str | None = None + + +class CalendarMigrationBatchResponse(BaseModel): + id: str + migration_kind: str + status: str + phase: str + source_calendar_id: str + target_calendar_id: str + source_sync_source_id: str + target_sync_source_id: str + total_resources: int + copied_resources: int + deleted_source_resources: int + conflict_count: int + total_events: int + last_error: str | None = None + can_cancel: bool = False + authorization_evidence: dict[str, Any] = Field(default_factory=dict) + cancellation_evidence: dict[str, Any] | None = None + created_by_user_id: str | None = None + created_by_api_key_id: str | None = None + created_at: datetime + updated_at: datetime + completed_at: datetime | None = None + resources: list[CalendarMigrationResourceResponse] = Field(default_factory=list) + + +class CalendarMigrationBatchListResponse(BaseModel): + migrations: list[CalendarMigrationBatchResponse] = Field(default_factory=list) + + +class CalendarMigrationCancelRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + evidence_note: str = Field(min_length=10, max_length=2000) + + class CalendarSyncSourceCreateRequest(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/src/govoplan_calendar/backend/service.py b/src/govoplan_calendar/backend/service.py index 79064cc..3d7ce71 100644 --- a/src/govoplan_calendar/backend/service.py +++ b/src/govoplan_calendar/backend/service.py @@ -37,10 +37,20 @@ from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNo from govoplan_calendar.backend.db.models import ( CalendarCollection, CalendarEvent, + CalendarMigrationBatch, CalendarSyncCredential, CalendarSyncSource, CalendarViewPreference, ) +from govoplan_calendar.backend.migrations_saga import ( + CalendarMigrationError, + active_tenant_migration, + assert_calendar_not_migrating, + assert_event_not_migrating, + assert_source_not_migrating, + migration_source_ids_in_progress, + start_remote_move, +) from govoplan_calendar.backend.ews import ( EwsAdapterError, ews_find_item_body, @@ -103,6 +113,42 @@ DEFAULT_CALENDAR_VIEW_PREFERENCES: dict[str, bool | int] = { } +def _assert_calendar_mutation_allowed(session: Session, *, tenant_id: str, calendar_id: str) -> None: + if not hasattr(session, "query"): + return + try: + assert_calendar_not_migrating(session, tenant_id=tenant_id, calendar_id=calendar_id) + except CalendarMigrationError as exc: + raise CalendarError(str(exc)) from exc + + +def _assert_source_mutation_allowed(session: Session, *, tenant_id: str, source_id: str) -> None: + if not hasattr(session, "query"): + return + try: + assert_source_not_migrating(session, tenant_id=tenant_id, source_id=source_id) + except CalendarMigrationError as exc: + raise CalendarError(str(exc)) from exc + + +def _assert_event_mutation_allowed(event: CalendarEvent) -> None: + try: + assert_event_not_migrating(event) + except CalendarMigrationError as exc: + raise CalendarError(str(exc)) from exc + + +def _assert_default_calendar_mutation_allowed(session: Session, *, tenant_id: str) -> None: + if not hasattr(session, "query"): + return + batch = active_tenant_migration(session, tenant_id=tenant_id) + if batch is not None: + raise CalendarError( + "The default calendar cannot change while remote move " + f"{batch.id} is {batch.phase}." + ) + + def calendar_credential_context( *, tenant_id: str, @@ -585,6 +631,7 @@ def create_calendar(session: Session, *, tenant_id: str, user_id: str | None, pa if calendar_slug_exists(session, tenant_id=tenant_id, slug=slug): raise CalendarError(f"Calendar slug already exists: {slug}") if payload.is_default: + _assert_default_calendar_mutation_allowed(session, tenant_id=tenant_id) clear_default_calendar(session, tenant_id=tenant_id) calendar = CalendarCollection( tenant_id=tenant_id, @@ -607,7 +654,9 @@ def create_calendar(session: Session, *, tenant_id: str, user_id: str | None, pa def update_calendar(session: Session, *, tenant_id: str, calendar_id: str, payload: CalendarCollectionUpdateRequest) -> CalendarCollection: calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id) + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=calendar.id) if payload.is_default is True: + _assert_default_calendar_mutation_allowed(session, tenant_id=tenant_id) clear_default_calendar(session, tenant_id=tenant_id) calendar.is_default = True elif payload.is_default is False: @@ -680,15 +729,14 @@ def _validate_calendar_move_source_pair( source: CalendarSyncSource | None, target_source: CalendarSyncSource | None, ) -> None: - if payload.external_action == "remote_move": - raise CalendarError( - "external_action='remote_move' is not supported; no destructive remote move was performed" - ) if source is not None and target_source is not None: + if payload.external_action == "remote_move": + return raise CalendarError( - "Events cannot be bulk-moved between synchronized calendars; " - "external_action='remote_move' remains unsupported" + "Moving events between synchronized calendars requires external_action='remote_move'" ) + if payload.external_action == "remote_move": + raise CalendarError("external_action='remote_move' requires synchronized source and target calendars") def _prepare_calendar_move_external_action( @@ -700,6 +748,8 @@ def _prepare_calendar_move_external_action( target_source: CalendarSyncSource | None, deleted_at: datetime, ) -> None: + if source is not None and target_source is not None: + return if source is not None: if payload.external_action != "detach_keep_remote": raise CalendarError( @@ -801,7 +851,9 @@ def _move_calendar_before_delete( calendar: CalendarCollection, payload: CalendarCollectionDeleteRequest, deleted_at: datetime, -) -> CalendarCollection: + user_id: str | None, + api_key_id: str | None, +) -> tuple[CalendarCollection, CalendarMigrationBatch | None]: if not payload.target_calendar_id: raise CalendarError("Target calendar is required when moving events") if payload.target_calendar_id == calendar.id: @@ -817,6 +869,8 @@ def _move_calendar_before_delete( calendar=calendar, target_calendar=target_calendar, ) + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=calendar.id) + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=target_calendar.id) _validate_calendar_move_source_pair( payload, source=source, @@ -826,6 +880,26 @@ def _move_calendar_before_delete( session, calendar, ) + if source is not None and target_source is not None: + try: + batch = start_remote_move( + session, + tenant_id=tenant_id, + source_calendar=calendar, + target_calendar=target_calendar, + source=source, + target_source=target_source, + events=active_events, + previous_event_states=previous_event_states, + make_target_default=payload.make_target_default, + confirmation=payload.destructive_confirmation, + evidence_note=payload.evidence_note, + user_id=user_id, + api_key_id=api_key_id, + ) + except CalendarMigrationError as exc: + raise CalendarError(str(exc)) from exc + return calendar, batch _prepare_calendar_move_external_action( session, tenant_id=tenant_id, @@ -844,7 +918,7 @@ def _move_calendar_before_delete( active_events=active_events, previous_event_states=previous_event_states, ) - return calendar + return calendar, None def delete_calendar( @@ -855,22 +929,29 @@ def delete_calendar( payload: CalendarCollectionDeleteRequest | None = None, user_id: str | None = None, api_key_id: str | None = None, -) -> None: +) -> CalendarMigrationBatch | None: payload = payload or CalendarCollectionDeleteRequest() calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id) + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=calendar.id) deleted_at = utcnow() + migration_batch: CalendarMigrationBatch | None = None if payload.event_action == "move": - calendar = _move_calendar_before_delete( + calendar, migration_batch = _move_calendar_before_delete( session, tenant_id=tenant_id, calendar=calendar, payload=payload, deleted_at=deleted_at, + user_id=user_id, + api_key_id=api_key_id, ) elif payload.event_action != "delete": raise CalendarError(f"Unsupported calendar delete event action: {payload.event_action}") elif payload.external_action is not None: raise CalendarError("external_action is only valid when event_action='move'") + if migration_batch is not None: + session.flush() + return migration_batch if calendar.is_default: calendar.is_default = False calendar.deleted_at = deleted_at @@ -887,6 +968,7 @@ def delete_calendar( if payload.event_action == "delete" and event.deleted_at is None: event.deleted_at = deleted_at session.flush() + return None def get_calendar(session: Session, *, tenant_id: str, calendar_id: str) -> CalendarCollection: @@ -1641,6 +1723,7 @@ def update_sync_source( api_key_id: str | None = None, ) -> CalendarSyncSource: source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id) + _assert_source_mutation_allowed(session, tenant_id=tenant_id, source_id=source.id) source = _lock_sync_source_for_update( session, tenant_id=tenant_id, @@ -1736,6 +1819,7 @@ def delete_sync_source( api_key_id: str | None = None, ) -> None: source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id) + _assert_source_mutation_allowed(session, tenant_id=tenant_id, source_id=source.id) retire_sync_source( session, tenant_id=tenant_id, @@ -1757,6 +1841,7 @@ def delete_caldav_source( api_key_id: str | None = None, ) -> None: source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id, source_kind="caldav") + _assert_source_mutation_allowed(session, tenant_id=tenant_id, source_id=source.id) retire_sync_source( session, tenant_id=tenant_id, @@ -2317,6 +2402,8 @@ def sync_source( bearer_token: str | None = None, force_full: bool = False, ) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]: + source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id) + _assert_source_mutation_allowed(session, tenant_id=tenant_id, source_id=source.id) # Sync is an explicit unit-of-work boundary. Commit configuration changes # and release any request/authentication transaction before remote I/O. session.commit() @@ -2569,6 +2656,7 @@ def _prepare_caldav_sync( tenant_id=tenant_id, source_id=source_id, ) + _assert_source_mutation_allowed(preparation_session, tenant_id=tenant_id, source_id=source.id) get_calendar( preparation_session, tenant_id=tenant_id, @@ -2710,6 +2798,7 @@ def _prepare_remote_sync( tenant_id=tenant_id, source_id=source_id, ) + _assert_source_mutation_allowed(preparation_session, tenant_id=tenant_id, source_id=source.id) if source.source_kind not in expected_kinds: expected = "/".join(sorted(expected_kinds)) raise CalendarError(f"{expected} sync source not found") @@ -3354,6 +3443,9 @@ def _sync_due_sources( ) if tenant_id is not None: query = query.filter(CalendarSyncSource.tenant_id == tenant_id) + migrating_source_ids = migration_source_ids_in_progress(session, tenant_id=tenant_id) + if migrating_source_ids: + query = query.filter(CalendarSyncSource.id.notin_(sorted(migrating_source_ids))) sources = ( query.order_by( CalendarSyncSource.next_sync_at.asc(), @@ -4095,7 +4187,8 @@ def create_event(session: Session, *, tenant_id: str, user_id: str | None, paylo calendar_id = payload.calendar_id or (default_calendar.id if default_calendar else None) if not calendar_id: raise CalendarError("calendar_id is required because no default calendar exists") - get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id) + calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id) + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=calendar.id) source = active_sync_source_for_calendar( session, tenant_id=tenant_id, @@ -4180,6 +4273,9 @@ def _event_update_calendar_ids( raise CalendarError("Calendar event not found") original_calendar_id = event_locator[0] target_calendar_id = payload.calendar_id or original_calendar_id + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=original_calendar_id) + if target_calendar_id != original_calendar_id: + _assert_calendar_mutation_allowed(session, tenant_id=tenant_id, calendar_id=target_calendar_id) if payload.calendar_id is not None: get_calendar(session, tenant_id=tenant_id, calendar_id=target_calendar_id) return original_calendar_id, target_calendar_id @@ -4317,6 +4413,7 @@ def update_event(session: Session, *, tenant_id: str, user_id: str | None, event target_calendar_id=target_calendar_id, ) event = get_event(session, tenant_id=tenant_id, event_id=event_id) + _assert_event_mutation_allowed(event) previous = calendar_event_change_payload(event, prefix="previous_") _assert_event_update_sync_fields( payload, @@ -4355,6 +4452,7 @@ def update_event_occurrence( tenant_id=tenant_id, event_id=series_event_id, ) + _assert_event_mutation_allowed(master) if master.recurrence_id is not None or not (master.rrule or master.rdate): raise CalendarError("Calendar event is not a recurring series master") occurrence = recurrence_occurrence(master, payload.recurrence_id) @@ -4552,6 +4650,7 @@ def delete_event(session: Session, *, tenant_id: str, event_id: str, user_id: st for_update=True, ) event = get_event(session, tenant_id=tenant_id, event_id=event_id) + _assert_event_mutation_allowed(event) assert_sync_mutation_allowed(source) series_events = [event] if event.recurrence_id is None and (event.rrule or event.rdate): diff --git a/tests/test_outbox.py b/tests/test_outbox.py index 912c355..fa0e2c1 100644 --- a/tests/test_outbox.py +++ b/tests/test_outbox.py @@ -996,7 +996,7 @@ class CalendarOutboxTests(unittest.TestCase): ) self.session.rollback() - with self.assertRaisesRegex(CalendarError, "remote_move.*not supported"): + with self.assertRaisesRegex(CalendarError, "requires synchronized"): delete_calendar( self.session, tenant_id="tenant-1", diff --git a/tests/test_remote_move.py b/tests/test_remote_move.py new file mode 100644 index 0000000..5dc6dbd --- /dev/null +++ b/tests/test_remote_move.py @@ -0,0 +1,380 @@ +from __future__ import annotations + +import unittest +from datetime import datetime, timedelta, timezone + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_access.backend.db import models as access_models # noqa: F401 +from govoplan_calendar.backend.db.models import CalendarMigrationBatch +from govoplan_calendar.backend.migrations_saga import ( + cancel_calendar_migration_batch, + get_calendar_migration_batch, +) +from govoplan_calendar.backend.outbox import dispatch_calendar_outbox +from govoplan_calendar.backend.schemas import ( + CalendarCalDavSourceCreateRequest, + CalendarCollectionCreateRequest, + CalendarCollectionDeleteRequest, + CalendarEventCreateRequest, + CalendarEventUpdateRequest, +) +from govoplan_calendar.backend.service import ( + CalendarError, + create_caldav_source, + create_calendar, + create_event, + delete_calendar, + update_event, +) +from govoplan_core.db.base import Base, utcnow +from govoplan_core.tenancy.scope import create_scope_tables +from govoplan_tenancy.backend.db.models import Tenant +from tests.test_outbox import StatefulCalDAVClient + + +class CalendarRemoteMoveTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:", future=True) + create_scope_tables(self.engine) + Base.metadata.create_all(self.engine) + self.Session = sessionmaker(bind=self.engine, expire_on_commit=False) + self.session = self.Session() + self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) + self.calendar = create_calendar( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Remote source"), + ) + self.source = create_caldav_source( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=self.calendar.id, + collection_url="https://dav.example.test/source", + ), + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def create_event(self, uid: str): + return create_event( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarEventCreateRequest( + calendar_id=self.calendar.id, + uid=uid, + summary="Planning", + start_at=datetime(2026, 7, 8, 9, 0, tzinfo=timezone.utc), + end_at=datetime(2026, 7, 8, 10, 0, tzinfo=timezone.utc), + ), + ) + + def create_target(self): + calendar = create_calendar( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Remote target"), + ) + source = create_caldav_source( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=calendar.id, + collection_url="https://dav.example.test/target", + ), + ) + self.session.commit() + return calendar, source + + def deliver_source_events(self, *events): + self.session.commit() + client = StatefulCalDAVClient() + result = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=lambda _session, _source: client, + ) + self.assertEqual(result["succeeded"], len(events)) + for event in events: + self.assertEqual(event.source_kind, "caldav") + self.assertIsNotNone(event.source_href) + self.assertIsNotNone(event.etag) + return client + + def start_move( + self, + target_calendar, + *, + make_target_default: bool = False, + ) -> CalendarMigrationBatch: + batch = delete_calendar( + self.session, + tenant_id="tenant-1", + calendar_id=self.calendar.id, + payload=CalendarCollectionDeleteRequest( + event_action="move", + target_calendar_id=target_calendar.id, + external_action="remote_move", + destructive_confirmation="MOVE REMOTE EVENTS", + evidence_note="Approved migration window 42", + make_target_default=make_target_default, + ), + ) + self.assertIsInstance(batch, CalendarMigrationBatch) + self.session.commit() + return batch + + def clients(self, source_client, target_client): + return lambda _session, source: ( + source_client if source.id == self.source.id else target_client + ) + + def test_all_resources_are_copied_before_conditional_source_delete(self) -> None: + first = self.create_event("move-first@example.test") + second = self.create_event("move-second@example.test") + source_client = self.deliver_source_events(first, second) + target_calendar, target_source = self.create_target() + target_client = StatefulCalDAVClient() + self.source.conflict_policy = "overwrite" + target_source.conflict_policy = "overwrite" + self.calendar.is_default = True + self.session.commit() + batch = self.start_move(target_calendar, make_target_default=True) + + result = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=self.clients(source_client, target_client), + ) + + self.assertEqual(result["failed"], 0) + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.status, "completed") + self.assertEqual(len(target_client.resources), 2) + self.assertEqual(source_client.resources, {}) + self.assertEqual(len(source_client.deletes), 2) + self.assertTrue(all(item["etag"] for item in source_client.deletes)) + self.assertTrue(all(item["overwrite"] is False for item in source_client.deletes)) + self.assertTrue(all(item["overwrite"] is False for item in target_client.puts)) + self.assertIsNotNone(self.calendar.deleted_at) + self.assertIsNotNone(self.source.deleted_at) + self.assertEqual(first.calendar_id, target_calendar.id) + self.assertNotIn("calendar_migration", first.metadata_ or {}) + self.assertIsNone(target_source.deleted_at) + self.assertTrue(target_calendar.is_default) + + def test_partial_copy_failure_never_deletes_source(self) -> None: + first = self.create_event("partial-first@example.test") + second = self.create_event("partial-second@example.test") + source_client = self.deliver_source_events(first, second) + target_calendar, _target_source = self.create_target() + target_client = StatefulCalDAVClient() + target_client.fail_before_write = True + batch = self.start_move(target_calendar) + + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=self.clients(source_client, target_client), + ) + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.phase, "copying_destination") + self.assertEqual(source_client.deletes, []) + self.assertEqual(len(source_client.resources), 2) + self.assertIsNone(self.calendar.deleted_at) + + def test_uid_collision_is_rejected_before_mutation(self) -> None: + source_event = self.create_event("collision@example.test") + self.deliver_source_events(source_event) + target_calendar, target_source = self.create_target() + target_event = create_event( + self.session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarEventCreateRequest( + calendar_id=target_calendar.id, + uid=source_event.uid, + summary="Existing target", + start_at=datetime(2026, 7, 9, 9, 0, tzinfo=timezone.utc), + ), + ) + self.session.commit() + target_client = StatefulCalDAVClient() + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=lambda _session, _source: target_client, + ) + + with self.assertRaisesRegex(CalendarError, "already contains UID"): + self.start_move(target_calendar) + self.session.rollback() + self.assertEqual(source_event.calendar_id, self.calendar.id) + self.assertEqual(target_event.calendar_id, target_calendar.id) + self.assertIsNone(self.calendar.deleted_at) + self.assertIsNotNone(target_source.id) + + def test_cancel_before_source_delete_retains_both_remote_copies(self) -> None: + event = self.create_event("cancel@example.test") + source_client = self.deliver_source_events(event) + target_calendar, _target_source = self.create_target() + target_client = StatefulCalDAVClient() + self.calendar.is_default = True + self.session.commit() + batch = self.start_move(target_calendar, make_target_default=True) + + batch = cancel_calendar_migration_batch( + self.session, + tenant_id="tenant-1", + batch_id=batch.id, + evidence_note="Operator cancelled approved migration", + user_id=None, + api_key_id=None, + ) + self.session.commit() + self.assertEqual(batch.status, "cancel_requested") + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=self.clients(source_client, target_client), + ) + + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.status, "cancelled") + self.assertEqual(source_client.deletes, []) + self.assertIsNone(self.calendar.deleted_at) + self.assertTrue(self.source.sync_enabled) + self.assertEqual(len(target_client.resources), 1) + self.assertTrue(self.calendar.is_default) + self.assertFalse(target_calendar.is_default) + + def test_concurrent_event_edit_is_blocked(self) -> None: + event = self.create_event("locked@example.test") + self.deliver_source_events(event) + target_calendar, _target_source = self.create_target() + self.start_move(target_calendar) + + with self.assertRaisesRegex(CalendarError, "blocked while"): + update_event( + self.session, + tenant_id="tenant-1", + user_id=None, + event_id=event.id, + payload=CalendarEventUpdateRequest(summary="Unsafe edit"), + ) + self.session.rollback() + + def test_source_etag_change_becomes_reconciliation_conflict(self) -> None: + event = self.create_event("etag-conflict@example.test") + source_client = self.deliver_source_events(event) + target_calendar, _target_source = self.create_target() + target_client = StatefulCalDAVClient() + batch = self.start_move(target_calendar) + + first = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + limit=1, + client_factory=self.clients(source_client, target_client), + ) + self.assertEqual(first["succeeded"], 1) + href = next(iter(source_client.resources)) + ics, _etag = source_client.resources[href] + source_client.resources[href] = (ics, '"concurrent-edit"') + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=self.clients(source_client, target_client), + ) + + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.status, "blocked") + self.assertEqual(batch.phase, "source_delete_conflict") + self.assertIn(href, source_client.resources) + self.assertIsNone(self.calendar.deleted_at) + + def test_crash_after_destination_write_is_reconciled_without_duplicate_put(self) -> None: + event = self.create_event("crash-recovery@example.test") + source_client = self.deliver_source_events(event) + target_calendar, _target_source = self.create_target() + target_client = StatefulCalDAVClient() + target_client.fail_after_write = True + batch = self.start_move(target_calendar) + + first = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + limit=1, + client_factory=self.clients(source_client, target_client), + ) + self.assertEqual(first["succeeded"], 1) + target_client.fail_after_write = False + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + now=utcnow() + timedelta(seconds=10), + client_factory=self.clients(source_client, target_client), + ) + + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.status, "completed") + self.assertEqual(len(target_client.puts), 1) + + def test_crash_after_source_delete_is_reconciled_without_duplicate_delete(self) -> None: + event = self.create_event("delete-crash@example.test") + source_client = self.deliver_source_events(event) + target_calendar, _target_source = self.create_target() + target_client = StatefulCalDAVClient() + batch = self.start_move(target_calendar) + + copied = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + limit=1, + client_factory=self.clients(source_client, target_client), + ) + self.assertEqual(copied["succeeded"], 1) + source_client.fail_after_write = True + deleted = dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + limit=1, + client_factory=self.clients(source_client, target_client), + ) + self.assertEqual(deleted["succeeded"], 1) + source_client.fail_after_write = False + dispatch_calendar_outbox( + self.session, + tenant_id="tenant-1", + client_factory=self.clients(source_client, target_client), + ) + + batch = get_calendar_migration_batch( + self.session, tenant_id="tenant-1", batch_id=batch.id + ) + self.assertEqual(batch.status, "completed") + self.assertEqual(len(source_client.deletes), 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/calendar.ts b/webui/src/api/calendar.ts index f7898ad..67bae1a 100644 --- a/webui/src/api/calendar.ts +++ b/webui/src/api/calendar.ts @@ -220,6 +220,49 @@ export type CalendarOutboxOperationListResponse = { operations: CalendarOutboxOperation[]; }; +export type CalendarMigrationResource = { + id: string; + source_href: string; + destination_href: string; + event_ids: string[]; + status: string; + destination_operation_id?: string | null; + destination_operation_status?: string | null; + source_delete_operation_id?: string | null; + source_delete_operation_status?: string | null; + last_error?: string | null; +}; + +export type CalendarMigrationBatch = { + id: string; + migration_kind: string; + status: string; + phase: string; + source_calendar_id: string; + target_calendar_id: string; + source_sync_source_id: string; + target_sync_source_id: string; + total_resources: number; + copied_resources: number; + deleted_source_resources: number; + conflict_count: number; + total_events: number; + last_error?: string | null; + can_cancel: boolean; + authorization_evidence: Record; + cancellation_evidence?: Record | null; + created_by_user_id?: string | null; + created_by_api_key_id?: string | null; + created_at: string; + updated_at: string; + completed_at?: string | null; + resources: CalendarMigrationResource[]; +}; + +export type CalendarMigrationBatchListResponse = { + migrations: CalendarMigrationBatch[]; +}; + export type CalendarCollectionCreatePayload = { name: string; slug?: string | null; @@ -241,6 +284,8 @@ export type CalendarCollectionDeletePayload = { target_calendar_id?: string | null; make_target_default?: boolean; external_action?: CalendarBulkMoveExternalAction | null; + destructive_confirmation?: string | null; + evidence_note?: string | null; }; export type CalendarEventCreatePayload = { @@ -396,6 +441,35 @@ export function recoverCalendarOutboxOperation( }); } +export function listCalendarMigrations( + settings: ApiSettings, + params: { calendar_id?: string; limit?: number } = {}, +): Promise { + const search = new URLSearchParams(); + if (params.calendar_id) search.set("calendar_id", params.calendar_id); + if (params.limit) search.set("limit", String(params.limit)); + const suffix = search.toString() ? `?${search.toString()}` : ""; + return apiFetch(settings, `/api/v1/calendar/migrations${suffix}`); +} + +export function getCalendarMigration( + settings: ApiSettings, + batchId: string, +): Promise { + return apiFetch(settings, `/api/v1/calendar/migrations/${batchId}`); +} + +export function cancelCalendarMigration( + settings: ApiSettings, + batchId: string, + evidenceNote: string, +): Promise { + return apiFetch(settings, `/api/v1/calendar/migrations/${batchId}/cancel`, { + method: "POST", + body: JSON.stringify({ evidence_note: evidenceNote }), + }); +} + export function listCalendarEvents( settings: ApiSettings, params: { calendar_id?: string; start_at?: string; end_at?: string; expand_recurring?: boolean } = {} diff --git a/webui/src/features/calendar/CalendarCollectionDialogs.tsx b/webui/src/features/calendar/CalendarCollectionDialogs.tsx index bcc1d0c..99be022 100644 --- a/webui/src/features/calendar/CalendarCollectionDialogs.tsx +++ b/webui/src/features/calendar/CalendarCollectionDialogs.tsx @@ -5,7 +5,7 @@ import { type ChangeEvent, type FormEvent, } from "react"; -import { ListChecks, RefreshCw, Trash2 } from "lucide-react"; +import { ArrowRightLeft, ListChecks, RefreshCw, Trash2 } from "lucide-react"; import { Button, ColorPickerField, @@ -102,6 +102,7 @@ export function CalendarCollectionDialog({ onRequestDelete, onSync, onOpenOutbox, + onOpenMigration, onDiscover @@ -117,7 +118,7 @@ export function CalendarCollectionDialog({ -}: {state: CalendarCollectionDialogState;settings: ApiSettings;source: CalendarSyncSource | null;saving: boolean;syncingSourceId: string;canWrite: boolean;canDelete: boolean;canManageSources: boolean;canSyncSources: boolean;onCancel: () => void;onSave: (payload: CalendarCollectionFormPayload) => Promise;onRequestDelete: (calendar: CalendarCollection, eventCount: number | null, loadingEventCount: boolean) => void;onSync: (source: CalendarSyncSource, payload?: {password?: string | null;bearer_token?: string | null;force_full?: boolean;}) => Promise;onOpenOutbox: (source: CalendarSyncSource) => void;onDiscover: (payload: CalendarCalDavDiscoveryPayload) => Promise<{calendars: CalendarCalDavDiscoveryCandidate[];}>;}) { +}: {state: CalendarCollectionDialogState;settings: ApiSettings;source: CalendarSyncSource | null;saving: boolean;syncingSourceId: string;canWrite: boolean;canDelete: boolean;canManageSources: boolean;canSyncSources: boolean;onCancel: () => void;onSave: (payload: CalendarCollectionFormPayload) => Promise;onRequestDelete: (calendar: CalendarCollection, eventCount: number | null, loadingEventCount: boolean) => void;onSync: (source: CalendarSyncSource, payload?: {password?: string | null;bearer_token?: string | null;force_full?: boolean;}) => Promise;onOpenOutbox: (source: CalendarSyncSource) => void;onOpenMigration: (batchId: string) => void;onDiscover: (payload: CalendarCalDavDiscoveryPayload) => Promise<{calendars: CalendarCalDavDiscoveryCandidate[];}>;}) { const calendar = state.kind === "edit" ? state.calendar : null; const isEdit = Boolean(calendar); const [sourceMode, setSourceMode] = useState(source ? calendarSourceModeForSource(source) : "local"); @@ -146,6 +147,8 @@ export function CalendarCollectionDialog({ const [discoveryError, setDiscoveryError] = useState(""); const formId = "calendar-collection-form"; const isExistingSyncSource = isEdit && Boolean(source); + const migrationBatchId = calendar ? calendarMigrationBatchId(calendar) : ""; + const migrationLocked = Boolean(migrationBatchId); const canEditSource = canManageSources; const canEditMutableSourceSettings = canEditSource && !isExistingSyncSource; const effectiveCollectionUrl = (collectionUrl || davUrl).trim(); @@ -160,6 +163,7 @@ export function CalendarCollectionDialog({ const saveDisabled = saving || + migrationLocked || !canWrite || !name.trim() || sourceDetailsInvalid; @@ -344,7 +348,7 @@ export function CalendarCollectionDialog({ <>
{calendar && canDelete && - } @@ -357,6 +361,11 @@ export function CalendarCollectionDialog({ }>
+ {migrationLocked && +

+ Calendar and event changes are locked while the destructive remote move is being reconciled. +

+ } {sourceMode !== "local" && !canManageSources &&

i18n:govoplan-calendar.managing_sync_sources_requires_calendar_administ.835e29fa

} {!isEdit && void onSync(source, syncTransientPayload(effectiveAuthType, password, bearerToken))} - disabled={saving || syncing || !canSyncSources}> + disabled={saving || syncing || migrationLocked || !canSyncSources}> {syncing ? "i18n:govoplan-calendar.syncing.e5c7727a" : "i18n:govoplan-calendar.sync_now.2b7d938e"} - {source.source_kind === "caldav" && canManageSources && ( @@ -565,6 +574,11 @@ export function CalendarCollectionDialog({ i18n:govoplan-calendar.outbound_changes.7038a839 )} + {migrationBatchId && canManageSources && ( + + )}
} @@ -602,18 +616,28 @@ export function CalendarCollectionDeleteDialog({ const [eventAction, setEventAction] = useState("delete"); const [targetCalendarId, setTargetCalendarId] = useState(firstMoveTargetId); const [makeTargetDefault, setMakeTargetDefault] = useState(calendar.is_default && Boolean(firstMoveTargetId)); + const [remoteMoveConfirmation, setRemoteMoveConfirmation] = useState(""); + const [remoteMoveEvidence, setRemoteMoveEvidence] = useState(""); const effectiveEventAction: CalendarDeleteEventAction = canMoveEvents ? eventAction : "delete"; - const confirmDisabled = saving || loadingEventCount || canMoveEvents && effectiveEventAction === "move" && !targetCalendarId; const actionLabel = calendarDeleteActionLabel(calendar); const targetSource = syncSourceByCalendarId.get(targetCalendarId) ?? null; const externalAction = calendarBulkMoveExternalAction(source, targetSource); + const isRemoteMove = effectiveEventAction === "move" && externalAction === "remote_move"; + const confirmDisabled = saving || loadingEventCount || + (canMoveEvents && effectiveEventAction === "move" && !targetCalendarId) || + (isRemoteMove && ( + remoteMoveConfirmation !== "MOVE REMOTE EVENTS" || + remoteMoveEvidence.trim().length < 10 + )); function confirm() { void onDelete(calendar, { event_action: effectiveEventAction, target_calendar_id: effectiveEventAction === "move" ? targetCalendarId : null, make_target_default: effectiveEventAction === "move" && calendar.is_default && makeTargetDefault, - external_action: effectiveEventAction === "move" ? externalAction : null + external_action: effectiveEventAction === "move" ? externalAction : null, + destructive_confirmation: isRemoteMove ? remoteMoveConfirmation : null, + evidence_note: isRemoteMove ? remoteMoveEvidence.trim() : null }); } @@ -679,6 +703,32 @@ export function CalendarCollectionDeleteDialog({ }

{calendarBulkMoveConsequence(externalAction)}

+ {isRemoteMove && +
+

+ This operation copies every CalDAV resource before conditionally deleting the source resources. Calendar and event edits remain locked until the batch completes or is reconciled. +

+ +