govoplan-calendar/src/govoplan_calendar/backend/service.py

from __future__ import annotations

import copy
import json
import posixpath
import re
import uuid
import os
import urllib.parse
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, Iterable

from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from govoplan_core.security.outbound_http import (
    OutboundHttpError,
    bounded_response_bytes,
    build_outbound_http_opener,
    validate_outbound_http_url,
)
from govoplan_core.security.credential_envelopes import (
    CredentialAccessContext,
    CredentialEnvelopeError,
    ResolvedCredentialEnvelope,
    credential_envelope_summary,
    get_credential_envelope,
    list_credential_envelopes,
    resolve_credential_envelope,
)
from govoplan_core.audit.logging import audit_event

from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
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,
    parse_ews_calendar_items,
)
from govoplan_calendar.backend.graph import graph_event_payload
from govoplan_calendar.backend.ical import (
    expand_event_occurrences,
    normalized_recurrence_id,
    parse_vevent,
    parse_vevents,
    recurrence_id_datetime,
)
from govoplan_calendar.backend.runtime import get_registry
from govoplan_calendar.backend.schemas import (
    CalendarCalDavDiscoveryRequest,
    CalendarCalDavSourceCreateRequest,
    CalendarCalDavSourceUpdateRequest,
    CalendarCollectionCreateRequest,
    CalendarCollectionDeleteRequest,
    CalendarCollectionUpdateRequest,
    CalendarEventCreateRequest,
    CalendarEventOccurrenceUpdateRequest,
    CalendarEventUpdateRequest,
    CalendarSyncSourceCreateRequest,
    CalendarSyncSourceUpdateRequest,
    CalendarViewPreferencesUpdateRequest,
)
from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider
from govoplan_core.db.base import utcnow
from govoplan_core.core.change_sequence import record_change
from govoplan_core.security.secrets import CAPABILITY_SECURITY_SECRET_PROVIDER, decrypt_secret, encrypt_secret


class CalendarError(ValueError):
    pass


CALDAV_INTERNAL_CREDENTIAL_PREFIX = "calendar-sync-credential:"
CORE_CREDENTIAL_ENVELOPE_PREFIX = "credential-envelope:"
CALDAV_ENV_CREDENTIAL_PREFIX = "env:"
CALDAV_DEFAULT_SYNC_INTERVAL_SECONDS = 900
SYNC_SOURCE_KINDS = {"caldav", "ics", "webcal", "graph", "ews"}
READ_ONLY_SYNC_SOURCE_KINDS = {"ics", "webcal", "graph", "ews"}
GRAPH_DEFAULT_BASE_URL = "https://graph.microsoft.com/v1.0/"
CALENDAR_MODULE_ID = "calendar"
CALENDAR_EVENTS_COLLECTION = "calendar.events"
CALENDAR_EVENT_RESOURCE = "calendar_event"
SOURCE_EVENT_CLEANUP_BATCH_SIZE = 500
REMOTE_SYNC_MAX_ITEMS = 10_000
MAX_OCCURRENCE_RANGE_DAYS = 400
DEFAULT_CALENDAR_VIEW_PREFERENCES: dict[str, bool | int] = {
    "dim_weekends": True,
    "dim_off_hours": True,
    "workday_start_hour": 6,
    "workday_end_hour": 20,
    "continuous_virtualization": True,
    "continuous_overscan_weeks": 6,
    "alternate_continuous_months": True,
}


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,
    source_id: str | None = None,
) -> CredentialAccessContext:
    return CredentialAccessContext(
        tenant_id=tenant_id,
        target_scope_type="tenant",
        target_scope_id=tenant_id,
        module_id=CALENDAR_MODULE_ID,
        server_ref=f"calendar:{source_id}" if source_id else None,
    )


def available_calendar_credentials(
    session: Session,
    *,
    tenant_id: str,
    source_id: str | None = None,
) -> list[dict[str, Any]]:
    context = calendar_credential_context(tenant_id=tenant_id, source_id=source_id)
    return [
        credential_envelope_summary(row)
        for row in list_credential_envelopes(session, context=context)
    ]


def calendar_event_change_payload(event: CalendarEvent, *, prefix: str = "") -> dict[str, Any]:
    metadata = event.metadata_ or {}
    caldav = metadata.get("caldav") if isinstance(metadata, dict) else None
    return {
        f"{prefix}calendar_id": event.calendar_id,
        f"{prefix}start_at": response_datetime(event.start_at).isoformat() if event.start_at else None,
        f"{prefix}end_at": response_datetime(event.end_at).isoformat() if event.end_at else None,
        f"{prefix}summary": event.summary,
        f"{prefix}status": event.status,
        f"{prefix}external_state": (
            caldav.get("external_state") if isinstance(caldav, dict) else "local"
        ),
        f"{prefix}outbox_operation_id": (
            caldav.get("outbox_operation_id") if isinstance(caldav, dict) else None
        ),
    }


def record_calendar_event_change(
    session: Session,
    *,
    event: CalendarEvent,
    operation: str,
    user_id: str | None,
    previous: dict[str, Any] | None = None,
) -> None:
    payload = calendar_event_change_payload(event)
    if previous:
        payload.update(previous)
    record_change(
        session,
        module_id=CALENDAR_MODULE_ID,
        collection=CALENDAR_EVENTS_COLLECTION,
        resource_type=CALENDAR_EVENT_RESOURCE,
        resource_id=event.id,
        operation=operation,
        tenant_id=event.tenant_id,
        actor_type="user" if user_id else "system",
        actor_id=user_id,
        payload=payload,
    )


@dataclass(slots=True)
class CalendarCalDavSyncStats:
    created: int = 0
    updated: int = 0
    deleted: int = 0
    unchanged: int = 0
    fetched: int = 0
    full_sync: bool = False
    used_sync_token: bool = False
    sync_token: str | None = None
    ctag: str | None = None
    errors: list[str] = field(default_factory=list)


@dataclass(slots=True)
class CalendarCalDavDueSyncResult:
    source_id: str
    calendar_id: str
    status: str
    stats: CalendarCalDavSyncStats | None = None
    error: str | None = None


@dataclass(frozen=True, slots=True)
class _CalDAVSyncSnapshot:
    source_id: str
    tenant_id: str
    calendar_id: str
    collection_url: str
    auth_type: str
    username: str | None
    credential_ref: str | None
    sync_direction: str
    conflict_policy: str
    sync_token: str | None
    ctag: str | None


@dataclass(frozen=True, slots=True)
class _RemoteSyncSnapshot:
    source_id: str
    tenant_id: str
    calendar_id: str
    source_kind: str
    collection_url: str
    auth_type: str
    username: str | None
    credential_ref: str | None
    sync_token: str | None
    ctag: str | None
    metadata: dict[str, Any]


@dataclass(frozen=True, slots=True)
class _CalDAVDiscoveryAuth:
    auth_type: str
    username: str | None
    credential_ref: str | None
    secret: str | None = field(repr=False)


@dataclass(frozen=True, slots=True)
class _SyncSourceCreationPlan:
    source_kind: str
    collection_url: str
    sync_direction: str
    auth_type: str
    username: str | None
    credential_ref: str | None
    has_inline_secret: bool

    def build_source(
        self,
        *,
        tenant_id: str,
        calendar_id: str,
        payload: CalendarSyncSourceCreateRequest,
    ) -> CalendarSyncSource:
        return CalendarSyncSource(
            tenant_id=tenant_id,
            calendar_id=calendar_id,
            source_kind=self.source_kind,
            collection_url=self.collection_url,
            display_name=payload.display_name,
            auth_type=self.auth_type,
            username=self.username,
            credential_ref=self.credential_ref,
            sync_enabled=payload.sync_enabled,
            sync_interval_seconds=payload.sync_interval_seconds,
            sync_direction=self.sync_direction,
            conflict_policy=payload.conflict_policy,
            metadata_=dict(payload.metadata),
        )


def slugify(value: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
    return slug or "calendar"


def normalize_source_kind(value: str) -> str:
    source_kind = (value or "").strip().lower()
    if source_kind not in SYNC_SOURCE_KINDS:
        raise CalendarError(f"Unsupported calendar sync source kind: {value}")
    return source_kind


def normalize_sync_source_url(source_kind: str, value: str) -> str:
    url = value.strip()
    if not url:
        raise CalendarError("Calendar sync source URL is required")
    if source_kind == "caldav":
        return ensure_collection_url(url)
    if source_kind == "webcal":
        if url.lower().startswith("webcal://"):
            url = "https://" + url[len("webcal://"):]
        return validate_http_url(url, label="webcal URL")
    if source_kind == "ics":
        if url.lower().startswith("webcal://"):
            return normalize_sync_source_url("webcal", url)
        return validate_http_url(url, label="ICS URL")
    if source_kind == "graph":
        return normalize_graph_collection_url(url)
    if source_kind == "ews":
        return validate_http_url(url, label="Exchange Web Services URL")
    return url


def validate_http_url(url: str, *, label: str) -> str:
    parsed = urllib.parse.urlparse(url.strip())
    if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname:
        raise CalendarError(f"{label} must be an absolute HTTP(S) URL")
    if parsed.username or parsed.password:
        raise CalendarError(f"{label} must not include embedded credentials")
    _parsed_http_origin(parsed, label=label)
    return urllib.parse.urlunparse(parsed)


def same_origin_http_url(base_url: str, candidate_url: str, *, label: str) -> str:
    base = validate_http_url(base_url, label=label)
    candidate = validate_http_url(candidate_url, label=label)
    if _parsed_http_origin(urllib.parse.urlparse(candidate), label=label) != _parsed_http_origin(
        urllib.parse.urlparse(base),
        label=label,
    ):
        raise CalendarError(f"{label} must remain on the configured source origin")
    return candidate


def _parsed_http_origin(parsed: urllib.parse.ParseResult, *, label: str) -> tuple[str, str, int]:
    try:
        port = parsed.port
    except ValueError as exc:
        raise CalendarError(f"{label} has an invalid port") from exc
    scheme = parsed.scheme.lower()
    if port is None:
        port = 443 if scheme == "https" else 80
    return scheme, (parsed.hostname or "").lower(), port


class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler):
    def __init__(self, source_url: str) -> None:
        super().__init__()
        parsed = urllib.parse.urlparse(validate_http_url(source_url, label="Calendar source URL"))
        self._source_origin = _parsed_http_origin(parsed, label="Calendar source URL")

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # type: ignore[no-untyped-def]
        try:
            candidate = validate_http_url(newurl, label="Calendar source redirect URL")
            candidate = validate_outbound_http_url(candidate, label="Calendar source redirect URL")
            candidate_origin = _parsed_http_origin(
                urllib.parse.urlparse(candidate),
                label="Calendar source redirect URL",
            )
        except (CalendarError, OutboundHttpError):
            return None
        if candidate_origin != self._source_origin:
            return None
        return super().redirect_request(req, fp, code, msg, headers, candidate)


def normalize_graph_collection_url(value: str) -> str:
    url = value.strip()
    if url.startswith("/"):
        url = url[1:]
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme:
        return validate_http_url(url, label="Microsoft Graph calendar URL")
    if url in {"me/calendar", "me/calendar/events", "me/calendar/events/delta"}:
        path = "me/calendar/events/delta"
    elif url in {"me/events", "me/events/delta"}:
        path = "me/events/delta"
    elif url.endswith("/events/delta") or url.endswith("/calendarView/delta"):
        path = url
    elif url.endswith("/events"):
        path = f"{url}/delta"
    else:
        path = f"{url.rstrip('/')}/events/delta"
    return urllib.parse.urljoin(GRAPH_DEFAULT_BASE_URL, path)


def sync_source_label(source_kind: str) -> str:
    return {
        "caldav": "CalDAV",
        "ics": "ICS",
        "webcal": "webcal",
        "graph": "Microsoft Graph",
        "ews": "Exchange Web Services",
    }.get(source_kind, source_kind)


def _sync_stats_payload(stats: CalendarCalDavSyncStats | None) -> dict[str, object]:
    if stats is None:
        return {}
    return {
        "created": stats.created,
        "updated": stats.updated,
        "deleted": stats.deleted,
        "unchanged": stats.unchanged,
        "fetched": stats.fetched,
        "full_sync": stats.full_sync,
        "used_sync_token": stats.used_sync_token,
        "errors": list(stats.errors),
    }


def _emit_calendar_sync_notification(
    session: Session,
    *,
    source: CalendarSyncSource,
    status: str,
    previous_status: str | None,
    stats: CalendarCalDavSyncStats | None = None,
    error: str | None = None,
) -> None:
    if status == "ok" and previous_status != "error":
        return
    if status == "error" and previous_status == "error":
        return
    provider = notification_dispatch_provider(get_registry())
    if provider is None:
        return

    label = source.display_name or sync_source_label(source.source_kind)
    calendar_name = source.calendar.name if source.calendar else "calendar"
    if status == "ok":
        subject = f"Calendar sync recovered: {calendar_name}"
        body_text = f"{label} synchronized successfully after a previous failure."
        priority = 1
    else:
        subject = f"Calendar sync failed: {calendar_name}"
        body_text = f"{label} could not be synchronized. {error or 'No error detail was provided.'}"
        priority = 5
    try:
        provider.enqueue_notification(
            session,
            NotificationDispatchRequest(
                tenant_id=source.tenant_id,
                source_module=CALENDAR_MODULE_ID,
                source_resource_type="calendar_sync_source",
                source_resource_id=source.id,
                event_kind=f"calendar.sync.{status}",
                channel="inbox",
                subject=subject,
                body_text=body_text,
                action_url="/calendar",
                priority=priority,
                payload={
                    "calendar_id": source.calendar_id,
                    "source_id": source.id,
                    "source_kind": source.source_kind,
                    "status": status,
                    "previous_status": previous_status,
                    "error": error,
                    "stats": _sync_stats_payload(stats),
                },
                metadata={"calendar_id": source.calendar_id, "source_kind": source.source_kind},
            ),
            enqueue_delivery=False,
        )
    except Exception:
        return


def _finalize_sync_success(
    session: Session,
    *,
    source: CalendarSyncSource,
    stats: CalendarCalDavSyncStats | None = None,
    mark_calendar: Callable[[CalendarCollection, CalendarSyncSource], None] | None = None,
    update_stats_tokens: bool = False,
) -> None:
    source.last_synced_at = utcnow()
    source.last_status = "ok"
    source.last_error = None
    schedule_next_caldav_sync(source)
    if stats is not None and update_stats_tokens:
        stats.sync_token = source.sync_token
        stats.ctag = source.ctag
    if mark_calendar is not None and source.calendar is not None:
        mark_calendar(source.calendar, source)
    session.flush()


def _finalize_sync_error(session: Session, *, source: CalendarSyncSource, error: BaseException) -> None:
    source.last_attempt_at = utcnow()
    source.last_status = "error"
    source.last_error = str(error)
    schedule_next_caldav_sync(source)
    session.flush()


def ensure_default_calendar(session: Session, *, tenant_id: str, user_id: str | None = None) -> CalendarCollection:
    existing = (
        session.query(CalendarCollection)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.is_default.is_(True), CalendarCollection.deleted_at.is_(None))
        .order_by(CalendarCollection.created_at.asc())
        .first()
    )
    if existing:
        return existing
    calendar = CalendarCollection(
        tenant_id=tenant_id,
        slug="default",
        name="Calendar",
        timezone="UTC",
        color="#0f766e",
        owner_type="tenant",
        owner_id=None,
        visibility="tenant",
        is_default=True,
        created_by_user_id=user_id,
        metadata_={},
    )
    session.add(calendar)
    session.flush()
    return calendar


def get_default_calendar(session: Session, *, tenant_id: str) -> CalendarCollection | None:
    return (
        session.query(CalendarCollection)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.is_default.is_(True), CalendarCollection.deleted_at.is_(None))
        .order_by(CalendarCollection.created_at.asc())
        .first()
    )


def calendar_is_visible_to_principal(
    calendar: CalendarCollection,
    *,
    user_id: str | None,
    group_ids: Iterable[str] = (),
    can_admin: bool = False,
) -> bool:
    """Return whether a principal may discover a calendar collection.

    Collection visibility is deliberately enforced before serializing picker
    metadata.  ``shared`` currently means tenant-visible; explicit per-user
    sharing can narrow this rule when Calendar gains a share relation.
    """

    if can_admin:
        return True
    if calendar.visibility in {"tenant", "shared", "public"}:
        return True
    if calendar.visibility != "private" or user_id is None:
        return False
    if calendar.created_by_user_id == user_id:
        return True
    if calendar.owner_type == "user":
        return calendar.owner_id == user_id
    if calendar.owner_type == "group":
        return bool(calendar.owner_id and calendar.owner_id in set(group_ids))
    return False


def list_calendars(
    session: Session,
    *,
    tenant_id: str,
    ensure_default: bool = False,
    user_id: str | None = None,
    group_ids: Iterable[str] = (),
    can_admin: bool = False,
) -> list[CalendarCollection]:
    if ensure_default:
        ensure_default_calendar(session, tenant_id=tenant_id, user_id=user_id)
    calendars = (
        session.query(CalendarCollection)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.deleted_at.is_(None))
        .order_by(CalendarCollection.is_default.desc(), CalendarCollection.name.asc())
        .all()
    )
    principal_group_ids = tuple(group_ids)
    return [
        calendar
        for calendar in calendars
        if calendar_is_visible_to_principal(
            calendar,
            user_id=user_id,
            group_ids=principal_group_ids,
            can_admin=can_admin,
        )
    ]


def create_calendar(session: Session, *, tenant_id: str, user_id: str | None, payload: CalendarCollectionCreateRequest) -> CalendarCollection:
    slug = payload.slug or slugify(payload.name)
    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,
        slug=slug,
        name=payload.name,
        description=payload.description,
        timezone=payload.timezone,
        color=payload.color,
        owner_type=payload.owner_type,
        owner_id=payload.owner_id,
        visibility=payload.visibility,
        is_default=payload.is_default,
        created_by_user_id=user_id,
        metadata_=payload.metadata,
    )
    session.add(calendar)
    session.flush()
    return calendar


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:
        calendar.is_default = False
    for attr in ("name", "description", "timezone", "color", "visibility"):
        value = getattr(payload, attr)
        if value is not None:
            setattr(calendar, attr, value)
    if payload.metadata is not None:
        calendar.metadata_ = payload.metadata
    session.flush()
    return calendar


def _lock_calendar_move_context(
    session: Session,
    *,
    tenant_id: str,
    calendar: CalendarCollection,
    target_calendar: CalendarCollection,
) -> tuple[
    CalendarCollection,
    CalendarCollection,
    CalendarSyncSource | None,
    CalendarSyncSource | None,
]:
    if not hasattr(session, "query"):
        return calendar, target_calendar, None, None
    # Source creation takes the collection row lock before linking a source.
    # Lock both collections in stable order so their modes cannot change
    # between validation and the local/outbox mutation.
    locked_calendars = (
        session.query(CalendarCollection)
        .filter(
            CalendarCollection.tenant_id == tenant_id,
            CalendarCollection.id.in_((calendar.id, target_calendar.id)),
            CalendarCollection.deleted_at.is_(None),
        )
        .order_by(CalendarCollection.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    locked_calendar_by_id = {item.id: item for item in locked_calendars}
    if len(locked_calendar_by_id) != 2:
        raise CalendarError("Source or target calendar is no longer available")
    calendar = locked_calendar_by_id[calendar.id]
    target_calendar = locked_calendar_by_id[target_calendar.id]
    source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=calendar.id,
    )
    target_source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=target_calendar.id,
    )
    locked_sources = lock_sync_sources(session, source, target_source)
    if source is not None:
        source = locked_sources[source.id]
    if target_source is not None:
        target_source = locked_sources[target_source.id]
    return calendar, target_calendar, source, target_source


def _validate_calendar_move_source_pair(
    payload: CalendarCollectionDeleteRequest,
    *,
    source: CalendarSyncSource | None,
    target_source: CalendarSyncSource | None,
) -> None:
    if source is not None and target_source is not None:
        if payload.external_action == "remote_move":
            return
        raise CalendarError(
            "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(
    session: Session,
    *,
    tenant_id: str,
    payload: CalendarCollectionDeleteRequest,
    source: CalendarSyncSource | None,
    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(
                "Moving events from a synchronized calendar to a local calendar requires "
                "external_action='detach_keep_remote'"
            )
        # Retirement rejects a live delivery lease before any local event
        # projection is changed, and cancels all remaining undelivered desired
        # state. It never creates a remote DELETE operation.
        retire_sync_source(
            session,
            tenant_id=tenant_id,
            source=source,
            deleted_at=deleted_at,
        )
        return
    if target_source is not None:
        if payload.external_action != "copy_to_remote":
            raise CalendarError(
                "Moving local events to a synchronized calendar requires "
                "external_action='copy_to_remote'"
            )
        if (
            target_source.source_kind != "caldav"
            or target_source.sync_direction != "two_way"
            or not target_source.sync_enabled
        ):
            raise CalendarError(
                "external_action='copy_to_remote' requires an active two-way CalDAV target"
            )
        assert_sync_mutation_allowed(target_source)
        return
    if payload.external_action is not None:
        raise CalendarError(
            "external_action is only valid when moving events to or from a synchronized calendar"
        )


def _active_events_with_previous_state(
    session: Session,
    calendar: CalendarCollection,
) -> tuple[list[CalendarEvent], dict[str, dict[str, Any]]]:
    active_events = [event for event in calendar.events if event.deleted_at is None]
    previous_event_states = (
        {
            event.id: calendar_event_change_payload(event, prefix="previous_")
            for event in active_events
        }
        if hasattr(session, "query")
        else {}
    )
    return active_events, previous_event_states


def _move_calendar_events(
    session: Session,
    *,
    tenant_id: str,
    payload: CalendarCollectionDeleteRequest,
    target_calendar: CalendarCollection,
    source: CalendarSyncSource | None,
    target_source: CalendarSyncSource | None,
    active_events: list[CalendarEvent],
    previous_event_states: dict[str, dict[str, Any]],
) -> None:
    for event in active_events:
        if source is not None or target_source is not None:
            event.source_kind = "local"
            event.source_href = None
            event.etag = None
            metadata = dict(event.metadata_ or {})
            metadata.pop("caldav", None)
            event.metadata_ = metadata
        event.calendar_id = target_calendar.id
    session.flush()
    if target_source is not None:
        from govoplan_calendar.backend.outbox import enqueue_caldav_put

        for event in active_events:
            enqueue_caldav_put(session, source=target_source, event_model=event)
    if previous_event_states:
        for event in active_events:
            record_calendar_event_change(
                session,
                event=event,
                operation="updated",
                user_id=None,
                previous=previous_event_states[event.id],
            )
    if payload.make_target_default:
        clear_default_calendar(session, tenant_id=tenant_id)
        target_calendar.is_default = True


def _move_calendar_before_delete(
    session: Session,
    *,
    tenant_id: str,
    calendar: CalendarCollection,
    payload: CalendarCollectionDeleteRequest,
    deleted_at: datetime,
    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:
        raise CalendarError("Target calendar must be different from the deleted calendar")
    target_calendar = get_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=payload.target_calendar_id,
    )
    calendar, target_calendar, source, target_source = _lock_calendar_move_context(
        session,
        tenant_id=tenant_id,
        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,
        target_source=target_source,
    )
    active_events, previous_event_states = _active_events_with_previous_state(
        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,
        payload=payload,
        source=source,
        target_source=target_source,
        deleted_at=deleted_at,
    )
    _move_calendar_events(
        session,
        tenant_id=tenant_id,
        payload=payload,
        target_calendar=target_calendar,
        source=source,
        target_source=target_source,
        active_events=active_events,
        previous_event_states=previous_event_states,
    )
    return calendar, None


def delete_calendar(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str,
    payload: CalendarCollectionDeleteRequest | None = None,
    user_id: str | None = None,
    api_key_id: str | 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, 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
    retire_sync_sources_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=calendar.id,
        deleted_at=deleted_at,
        deletion_reason="calendar_deleted",
        user_id=user_id,
        api_key_id=api_key_id,
    )
    for event in calendar.events:
        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:
    calendar = (
        session.query(CalendarCollection)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.id == calendar_id, CalendarCollection.deleted_at.is_(None))
        .first()
    )
    if not calendar:
        raise CalendarError("Calendar not found")
    return calendar


def calendar_slug_exists(session: Session, *, tenant_id: str, slug: str) -> bool:
    return (
        session.query(CalendarCollection.id)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.slug == slug, CalendarCollection.deleted_at.is_(None))
        .first()
        is not None
    )


def clear_default_calendar(session: Session, *, tenant_id: str) -> None:
    for calendar in (
        session.query(CalendarCollection)
        .filter(CalendarCollection.tenant_id == tenant_id, CalendarCollection.is_default.is_(True), CalendarCollection.deleted_at.is_(None))
        .all()
    ):
        calendar.is_default = False


def list_sync_sources(session: Session, *, tenant_id: str, calendar_id: str | None = None, source_kind: str | None = None) -> list[CalendarSyncSource]:
    query = session.query(CalendarSyncSource).join(CalendarCollection, CalendarCollection.id == CalendarSyncSource.calendar_id).filter(
        CalendarSyncSource.tenant_id == tenant_id,
        CalendarSyncSource.deleted_at.is_(None),
        CalendarCollection.tenant_id == tenant_id,
        CalendarCollection.deleted_at.is_(None),
    )
    if source_kind:
        query = query.filter(CalendarSyncSource.source_kind == source_kind)
    if calendar_id:
        query = query.filter(CalendarSyncSource.calendar_id == calendar_id)
    return query.order_by(CalendarSyncSource.source_kind.asc(), CalendarSyncSource.display_name.asc(), CalendarSyncSource.created_at.asc()).all()


def list_caldav_sources(session: Session, *, tenant_id: str, calendar_id: str | None = None) -> list[CalendarSyncSource]:
    return list_sync_sources(session, tenant_id=tenant_id, calendar_id=calendar_id, source_kind="caldav")


def get_sync_source(session: Session, *, tenant_id: str, source_id: str, source_kind: str | None = None) -> CalendarSyncSource:
    source = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.id == source_id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .first()
    )
    if source is not None and source_kind is not None and source.source_kind != source_kind:
        source = None
    if not source:
        raise CalendarError("Calendar sync source not found")
    return source


def get_caldav_source(session: Session, *, tenant_id: str, source_id: str) -> CalendarSyncSource:
    try:
        return get_sync_source(session, tenant_id=tenant_id, source_id=source_id, source_kind="caldav")
    except CalendarError as exc:
        raise CalendarError("CalDAV sync source not found") from exc


def _resolve_caldav_discovery_auth(
    session: Session,
    *,
    tenant_id: str,
    payload: CalendarCalDavDiscoveryRequest,
    source: CalendarSyncSource | None,
) -> _CalDAVDiscoveryAuth:
    auth_type, username, credential_ref = _caldav_discovery_inputs(
        payload,
        source,
    )
    resolved_envelope = _caldav_discovery_envelope(
        session,
        tenant_id=tenant_id,
        payload=payload,
        source=source,
        credential_ref=credential_ref,
    )
    username = username or _credential_username(resolved_envelope)
    secret = _caldav_discovery_secret(
        session,
        payload=payload,
        source=source,
        auth_type=auth_type,
        credential_ref=credential_ref,
        resolved_envelope=resolved_envelope,
    )
    _validate_caldav_discovery_auth(
        auth_type=auth_type,
        username=username,
        secret=secret,
    )
    return _CalDAVDiscoveryAuth(
        auth_type=auth_type,
        username=username,
        credential_ref=credential_ref,
        secret=secret,
    )


def _caldav_discovery_inputs(
    payload: CalendarCalDavDiscoveryRequest,
    source: CalendarSyncSource | None,
) -> tuple[str, str | None, str | None]:
    return (
        payload.auth_type or (source.auth_type if source else "none"),
        (
            payload.username
            if payload.username is not None
            else (source.username if source else None)
        ),
        (
            payload.credential_ref
            if payload.credential_ref is not None
            else (source.credential_ref if source else None)
        ),
    )


def _caldav_discovery_envelope(
    session: Session,
    *,
    tenant_id: str,
    payload: CalendarCalDavDiscoveryRequest,
    source: CalendarSyncSource | None,
    credential_ref: str | None,
) -> ResolvedCredentialEnvelope | None:
    resolved = _resolve_core_calendar_credential(
        session,
        tenant_id=tenant_id,
        source_id=source.id if source else None,
        credential_ref=credential_ref,
    )
    if payload.credential_ref is not None and resolved is None and (
        source is None or payload.credential_ref != source.credential_ref
    ):
        raise CalendarError(
            "Caller-supplied credential references are accepted only for visible reusable credential envelopes"
        )
    return resolved


def _caldav_discovery_secret(
    session: Session,
    *,
    payload: CalendarCalDavDiscoveryRequest,
    source: CalendarSyncSource | None,
    auth_type: str,
    credential_ref: str | None,
    resolved_envelope: ResolvedCredentialEnvelope | None,
) -> str | None:
    secret = caldav_secret_from_payload(
        auth_type=auth_type,
        password=payload.password,
        bearer_token=payload.bearer_token,
    )
    if secret is not None:
        return secret
    if resolved_envelope is not None:
        return _credential_secret(resolved_envelope, auth_type=auth_type)
    if source is not None and credential_ref == source.credential_ref:
        return resolve_caldav_secret(session, source=source)
    return None


def _validate_caldav_discovery_auth(
    *,
    auth_type: str,
    username: str | None,
    secret: str | None,
) -> None:
    if auth_type == "basic":
        if not username:
            raise CalendarError("CalDAV discovery with basic auth requires a username")
        if not secret:
            raise CalendarError("CalDAV discovery with basic auth requires a password or credential reference")
    if auth_type == "bearer" and not secret:
        raise CalendarError(
            "CalDAV discovery with bearer auth requires a token or credential reference"
        )


def _caldav_discovery_client(
    url: str,
    auth: _CalDAVDiscoveryAuth,
) -> CalDAVClient:
    if auth.auth_type == "basic":
        return CalDAVClient(
            collection_url=url,
            username=auth.username,
            password=auth.secret,
        )
    if auth.auth_type == "bearer":
        return CalDAVClient(
            collection_url=url,
            bearer_token=auth.secret,
        )
    return CalDAVClient(collection_url=url)


def _caldav_discovery_response(
    calendars: Iterable[object],
) -> list[dict[str, Any]]:
    return [
        {
            "collection_url": calendar.collection_url,
            "href": calendar.href,
            "display_name": calendar.display_name,
            "color": calendar.color,
            "ctag": calendar.ctag,
            "sync_token": calendar.sync_token,
        }
        for calendar in calendars
    ]


def discover_caldav_calendars(
    session: Session,
    *,
    tenant_id: str,
    payload: CalendarCalDavDiscoveryRequest,
) -> list[dict[str, Any]]:
    source = (
        get_caldav_source(
            session,
            tenant_id=tenant_id,
            source_id=payload.source_id,
        )
        if payload.source_id
        else None
    )
    auth = _resolve_caldav_discovery_auth(
        session,
        tenant_id=tenant_id,
        payload=payload,
        source=source,
    )
    client = _caldav_discovery_client(payload.url, auth)
    return _caldav_discovery_response(client.discover_calendars())


def _plan_sync_source_creation(
    payload: CalendarSyncSourceCreateRequest,
) -> _SyncSourceCreationPlan:
    if payload.credential_ref is not None and not _core_credential_id(
        payload.credential_ref
    ):
        raise CalendarError(
            "Caller-supplied credential references are accepted only for visible reusable credential envelopes"
        )
    has_inline_secret = (
        payload.password is not None or payload.bearer_token is not None
    )
    if payload.credential_ref is not None and has_inline_secret:
        raise CalendarError(
            "Select a reusable credential or enter a new secret, not both"
        )
    source_kind = normalize_source_kind(payload.source_kind)
    collection_url = normalize_sync_source_url(
        source_kind,
        payload.collection_url,
    )
    _validate_sync_source_create_auth(source_kind, payload.auth_type)
    return _SyncSourceCreationPlan(
        source_kind=source_kind,
        collection_url=collection_url,
        sync_direction=(
            "inbound"
            if source_kind in READ_ONLY_SYNC_SOURCE_KINDS
            else payload.sync_direction
        ),
        auth_type=payload.auth_type,
        username=payload.username,
        credential_ref=payload.credential_ref,
        has_inline_secret=has_inline_secret,
    )


def _validate_sync_source_create_auth(
    source_kind: str,
    auth_type: str,
) -> None:
    if source_kind == "graph" and auth_type != "bearer":
        raise CalendarError(
            "Microsoft Graph calendar sync requires bearer token authentication"
        )
    if source_kind in {"ics", "webcal"} and auth_type not in {
        "none",
        "basic",
        "bearer",
    }:
        raise CalendarError(
            "ICS/webcal subscriptions support none, basic, or bearer authentication"
        )
    if source_kind == "ews" and auth_type not in {"basic", "bearer"}:
        raise CalendarError(
            "Exchange Web Services sync requires basic or bearer authentication"
        )


def _lock_calendar_for_sync_source(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str,
) -> CalendarCollection:
    calendar = get_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=calendar_id,
    )
    return (
        session.query(CalendarCollection)
        .filter(
            CalendarCollection.id == calendar.id,
            CalendarCollection.tenant_id == tenant_id,
            CalendarCollection.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )


def _ensure_calendar_has_no_sync_source(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str,
) -> None:
    existing = (
        session.query(CalendarSyncSource.id)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.calendar_id == calendar_id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .first()
    )
    if existing is not None:
        raise CalendarError(
            "A calendar can have only one active synchronization source"
        )


def _ensure_sync_source_url_available(
    session: Session,
    *,
    tenant_id: str,
    plan: _SyncSourceCreationPlan,
) -> None:
    retire_stale_sync_sources_for_url(
        session,
        tenant_id=tenant_id,
        source_kind=plan.source_kind,
        collection_url=plan.collection_url,
    )
    existing = active_sync_source_for_url(
        session,
        tenant_id=tenant_id,
        source_kind=plan.source_kind,
        collection_url=plan.collection_url,
    )
    if existing is None:
        return
    existing_calendar = existing.calendar
    calendar_name = (
        existing_calendar.name
        if existing_calendar and existing_calendar.deleted_at is None
        else "another calendar"
    )
    raise CalendarError(
        f"{sync_source_label(plan.source_kind)} source is already linked to {calendar_name}"
    )


def _persist_sync_source_credential(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source: CalendarSyncSource,
    payload: CalendarSyncSourceCreateRequest,
) -> None:
    if source.credential_ref:
        _require_visible_core_calendar_credential(
            session,
            tenant_id=tenant_id,
            source_id=source.id,
            credential_ref=source.credential_ref,
        )
        if not source.username:
            resolved = _resolve_core_calendar_credential(
                session,
                tenant_id=tenant_id,
                source_id=source.id,
                credential_ref=source.credential_ref,
            )
            source.username = _credential_username(resolved)
    credential_value = caldav_secret_from_payload(
        auth_type=source.auth_type,
        password=payload.password,
        bearer_token=payload.bearer_token,
    )
    if credential_value is not None:
        source.credential_ref = store_caldav_credential(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            source=source,
            secret=credential_value,
        )

def create_sync_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    payload: CalendarSyncSourceCreateRequest,
) -> CalendarSyncSource:
    plan = _plan_sync_source_creation(payload)
    calendar = _lock_calendar_for_sync_source(
        session,
        tenant_id=tenant_id,
        calendar_id=payload.calendar_id,
    )
    _ensure_calendar_has_no_sync_source(
        session,
        tenant_id=tenant_id,
        calendar_id=calendar.id,
    )
    _ensure_sync_source_url_available(
        session,
        tenant_id=tenant_id,
        plan=plan,
    )
    source = plan.build_source(
        tenant_id=tenant_id,
        calendar_id=calendar.id,
        payload=payload,
    )
    session.add(source)
    session.flush()
    _persist_sync_source_credential(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        source=source,
        payload=payload,
    )
    source.next_sync_at = utcnow() if source.sync_enabled else None
    mark_calendar_sync_source(calendar, source)
    session.flush()
    return source


def create_caldav_source(session: Session, *, tenant_id: str, user_id: str | None, payload: CalendarCalDavSourceCreateRequest) -> CalendarSyncSource:
    return create_sync_source(session, tenant_id=tenant_id, user_id=user_id, payload=payload)


def _lock_sync_source_for_update(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
) -> CalendarSyncSource:
    return (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.id == source_id,
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )


def _sync_source_reconfiguration(
    source: CalendarSyncSource,
    payload: CalendarSyncSourceUpdateRequest,
) -> tuple[str, bool, bool]:
    normalized_collection_url = (
        normalize_sync_source_url(source.source_kind, payload.collection_url)
        if payload.collection_url is not None
        else source.collection_url
    )
    calendar_changed = bool(
        payload.calendar_id is not None and payload.calendar_id != source.calendar_id
    )
    endpoint_changed = normalized_collection_url != source.collection_url
    endpoint_or_calendar_changed = bool(
        source.source_kind == "caldav" and (calendar_changed or endpoint_changed)
    )
    materially_reconfigured = bool(
        source.source_kind == "caldav"
        and (
            calendar_changed
            or endpoint_changed
            or payload.sync_enabled is False
            or payload.sync_direction == "inbound"
        )
    )
    return (
        normalized_collection_url,
        materially_reconfigured,
        endpoint_or_calendar_changed,
    )


def _caldav_source_has_linked_events(
    session: Session,
    *,
    tenant_id: str,
    source: CalendarSyncSource,
) -> bool:
    return (
        session.query(CalendarEvent.id)
        .filter(
            CalendarEvent.tenant_id == tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == "caldav",
            CalendarEvent.source_href.is_not(None),
            CalendarEvent.deleted_at.is_(None),
        )
        .first()
        is not None
    )


def _guard_sync_source_reconfiguration(
    session: Session,
    *,
    tenant_id: str,
    source: CalendarSyncSource,
    payload: CalendarSyncSourceUpdateRequest,
    endpoint_or_calendar_changed: bool,
) -> None:
    from govoplan_calendar.backend.outbox import (
        calendar_outbox_has_live_lease,
        calendar_outbox_has_unresolved_desired_state,
        cancel_calendar_outbox_for_source,
    )

    if calendar_outbox_has_live_lease(session, source_id=source.id):
        raise CalendarError(
            "CalDAV source cannot be materially changed while an outbound operation has an active lease"
        )
    unresolved_desired_state = calendar_outbox_has_unresolved_desired_state(
        session,
        source_id=source.id,
    )
    stops_outbound_delivery = bool(
        (payload.sync_enabled is False and source.sync_enabled)
        or (
            payload.sync_direction == "inbound"
            and source.sync_direction == "two_way"
        )
    )
    if stops_outbound_delivery and unresolved_desired_state:
        raise CalendarError(
            "CalDAV source cannot disable outbound delivery while unresolved local desired "
            "changes exist; deliver or explicitly discard them first"
        )
    if endpoint_or_calendar_changed and unresolved_desired_state:
        raise CalendarError(
            "CalDAV source endpoint or calendar cannot change while unresolved local desired "
            "changes exist; deliver or explicitly discard them first"
        )
    if endpoint_or_calendar_changed and _caldav_source_has_linked_events(
        session,
        tenant_id=tenant_id,
        source=source,
    ):
        raise CalendarError(
            "CalDAV source endpoint or calendar cannot change while events still reference it; "
            "detach or resync those events first"
        )
    cancel_calendar_outbox_for_source(
        session,
        source_id=source.id,
        reason="CalDAV source endpoint/calendar changed, was disabled, or was made inbound-only",
    )


def _calendar_for_sync_source_update(
    session: Session,
    *,
    tenant_id: str,
    source: CalendarSyncSource,
    calendar_id: str | None,
) -> CalendarCollection:
    if calendar_id is None:
        return get_calendar(
            session,
            tenant_id=tenant_id,
            calendar_id=source.calendar_id,
        )
    calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id)
    calendar = (
        session.query(CalendarCollection)
        .filter(
            CalendarCollection.id == calendar.id,
            CalendarCollection.tenant_id == tenant_id,
            CalendarCollection.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )
    conflicting_source = (
        session.query(CalendarSyncSource.id)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.calendar_id == calendar.id,
            CalendarSyncSource.id != source.id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .first()
    )
    if conflicting_source is not None:
        raise CalendarError("A calendar can have only one active synchronization source")
    source.calendar_id = calendar.id
    return calendar


def _apply_sync_source_values(
    source: CalendarSyncSource,
    *,
    payload: CalendarSyncSourceUpdateRequest,
    normalized_collection_url: str,
) -> None:
    for field_name in (
        "display_name",
        "auth_type",
        "username",
        "credential_ref",
        "sync_enabled",
        "sync_interval_seconds",
        "sync_direction",
        "conflict_policy",
        "sync_token",
        "ctag",
    ):
        value = getattr(payload, field_name)
        if value is not None:
            setattr(source, field_name, value)
    if payload.collection_url is not None:
        source.collection_url = normalized_collection_url
    if payload.metadata is not None:
        source.metadata_ = payload.metadata
    if source.source_kind in READ_ONLY_SYNC_SOURCE_KINDS:
        source.sync_direction = "inbound"


def _validate_sync_source_auth(source: CalendarSyncSource) -> None:
    if source.source_kind == "graph" and source.auth_type != "bearer":
        raise CalendarError(
            "Microsoft Graph calendar sync requires bearer token authentication"
        )
    if source.source_kind == "ews" and source.auth_type not in {"basic", "bearer"}:
        raise CalendarError(
            "Exchange Web Services sync requires basic or bearer authentication"
        )


def _update_sync_source_credential_and_schedule(
    session: Session,
    *,
    tenant_id: str,
    source: CalendarSyncSource,
    payload: CalendarSyncSourceUpdateRequest,
    previous_auth_type: str,
    previous_credential_ref: str | None,
    user_id: str | None,
    api_key_id: str | None,
) -> None:
    credential_value = caldav_secret_from_payload(
        auth_type=source.auth_type,
        password=payload.password,
        bearer_token=payload.bearer_token,
    )
    if credential_value is not None:
        if previous_credential_ref and previous_credential_ref != source.credential_ref:
            delete_caldav_credential(
                session,
                tenant_id=tenant_id,
                source_id=source.id,
                credential_ref=previous_credential_ref,
                deletion_reason="credential_replaced",
                user_id=user_id,
                api_key_id=api_key_id,
            )
        source.credential_ref = store_caldav_credential(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            source=source,
            secret=credential_value,
            api_key_id=api_key_id,
        )
    elif "credential_ref" in payload.model_fields_set and source.credential_ref != previous_credential_ref:
        delete_caldav_credential(
            session,
            tenant_id=tenant_id,
            source_id=source.id,
            credential_ref=previous_credential_ref,
            deletion_reason="credential_replaced",
            user_id=user_id,
            api_key_id=api_key_id,
        )
    elif payload.auth_type is not None and source.auth_type != previous_auth_type:
        delete_caldav_credential(
            session,
            tenant_id=tenant_id,
            source_id=source.id,
            credential_ref=source.credential_ref,
            deletion_reason=(
                "authentication_disabled"
                if source.auth_type == "none"
                else "authentication_type_changed"
            ),
            user_id=user_id,
            api_key_id=api_key_id,
        )
        source.credential_ref = None
    if payload.sync_enabled is not None or payload.sync_interval_seconds is not None:
        source.next_sync_at = utcnow() if source.sync_enabled else None


def update_sync_source(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    payload: CalendarSyncSourceUpdateRequest,
    user_id: str | None = None,
    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,
        source_id=source.id,
    )
    if payload.credential_ref is not None and payload.credential_ref != source.credential_ref:
        _require_visible_core_calendar_credential(
            session,
            tenant_id=tenant_id,
            source_id=source.id,
            credential_ref=payload.credential_ref,
        )
    if payload.credential_ref is not None and (
        payload.password is not None or payload.bearer_token is not None
    ):
        raise CalendarError("Select a reusable credential or enter a replacement secret, not both")
    previous_auth_type = source.auth_type
    previous_credential_ref = source.credential_ref
    (
        normalized_collection_url,
        materially_reconfigured,
        endpoint_or_calendar_changed,
    ) = _sync_source_reconfiguration(source, payload)
    if materially_reconfigured:
        _guard_sync_source_reconfiguration(
            session,
            tenant_id=tenant_id,
            source=source,
            payload=payload,
            endpoint_or_calendar_changed=endpoint_or_calendar_changed,
        )
    calendar = _calendar_for_sync_source_update(
        session,
        tenant_id=tenant_id,
        source=source,
        calendar_id=payload.calendar_id,
    )
    _apply_sync_source_values(
        source,
        payload=payload,
        normalized_collection_url=normalized_collection_url,
    )
    if payload.credential_ref is not None and payload.username is None:
        reusable = _resolve_core_calendar_credential(
            session,
            tenant_id=tenant_id,
            source_id=source.id,
            credential_ref=payload.credential_ref,
        )
        source.username = _credential_username(reusable) or source.username
    _validate_sync_source_auth(source)
    _update_sync_source_credential_and_schedule(
        session,
        tenant_id=tenant_id,
        source=source,
        payload=payload,
        previous_auth_type=previous_auth_type,
        previous_credential_ref=previous_credential_ref,
        user_id=user_id,
        api_key_id=api_key_id,
    )
    mark_calendar_sync_source(calendar, source)
    session.flush()
    return source


def update_caldav_source(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    payload: CalendarCalDavSourceUpdateRequest,
    user_id: str | None = None,
    api_key_id: str | None = None,
) -> CalendarSyncSource:
    source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id, source_kind="caldav")
    return update_sync_source(
        session,
        tenant_id=tenant_id,
        source_id=source.id,
        payload=payload,
        user_id=user_id,
        api_key_id=api_key_id,
    )


def delete_sync_source(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    user_id: str | None = None,
    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,
        source=source,
        deleted_at=utcnow(),
        deletion_reason="sync_source_deleted",
        user_id=user_id,
        api_key_id=api_key_id,
    )
    session.flush()


def delete_caldav_source(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    user_id: str | None = None,
    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,
        source=source,
        deleted_at=utcnow(),
        deletion_reason="sync_source_deleted",
        user_id=user_id,
        api_key_id=api_key_id,
    )
    session.flush()


def retire_sync_sources_for_calendar(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str,
    deleted_at: datetime,
    deletion_reason: str = "calendar_deleted",
    user_id: str | None = None,
    api_key_id: str | None = None,
) -> None:
    if not hasattr(session, "query"):
        return
    sources = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.calendar_id == calendar_id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .all()
    )
    for source in sources:
        retire_sync_source(
            session,
            tenant_id=tenant_id,
            source=source,
            deleted_at=deleted_at,
            deletion_reason=deletion_reason,
            user_id=user_id,
            api_key_id=api_key_id,
        )


def retire_caldav_sources_for_calendar(session: Session, *, tenant_id: str, calendar_id: str, deleted_at: datetime) -> None:
    retire_sync_sources_for_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id, deleted_at=deleted_at)


def retire_stale_sync_sources_for_url(session: Session, *, tenant_id: str, source_kind: str, collection_url: str) -> None:
    sources = (
        session.query(CalendarSyncSource)
        .outerjoin(CalendarCollection, CalendarCollection.id == CalendarSyncSource.calendar_id)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.source_kind == source_kind,
            CalendarSyncSource.collection_url == collection_url,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .all()
    )
    now = utcnow()
    for source in sources:
        if source.calendar is None or source.calendar.deleted_at is not None:
            retire_sync_source(
                session,
                tenant_id=tenant_id,
                source=source,
                deleted_at=now,
                deletion_reason="stale_sync_source_retired",
            )
    session.flush()


def retire_stale_caldav_sources_for_url(session: Session, *, tenant_id: str, collection_url: str) -> None:
    retire_stale_sync_sources_for_url(session, tenant_id=tenant_id, source_kind="caldav", collection_url=collection_url)


def active_sync_source_for_url(session: Session, *, tenant_id: str, source_kind: str, collection_url: str) -> CalendarSyncSource | None:
    return (
        session.query(CalendarSyncSource)
        .join(CalendarCollection, CalendarCollection.id == CalendarSyncSource.calendar_id)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.source_kind == source_kind,
            CalendarSyncSource.collection_url == collection_url,
            CalendarSyncSource.deleted_at.is_(None),
            CalendarCollection.tenant_id == tenant_id,
            CalendarCollection.deleted_at.is_(None),
        )
        .order_by(CalendarSyncSource.created_at.asc())
        .first()
    )


def active_caldav_source_for_url(session: Session, *, tenant_id: str, collection_url: str) -> CalendarSyncSource | None:
    return active_sync_source_for_url(session, tenant_id=tenant_id, source_kind="caldav", collection_url=collection_url)


def retire_sync_source(
    session: Session,
    *,
    tenant_id: str,
    source: CalendarSyncSource,
    deleted_at: datetime,
    deletion_reason: str = "sync_source_retired",
    user_id: str | None = None,
    api_key_id: str | None = None,
) -> None:
    from govoplan_calendar.backend.outbox import (
        calendar_outbox_has_live_lease,
        cancel_calendar_outbox_for_source,
    )

    source = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.id == source.id,
            CalendarSyncSource.tenant_id == tenant_id,
        )
        .populate_existing()
        .with_for_update()
        .one()
    )
    if calendar_outbox_has_live_lease(session, source_id=source.id):
        raise CalendarError(
            "CalDAV source cannot be retired while an outbound operation has an active lease"
        )

    # Delete credentials first. An external secret provider cannot participate
    # in the SQL transaction, so provider failure must leave the source and its
    # queued work untouched. If the later SQL commit fails after provider
    # success, the source fails closed and the idempotent delete can be retried.
    delete_caldav_credential(
        session,
        tenant_id=tenant_id,
        source_id=source.id,
        credential_ref=source.credential_ref,
        deletion_reason=deletion_reason,
        user_id=user_id,
        api_key_id=api_key_id,
    )
    source.credential_ref = None
    cancel_calendar_outbox_for_source(
        session,
        source_id=source.id,
        reason="CalDAV source was retired before queued external changes were delivered",
    )
    source.deleted_at = deleted_at


def retire_caldav_source(session: Session, *, tenant_id: str, source: CalendarSyncSource, deleted_at: datetime) -> None:
    retire_sync_source(session, tenant_id=tenant_id, source=source, deleted_at=deleted_at)


def caldav_secret_from_payload(*, auth_type: str, password: Any | None, bearer_token: Any | None) -> str | None:
    if auth_type == "basic" and password is not None:
        return secret_value(password)
    if auth_type == "bearer" and bearer_token is not None:
        return secret_value(bearer_token)
    if auth_type == "none":
        return None
    return None


def secret_value(value: Any) -> str:
    if hasattr(value, "get_secret_value"):
        return str(value.get_secret_value())
    return str(value)


def store_caldav_credential(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source: CalendarSyncSource,
    secret: str,
    api_key_id: str | None = None,
) -> str:
    existing = internal_caldav_credential(
        session,
        tenant_id=tenant_id,
        source_id=source.id,
        credential_ref=source.credential_ref,
    )
    replacing_existing = existing is not None
    if existing is None:
        existing = CalendarSyncCredential(
            tenant_id=tenant_id,
            credential_kind=credential_kind_for_auth_type(source.auth_type),
            label=source.display_name or source.collection_url,
            created_by_user_id=user_id,
            metadata_={"source_id": source.id},
        )
        session.add(existing)
        session.flush()

    old_provider_ref = _credential_provider_ref(existing)
    provider = secret_provider()
    name = f"caldav:{source.id}:{source.auth_type}"
    if old_provider_ref and provider is None:
        raise CalendarError(
            "Stored sync credential cannot be replaced while its secret provider is unavailable"
        )
    if provider is not None:
        try:
            provider_ref = str(
                provider.store_secret(
                    scope=f"calendar:{tenant_id}",
                    name=name,
                    value=secret,
                )
            ).strip()
        except Exception:
            raise CalendarError("Stored sync credential could not be written to its secret provider") from None
        if not provider_ref:
            raise CalendarError("Secret provider returned an empty credential reference")
        if old_provider_ref and old_provider_ref != provider_ref:
            try:
                provider.delete_secret(old_provider_ref)
            except Exception:
                if not _provider_secret_is_absent(provider, old_provider_ref):
                    compensation_failed = False
                    try:
                        provider.delete_secret(provider_ref)
                    except Exception:
                        compensation_failed = not _provider_secret_is_absent(provider, provider_ref)
                    compensation_detail = (
                        "; replacement cleanup also failed and requires provider-side reconciliation"
                        if compensation_failed
                        else ""
                    )
                    raise CalendarError(
                        "Previous sync credential could not be deleted from its secret provider; "
                        f"the replacement was not activated{compensation_detail}"
                    ) from None
        existing.secret_encrypted = None
        existing.metadata_ = {"source_id": source.id, "provider_ref": provider_ref}
    else:
        existing.secret_encrypted = encrypt_secret(secret)
        existing.metadata_ = {"source_id": source.id}
    existing.credential_kind = credential_kind_for_auth_type(source.auth_type)
    existing.label = source.display_name or source.collection_url
    existing.deleted_at = None
    if replacing_existing:
        audit_event(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            api_key_id=api_key_id,
            action="calendar.sync_credential_rotated",
            object_type="calendar_sync_credential",
            object_id=existing.id,
            details={
                "sync_source_id": source.id,
                "storage_backend": (
                    "external_secret_provider"
                    if provider is not None
                    else "encrypted_database"
                ),
            },
        )
    session.flush()
    return f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{existing.id}"


def delete_caldav_credential(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    credential_ref: str | None,
    deletion_reason: str = "sync_source_retired",
    user_id: str | None = None,
    api_key_id: str | None = None,
) -> bool:
    if not credential_ref:
        return False
    credential = internal_caldav_credential(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        credential_ref=credential_ref,
    )
    if credential is None:
        # Legacy external/env references have no locally provable ownership and
        # must never be passed to a provider delete operation.
        return False
    return _delete_calendar_owned_credential(
        session,
        credential=credential,
        source_id=source_id,
        deletion_reason=deletion_reason,
        user_id=user_id,
        api_key_id=api_key_id,
    )


def _delete_calendar_owned_credential(
    session: Session,
    *,
    credential: CalendarSyncCredential,
    source_id: str | None,
    deletion_reason: str,
    user_id: str | None = None,
    api_key_id: str | None = None,
) -> bool:
    provider_ref = _credential_provider_ref(credential)
    if not credential.secret_encrypted and not provider_ref:
        return False
    provider = secret_provider()
    storage_backend = "encrypted_database"
    if provider_ref:
        if provider is None:
            raise CalendarError(
                "Stored sync credential cannot be deleted while its secret provider is unavailable; "
                "the requested deletion was not completed"
            )
        try:
            provider.delete_secret(provider_ref)
        except Exception:
            if not _provider_secret_is_absent(provider, provider_ref):
                raise CalendarError(
                    "Stored sync credential could not be deleted from its secret provider; "
                    "the requested deletion was not completed"
                ) from None
        storage_backend = "external_secret_provider"
    credential.secret_encrypted = None
    credential.metadata_ = {"source_id": source_id} if source_id else {}
    credential.deleted_at = utcnow()
    audit_event(
        session,
        tenant_id=credential.tenant_id,
        user_id=user_id,
        api_key_id=api_key_id,
        action="calendar.sync_credential_deleted",
        object_type="calendar_sync_credential",
        object_id=credential.id,
        details={
            "sync_source_id": source_id,
            "storage_backend": storage_backend,
            "deletion_reason": deletion_reason,
        },
    )
    return True


def _provider_secret_is_absent(provider: Any, provider_ref: str) -> bool:
    """Confirm an idempotent delete when a provider reports missing/error."""

    try:
        return provider.read_secret(provider_ref) is None
    except Exception:
        return False


def delete_calendar_credentials_for_retirement(session: Session) -> int:
    """Delete all Calendar-owned secrets before destructive module retirement.

    This includes pre-hardening soft-deleted rows whose provider references or
    ciphertext were retained. Provider deletion is intentionally fail-closed;
    the caller must not drop Calendar tables if any external deletion fails.
    """

    credentials = session.query(CalendarSyncCredential).order_by(CalendarSyncCredential.id.asc()).all()
    deleted = 0
    for credential in credentials:
        metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {}
        source_id = metadata.get("source_id")
        if _delete_calendar_owned_credential(
            session,
            credential=credential,
            source_id=str(source_id) if source_id else None,
            deletion_reason="module_data_retired",
        ):
            deleted += 1
    session.flush()
    return deleted


def resolve_caldav_secret(
    session: Session,
    *,
    source: CalendarSyncSource,
    password: str | None = None,
    bearer_token: str | None = None,
) -> str | None:
    if source.auth_type == "none":
        return None
    if source.auth_type == "basic" and password:
        return password
    if source.auth_type == "bearer" and bearer_token:
        return bearer_token
    if not source.credential_ref:
        return None
    reusable = _resolve_core_calendar_credential(
        session,
        tenant_id=source.tenant_id,
        source_id=source.id,
        credential_ref=source.credential_ref,
    )
    if reusable is not None:
        return _credential_secret(reusable, auth_type=source.auth_type)
    credential = internal_caldav_credential(
        session,
        tenant_id=source.tenant_id,
        source_id=source.id,
        credential_ref=source.credential_ref,
    )
    if credential is None:
        raise CalendarError(
            "The sync source credential reference is not a server-owned credential for this tenant/source"
        )
    provider_ref = _credential_provider_ref(credential)
    if provider_ref:
        provider = secret_provider()
        if provider is None:
            return None
        return provider.read_secret(provider_ref)
    return decrypt_secret(credential.secret_encrypted) if credential.secret_encrypted else None


def _core_credential_id(credential_ref: str | None) -> str | None:
    if not credential_ref or not credential_ref.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX):
        return None
    value = credential_ref.removeprefix(CORE_CREDENTIAL_ENVELOPE_PREFIX).strip()
    return value or None


def _resolve_core_calendar_credential(
    session: Session,
    *,
    tenant_id: str,
    source_id: str | None,
    credential_ref: str | None,
) -> ResolvedCredentialEnvelope | None:
    credential_id = _core_credential_id(credential_ref)
    if credential_id is None:
        return None
    try:
        return resolve_credential_envelope(
            session,
            credential_id=credential_id,
            context=calendar_credential_context(tenant_id=tenant_id, source_id=source_id),
        )
    except CredentialEnvelopeError as exc:
        raise CalendarError("The selected reusable credential is unavailable to this calendar source") from exc


def _require_visible_core_calendar_credential(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    credential_ref: str,
) -> None:
    credential_id = _core_credential_id(credential_ref)
    if credential_id is None:
        raise CalendarError(
            "Caller-supplied credential references are accepted only for visible reusable credential envelopes"
        )
    try:
        get_credential_envelope(
            session,
            credential_id=credential_id,
            context=calendar_credential_context(tenant_id=tenant_id, source_id=source_id),
        )
    except CredentialEnvelopeError as exc:
        raise CalendarError("The selected reusable credential is unavailable to this calendar source") from exc


def _credential_username(credential: ResolvedCredentialEnvelope | None) -> str | None:
    if credential is None:
        return None
    value = credential.public_data.get("username")
    return str(value).strip() if value is not None and str(value).strip() else None


def _credential_secret(credential: ResolvedCredentialEnvelope, *, auth_type: str) -> str | None:
    keys = (
        ("password", "secret", "token")
        if auth_type == "basic"
        else ("access_token", "bearer_token", "token", "password", "secret")
    )
    for key in keys:
        value = credential.secret_data.get(key)
        if value is not None and str(value):
            return str(value)
    return None


def resolve_caldav_credential_ref(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    credential_ref: str,
) -> str | None:
    credential = internal_caldav_credential(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        credential_ref=credential_ref,
    )
    if credential is None:
        return None
    provider_ref = _credential_provider_ref(credential)
    if provider_ref:
        provider = secret_provider()
        return provider.read_secret(provider_ref) if provider is not None else None
    return decrypt_secret(credential.secret_encrypted) if credential.secret_encrypted else None


def resolve_trusted_deployment_caldav_credential_ref(credential_ref: str) -> str | None:
    """Resolve an env reference only for trusted, out-of-band deployment code.

    HTTP/API source configuration deliberately never calls this function.
    """

    if not credential_ref.startswith(CALDAV_ENV_CREDENTIAL_PREFIX):
        raise CalendarError("Trusted deployment credential references must use the env: prefix")
    env_name = credential_ref.removeprefix(CALDAV_ENV_CREDENTIAL_PREFIX)
    if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", env_name):
        raise CalendarError("Trusted deployment credential reference contains an invalid environment variable name")
    return os.environ.get(env_name)


def caldav_client_for_source(
    session: Session,
    source: CalendarSyncSource,
    *,
    password: str | None = None,
    bearer_token: str | None = None,
) -> CalDAVClient:
    secret = resolve_caldav_secret(session, source=source, password=password, bearer_token=bearer_token)
    username = _source_credential_username(session, source)
    if source.auth_type == "basic":
        if not username:
            raise CalendarError("CalDAV basic auth requires a username")
        if not secret:
            raise CalendarError("CalDAV basic auth requires a stored or transient password")
        return CalDAVClient(collection_url=source.collection_url, username=username, password=secret)
    if source.auth_type == "bearer":
        if not secret:
            raise CalendarError("CalDAV bearer auth requires a stored or transient token")
        return CalDAVClient(collection_url=source.collection_url, bearer_token=secret)
    return CalDAVClient(collection_url=source.collection_url)


def sync_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source_id: str,
    password: str | None = None,
    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()
    with Session(bind=session.get_bind()) as preparation_session:
        source_kind = get_sync_source(
            preparation_session,
            tenant_id=tenant_id,
            source_id=source_id,
        ).source_kind
    if source_kind == "caldav":
        return sync_caldav_source(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            source_id=source_id,
            password=password,
            bearer_token=bearer_token,
            force_full=force_full,
        )
    if source_kind in {"ics", "webcal"}:
        return sync_ics_source(session, tenant_id=tenant_id, user_id=user_id, source_id=source_id, password=password, bearer_token=bearer_token, force_full=force_full)
    if source_kind == "graph":
        return sync_graph_source(session, tenant_id=tenant_id, user_id=user_id, source_id=source_id, bearer_token=bearer_token, force_full=force_full)
    if source_kind == "ews":
        return sync_ews_source(session, tenant_id=tenant_id, user_id=user_id, source_id=source_id, password=password, bearer_token=bearer_token, force_full=force_full)
    raise CalendarError(f"Unsupported calendar sync source kind: {source_kind}")


def http_request(
    url: str,
    *,
    method: str = "GET",
    headers: dict[str, str] | None = None,
    body: str | bytes | None = None,
    timeout: int = 30,
    credential_origin: str | None = None,
) -> tuple[int, dict[str, str], str]:
    data = body.encode("utf-8") if isinstance(body, str) else body
    url = validate_http_url(url, label="Calendar source URL")
    credential_origin = validate_http_url(
        credential_origin or url,
        label="Calendar credential origin",
    )
    url = same_origin_http_url(
        credential_origin,
        url,
        label="Calendar source request URL",
    )
    try:
        url = validate_outbound_http_url(url, label="Calendar source request URL")
    except OutboundHttpError as exc:
        raise CalendarError(str(exc)) from exc
    request = urllib.request.Request(  # noqa: S310 - URL is validated and origin-confined.
        url,
        data=data,
        method=method,
        headers=headers or {},
    )
    opener = build_outbound_http_opener(_SameOriginRedirectHandler(credential_origin))
    try:
        with opener.open(request, timeout=timeout) as response:  # noqa: S310 - validated, origin-confined Calendar URL.  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
            response_headers = {key.lower(): value for key, value in response.headers.items()}
            payload = bounded_response_bytes(
                response,
                headers=response_headers,
                label="Calendar source response",
            ).decode(response_headers.get("content-charset") or "utf-8", errors="replace")
            return int(response.status), response_headers, payload
    except urllib.error.HTTPError as exc:
        if exc.code == 304:
            return 304, {key.lower(): value for key, value in exc.headers.items()}, ""
        try:
            detail = bounded_response_bytes(
                exc,
                headers={key.lower(): value for key, value in exc.headers.items()},
                label="Calendar source error response",
            ).decode("utf-8", errors="replace")
        except OutboundHttpError as policy_exc:
            raise CalendarError(str(policy_exc)) from policy_exc
        raise CalendarError(f"HTTP {exc.code} from calendar source: {detail or exc.reason}") from exc
    except urllib.error.URLError as exc:
        raise CalendarError(f"Calendar source request failed: {exc.reason}") from exc


def source_auth_headers(
    session: Session,
    source: CalendarSyncSource,
    *,
    password: str | None = None,
    bearer_token: str | None = None,
) -> dict[str, str]:
    secret = resolve_caldav_secret(session, source=source, password=password, bearer_token=bearer_token)
    username = _source_credential_username(session, source)
    if source.auth_type == "basic":
        if not username:
            raise CalendarError(f"{sync_source_label(source.source_kind)} basic auth requires a username")
        if not secret:
            raise CalendarError(f"{sync_source_label(source.source_kind)} basic auth requires a stored or transient password")
        import base64

        token = base64.b64encode(f"{username}:{secret}".encode("utf-8")).decode("ascii")
        return {"Authorization": f"Basic {token}"}
    if source.auth_type == "bearer":
        if not secret:
            raise CalendarError(f"{sync_source_label(source.source_kind)} bearer auth requires a stored or transient token")
        return {"Authorization": f"Bearer {secret}"}
    return {}


def _source_credential_username(session: Session, source: CalendarSyncSource) -> str | None:
    reusable = _resolve_core_calendar_credential(
        session,
        tenant_id=source.tenant_id,
        source_id=source.id,
        credential_ref=source.credential_ref,
    )
    return _credential_username(reusable) or source.username


def internal_caldav_credential(
    session: Session,
    *,
    tenant_id: str,
    credential_ref: str | None,
    source_id: str | None = None,
) -> CalendarSyncCredential | None:
    if not credential_ref or not credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX):
        return None
    credential_id = credential_ref.removeprefix(CALDAV_INTERNAL_CREDENTIAL_PREFIX)
    credential = (
        session.query(CalendarSyncCredential)
        .filter(
            CalendarSyncCredential.tenant_id == tenant_id,
            CalendarSyncCredential.id == credential_id,
            CalendarSyncCredential.deleted_at.is_(None),
        )
        .first()
    )
    if credential is None or source_id is None:
        return credential
    metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {}
    return credential if metadata.get("source_id") == source_id else None


def _credential_provider_ref(credential: CalendarSyncCredential) -> str | None:
    metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {}
    value = metadata.get("provider_ref")
    return str(value) if isinstance(value, str) and value.strip() else None


def credential_kind_for_auth_type(auth_type: str) -> str:
    return "bearer_token" if auth_type == "bearer" else "password"


def secret_provider() -> Any | None:
    registry = get_registry()
    if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_SECURITY_SECRET_PROVIDER):
        return None
    try:
        return registry.capability(CAPABILITY_SECURITY_SECRET_PROVIDER)
    except Exception:
        return None


def sync_caldav_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source_id: str,
    client: CalDAVClient | None = None,
    password: str | None = None,
    bearer_token: str | None = None,
    force_full: bool = False,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
    session.commit()
    snapshot, client = _prepare_caldav_sync(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        client=client,
        password=password,
        bearer_token=bearer_token,
    )
    stats = CalendarCalDavSyncStats()
    collection_props = CalDAVReportResult()
    try:
        try:
            collection_props = client.propfind_collection()
        except CalDAVError as exc:
            stats.errors.append(f"Collection PROPFIND failed; continuing with REPORT: {exc}")

        if snapshot.sync_token and not force_full:
            try:
                report = client.sync_collection(snapshot.sync_token)
                stats.used_sync_token = True
            except CalDAVSyncUnsupported as exc:
                stats.errors.append(f"Sync token REPORT failed; falling back to full sync: {exc}")
                report = client.list_objects()
                stats.full_sync = True
        else:
            report = client.list_objects()
            stats.full_sync = True

        report = _hydrate_caldav_report(client, report=report, stats=stats)
        source = _lock_caldav_sync_source(session, snapshot)
        source.last_attempt_at = utcnow()
        apply_caldav_report(
            session,
            source=source,
            report=report,
            stats=stats,
            user_id=user_id,
        )
        source.sync_token = report.sync_token or collection_props.sync_token or source.sync_token
        source.ctag = report.ctag or collection_props.ctag or source.ctag
        _finalize_sync_success(
            session,
            source=source,
            stats=stats,
            mark_calendar=mark_calendar_caldav,
            update_stats_tokens=True,
        )
        return source, stats
    except Exception as exc:
        session.rollback()
        _record_caldav_sync_error(
            session,
            snapshot=snapshot,
            error=exc,
        )
        session.commit()
        raise


def _prepare_caldav_sync(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    client: CalDAVClient | None,
    password: str | None,
    bearer_token: str | None,
) -> tuple[_CalDAVSyncSnapshot, CalDAVClient]:
    """Resolve source configuration without retaining a database connection."""

    with Session(bind=session.get_bind()) as preparation_session:
        source = get_caldav_source(
            preparation_session,
            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,
            calendar_id=source.calendar_id,
        )
        resolved_client = client or caldav_client_for_source(
            preparation_session,
            source,
            password=password,
            bearer_token=bearer_token,
        )
        snapshot = _CalDAVSyncSnapshot(
            source_id=source.id,
            tenant_id=source.tenant_id,
            calendar_id=source.calendar_id,
            collection_url=source.collection_url,
            auth_type=source.auth_type,
            username=source.username,
            credential_ref=source.credential_ref,
            sync_direction=source.sync_direction,
            conflict_policy=source.conflict_policy,
            sync_token=source.sync_token,
            ctag=source.ctag,
        )
    return snapshot, resolved_client


def _lock_caldav_sync_source(
    session: Session,
    snapshot: _CalDAVSyncSnapshot,
) -> CalendarSyncSource:
    source = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.id == snapshot.source_id,
            CalendarSyncSource.tenant_id == snapshot.tenant_id,
            CalendarSyncSource.source_kind == "caldav",
            CalendarSyncSource.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )
    current = (
        source.calendar_id,
        source.collection_url,
        source.auth_type,
        source.username,
        source.credential_ref,
        source.sync_direction,
        source.conflict_policy,
        source.sync_token,
        source.ctag,
    )
    expected = (
        snapshot.calendar_id,
        snapshot.collection_url,
        snapshot.auth_type,
        snapshot.username,
        snapshot.credential_ref,
        snapshot.sync_direction,
        snapshot.conflict_policy,
        snapshot.sync_token,
        snapshot.ctag,
    )
    if current != expected:
        raise CalendarError(
            "CalDAV source changed while the remote response was being fetched; retry the sync."
        )
    return source


def _record_caldav_sync_error(
    session: Session,
    *,
    snapshot: _CalDAVSyncSnapshot,
    error: Exception,
) -> None:
    try:
        source = _lock_caldav_sync_source(session, snapshot)
    except Exception:
        return
    source.last_attempt_at = utcnow()
    _finalize_sync_error(session, source=source, error=error)


def _hydrate_caldav_report(
    client: CalDAVClient,
    *,
    report: CalDAVReportResult,
    stats: CalendarCalDavSyncStats,
) -> CalDAVReportResult:
    objects: list[CalDAVObject] = []
    for item in report.objects:
        if item.deleted or item.calendar_data is not None:
            objects.append(item)
            continue
        try:
            calendar_data = client.fetch_object(item.href)
            stats.fetched += 1
            objects.append(
                CalDAVObject(
                    href=item.href,
                    etag=item.etag,
                    calendar_data=calendar_data,
                )
            )
        except CalDAVNotFound:
            objects.append(
                CalDAVObject(
                    href=item.href,
                    etag=item.etag,
                    deleted=True,
                )
            )
            stats.errors.append(
                "CalDAV object disappeared during sync and was treated as deleted: "
                f"{item.href}"
            )
    return CalDAVReportResult(
        objects=objects,
        sync_token=report.sync_token,
        ctag=report.ctag,
    )


def _prepare_remote_sync(
    session: Session,
    *,
    tenant_id: str,
    source_id: str,
    expected_kinds: set[str],
    password: str | None = None,
    bearer_token: str | None = None,
) -> tuple[_RemoteSyncSnapshot, dict[str, str]]:
    with Session(bind=session.get_bind()) as preparation_session:
        source = get_sync_source(
            preparation_session,
            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")
        get_calendar(
            preparation_session,
            tenant_id=tenant_id,
            calendar_id=source.calendar_id,
        )
        headers = source_auth_headers(
            preparation_session,
            source,
            password=password,
            bearer_token=bearer_token,
        )
        snapshot = _RemoteSyncSnapshot(
            source_id=source.id,
            tenant_id=source.tenant_id,
            calendar_id=source.calendar_id,
            source_kind=source.source_kind,
            collection_url=source.collection_url,
            auth_type=source.auth_type,
            username=source.username,
            credential_ref=source.credential_ref,
            sync_token=source.sync_token,
            ctag=source.ctag,
            metadata=dict(source.metadata_ or {}),
        )
    return snapshot, headers


def _lock_remote_sync_source(
    session: Session,
    snapshot: _RemoteSyncSnapshot,
) -> CalendarSyncSource:
    source = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.id == snapshot.source_id,
            CalendarSyncSource.tenant_id == snapshot.tenant_id,
            CalendarSyncSource.source_kind == snapshot.source_kind,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )
    current = (
        source.calendar_id,
        source.collection_url,
        source.auth_type,
        source.username,
        source.credential_ref,
        source.sync_token,
        source.ctag,
        dict(source.metadata_ or {}),
    )
    expected = (
        snapshot.calendar_id,
        snapshot.collection_url,
        snapshot.auth_type,
        snapshot.username,
        snapshot.credential_ref,
        snapshot.sync_token,
        snapshot.ctag,
        snapshot.metadata,
    )
    if current != expected:
        raise CalendarError(
            f"{sync_source_label(snapshot.source_kind)} source changed while "
            "the remote response was being fetched; retry the sync."
        )
    return source


def _record_remote_sync_error(
    session: Session,
    *,
    snapshot: _RemoteSyncSnapshot,
    error: Exception,
) -> None:
    try:
        source = _lock_remote_sync_source(session, snapshot)
    except Exception:
        return
    source.last_attempt_at = utcnow()
    _finalize_sync_error(session, source=source, error=error)


def sync_ics_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source_id: str,
    password: str | None = None,
    bearer_token: str | None = None,
    force_full: bool = False,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
    session.commit()
    snapshot, auth_headers = _prepare_remote_sync(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        expected_kinds={"ics", "webcal"},
        password=password,
        bearer_token=bearer_token,
    )
    stats = CalendarCalDavSyncStats(full_sync=True)
    headers = {"Accept": "text/calendar, application/calendar+json;q=0.5, */*;q=0.1", "User-Agent": "govoplan-calendar-sync/0.1"}
    headers.update(auth_headers)
    if snapshot.ctag and not force_full:
        headers["If-None-Match"] = snapshot.ctag
    last_modified = snapshot.metadata.get("last_modified")
    if isinstance(last_modified, str) and last_modified and not force_full:
        headers["If-Modified-Since"] = last_modified
    try:
        status_code, response_headers, body = http_request(
            snapshot.collection_url,
            headers=headers,
        )
        source = _lock_remote_sync_source(session, snapshot)
        source.last_attempt_at = utcnow()
        if status_code == 304:
            _finalize_sync_success(session, source=source, stats=stats)
            return source, stats
        stats.fetched = 1
        stats.created, stats.updated, stats.unchanged, stats.deleted = import_ics_subscription(
            session,
            source=source,
            ics=body,
            etag=response_headers.get("etag"),
            user_id=user_id,
        )
        metadata = dict(source.metadata_ or {})
        if response_headers.get("last-modified"):
            metadata["last_modified"] = response_headers["last-modified"]
        source.metadata_ = metadata
        source.ctag = response_headers.get("etag") or source.ctag
        _finalize_sync_success(session, source=source, stats=stats, mark_calendar=mark_calendar_sync_source)
        return source, stats
    except Exception as exc:
        session.rollback()
        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
        session.commit()
        raise


def import_ics_subscription(
    session: Session,
    *,
    source: CalendarSyncSource,
    ics: str,
    etag: str | None,
    user_id: str | None,
) -> tuple[int, int, int, int]:
    parsed_events = parse_vevents(ics)
    remaining = {
        (event.uid, event.recurrence_id): event
        for event in session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == source.source_kind,
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    }
    created = updated = unchanged = deleted = 0
    changes: list[tuple[CalendarEvent, str, dict[str, Any] | None]] = []
    raw_ics = str(parsed_events[0].get("raw_ics") or ics)
    for parsed in parsed_events:
        key = (parsed["uid"], parsed.get("recurrence_id"))
        href = subscription_event_href(source, parsed)
        event = remaining.get(key) or find_event_for_subscription_component(session, source=source, parsed=parsed)
        previous = calendar_event_change_payload(event, prefix="previous_") if event else None
        was_deleted = bool(event and event.deleted_at is not None)
        if event is None:
            event = CalendarEvent(
                tenant_id=source.tenant_id,
                calendar_id=source.calendar_id,
                uid=parsed["uid"],
                recurrence_id=parsed.get("recurrence_id"),
                created_by_user_id=user_id,
            )
            session.add(event)
            created += 1
            changes.append((event, "created", previous))
        else:
            if was_deleted:
                created += 1
                changes.append((event, "created", previous))
            elif event.etag == etag and event.raw_ics == raw_ics:
                unchanged += 1
            else:
                updated += 1
                changes.append((event, "updated", previous))
        apply_parsed_event_to_model(event, parsed, source=source, href=href, etag=etag, raw_ics=raw_ics, user_id=user_id)
        remaining.pop(key, None)
    deleted_at = utcnow()
    for event in remaining.values():
        previous = calendar_event_change_payload(event, prefix="previous_")
        event.deleted_at = deleted_at
        deleted += 1
        changes.append((event, "deleted", previous))
    session.flush()
    for event, operation, previous in changes:
        record_calendar_event_change(session, event=event, operation=operation, user_id=user_id, previous=previous)
    return created, updated, unchanged, deleted


def find_event_for_subscription_component(session: Session, *, source: CalendarSyncSource, parsed: dict[str, Any]) -> CalendarEvent | None:
    return (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.uid == parsed["uid"],
            CalendarEvent.recurrence_id == parsed.get("recurrence_id"),
            CalendarEvent.source_kind == source.source_kind,
        )
        .order_by(CalendarEvent.deleted_at.is_(None).desc())
        .first()
    )


def subscription_event_href(source: CalendarSyncSource, parsed: dict[str, Any]) -> str:
    uid = urllib.parse.quote(str(parsed["uid"]), safe="-_.~@")
    recurrence_id = parsed.get("recurrence_id")
    if recurrence_id:
        return f"{source.collection_url}#{uid}/{urllib.parse.quote(str(recurrence_id), safe='-_.~@:')}"
    return f"{source.collection_url}#{uid}"


def sync_graph_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source_id: str,
    bearer_token: str | None = None,
    force_full: bool = False,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
    session.commit()
    snapshot, auth_headers = _prepare_remote_sync(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        expected_kinds={"graph"},
        bearer_token=bearer_token,
    )
    headers = {"Accept": "application/json", "User-Agent": "govoplan-calendar-graph/0.1", "Prefer": 'outlook.timezone="UTC"'}
    headers.update(auth_headers)
    stats = CalendarCalDavSyncStats(
        full_sync=force_full or not bool(snapshot.sync_token),
        used_sync_token=bool(snapshot.sync_token and not force_full),
    )
    try:
        next_url = (
            snapshot.sync_token
            if snapshot.sync_token and not force_full
            else snapshot.collection_url
        )
        provider_items: list[dict[str, Any]] = []
        delta_link: str | None = None
        for _page in range(50):
            next_url = same_origin_http_url(
                snapshot.collection_url,
                next_url,
                label="Microsoft Graph continuation URL",
            )
            _status, _headers, body = http_request(
                next_url,
                headers=headers,
                credential_origin=snapshot.collection_url,
            )
            payload = json.loads(body or "{}")
            for item in payload.get("value", []):
                if isinstance(item, dict):
                    provider_items.append(item)
                if len(provider_items) > REMOTE_SYNC_MAX_ITEMS:
                    raise CalendarError(
                        "Microsoft Graph sync exceeded the configured item limit."
                    )
            next_link = payload.get("@odata.nextLink")
            raw_delta_link = payload.get("@odata.deltaLink")
            if raw_delta_link:
                delta_link = same_origin_http_url(
                    snapshot.collection_url,
                    str(raw_delta_link),
                    label="Microsoft Graph delta URL",
                )
            if not next_link:
                break
            next_url = same_origin_http_url(
                snapshot.collection_url,
                str(next_link),
                label="Microsoft Graph continuation URL",
            )
        else:
            raise CalendarError("Microsoft Graph sync returned too many pages")
        source = _lock_remote_sync_source(session, snapshot)
        source.last_attempt_at = utcnow()
        seen_hrefs: set[str] = set()
        stats.fetched = len(provider_items)
        for item in provider_items:
            href = str(item.get("id") or "")
            if not href:
                continue
            if item.get("@removed"):
                stats.deleted += soft_delete_source_href(
                    session,
                    source=source,
                    href=href,
                )
                continue
            seen_hrefs.add(href)
            created, updated, unchanged = import_provider_event(
                session,
                source=source,
                href=href,
                provider_event=graph_event_payload(item),
                user_id=user_id,
            )
            stats.created += created
            stats.updated += updated
            stats.unchanged += unchanged
        if stats.full_sync:
            stats.deleted += soft_delete_unseen_source_events(session, source=source, seen_hrefs=seen_hrefs)
        source.sync_token = delta_link or source.sync_token
        source.last_synced_at = utcnow()
        source.last_status = "ok"
        source.last_error = None
        schedule_next_caldav_sync(source)
        mark_calendar_sync_source(source.calendar, source)
        session.flush()
        return source, stats
    except Exception as exc:
        session.rollback()
        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
        session.commit()
        raise


def sync_ews_source(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    source_id: str,
    password: str | None = None,
    bearer_token: str | None = None,
    force_full: bool = False,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
    session.commit()
    snapshot, auth_headers = _prepare_remote_sync(
        session,
        tenant_id=tenant_id,
        source_id=source_id,
        expected_kinds={"ews"},
        password=password,
        bearer_token=bearer_token,
    )
    metadata = snapshot.metadata
    window_before = int(metadata.get("sync_window_days_before", 365)) if isinstance(metadata, dict) else 365
    window_after = int(metadata.get("sync_window_days_after", 730)) if isinstance(metadata, dict) else 730
    now = utcnow()
    start = now - timedelta(days=max(window_before, 0))
    end = now + timedelta(days=max(window_after, 1))
    headers = {
        "Accept": "text/xml",
        "Content-Type": "text/xml; charset=utf-8",
        "User-Agent": "govoplan-calendar-ews/0.1",
    }
    headers.update(auth_headers)
    stats = CalendarCalDavSyncStats(full_sync=True)
    try:
        _status, _headers, body = http_request(
            snapshot.collection_url,
            method="POST",
            headers=headers,
            body=ews_find_item_body(
                start=start,
                end=end,
                mailbox=metadata.get("mailbox"),
            ),
        )
        try:
            items = parse_ews_calendar_items(body)
        except EwsAdapterError as exc:
            raise CalendarError(str(exc)) from exc
        if len(items) > REMOTE_SYNC_MAX_ITEMS:
            raise CalendarError(
                "Exchange Web Services sync exceeded the configured item limit."
            )
        source = _lock_remote_sync_source(session, snapshot)
        source.last_attempt_at = utcnow()
        seen_hrefs: set[str] = set()
        for item in items:
            href = item["href"]
            seen_hrefs.add(href)
            created, updated, unchanged = import_provider_event(session, source=source, href=href, provider_event=item, user_id=user_id)
            stats.created += created
            stats.updated += updated
            stats.unchanged += unchanged
        stats.fetched = len(items)
        stats.deleted += soft_delete_unseen_source_events(session, source=source, seen_hrefs=seen_hrefs)
        source.sync_token = f"{response_datetime(start).isoformat()}..{response_datetime(end).isoformat()}"
        source.last_synced_at = utcnow()
        source.last_status = "ok"
        source.last_error = None
        schedule_next_caldav_sync(source)
        mark_calendar_sync_source(source.calendar, source)
        session.flush()
        return source, stats
    except Exception as exc:
        session.rollback()
        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
        session.commit()
        raise


def import_provider_event(
    session: Session,
    *,
    source: CalendarSyncSource,
    href: str,
    provider_event: dict[str, Any],
    user_id: str | None,
) -> tuple[int, int, int]:
    event = find_event_for_source_href(session, source=source, href=href)
    previous = calendar_event_change_payload(event, prefix="previous_") if event else None
    was_deleted = bool(event and event.deleted_at is not None)
    created = updated = unchanged = 0
    raw_fingerprint = json.dumps(provider_event.get("metadata") or provider_event.get("icalendar") or provider_event, sort_keys=True, default=str)
    if event is None:
        event = CalendarEvent(
            tenant_id=source.tenant_id,
            calendar_id=source.calendar_id,
            uid=str(provider_event["uid"]),
            recurrence_id=provider_event.get("recurrence_id"),
            created_by_user_id=user_id,
        )
        session.add(event)
        created = 1
    elif was_deleted:
        created = 1
    elif event.etag == provider_event.get("etag") and event.raw_ics == raw_fingerprint:
        unchanged = 1
    else:
        updated = 1
    for field_name in (
        "sequence",
        "summary",
        "description",
        "location",
        "status",
        "transparency",
        "classification",
        "start_at",
        "end_at",
        "duration_seconds",
        "all_day",
        "timezone",
        "organizer",
        "attendees",
        "categories",
        "rrule",
        "rdate",
        "exdate",
        "reminders",
        "attachments",
        "related_to",
        "icalendar",
    ):
        if field_name in provider_event:
            setattr(event, field_name, provider_event[field_name])
    event.calendar_id = source.calendar_id
    event.source_kind = source.source_kind
    event.source_href = href
    event.etag = provider_event.get("etag")
    event.raw_ics = raw_fingerprint
    event.updated_by_user_id = user_id
    event.deleted_at = None
    metadata = dict(event.metadata_ or {})
    metadata[source.source_kind] = {"source_id": source.id, "collection_url": source.collection_url, "href": href, "etag": event.etag}
    if isinstance(provider_event.get("metadata"), dict):
        metadata[source.source_kind].update(provider_event["metadata"])
    event.metadata_ = metadata
    validate_event_time(event)
    session.flush()
    if created or updated:
        record_calendar_event_change(session, event=event, operation="created" if created else "updated", user_id=user_id, previous=previous)
    return created, updated, unchanged


def find_event_for_source_href(session: Session, *, source: CalendarSyncSource, href: str) -> CalendarEvent | None:
    return (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == source.source_kind,
            CalendarEvent.source_href == href,
        )
        .order_by(CalendarEvent.deleted_at.is_(None).desc())
        .first()
    )


def soft_delete_source_href(session: Session, *, source: CalendarSyncSource, href: str) -> int:
    deleted_at = utcnow()
    count = 0
    for event in (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == source.source_kind,
            CalendarEvent.source_href == href,
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    ):
        previous = calendar_event_change_payload(event, prefix="previous_")
        event.deleted_at = deleted_at
        record_calendar_event_change(session, event=event, operation="deleted", user_id=None, previous=previous)
        count += 1
    return count


def soft_delete_unseen_source_events(
    session: Session,
    *,
    source: CalendarSyncSource,
    seen_hrefs: set[str],
    source_kind: str | None = None,
    href_prefix: str | None = None,
    user_id: str | None = None,
    batch_size: int = SOURCE_EVENT_CLEANUP_BATCH_SIZE,
) -> int:
    deleted_at = utcnow()
    count = 0
    last_id: str | None = None
    effective_source_kind = source_kind or source.source_kind
    effective_batch_size = max(1, batch_size)
    while True:
        id_query = session.query(CalendarEvent.id).filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == effective_source_kind,
            CalendarEvent.deleted_at.is_(None),
        )
        if href_prefix:
            id_query = id_query.filter(CalendarEvent.source_href.like(f"{href_prefix}%"))
        if last_id:
            id_query = id_query.filter(CalendarEvent.id > last_id)
        ids = [row[0] for row in id_query.order_by(CalendarEvent.id.asc()).limit(effective_batch_size).all()]
        if not ids:
            break
        last_id = ids[-1]
        events = (
            session.query(CalendarEvent)
            .filter(CalendarEvent.id.in_(ids))
            .order_by(CalendarEvent.id.asc())
            .all()
        )
        for event in events:
            if effective_source_kind == "caldav" and event.source_href:
                from govoplan_calendar.backend.outbox import calendar_outbox_has_active_desired_state

                if calendar_outbox_has_active_desired_state(
                    session,
                    source_id=source.id,
                    href=event.source_href,
                ):
                    continue
            if event.source_href not in seen_hrefs:
                previous = calendar_event_change_payload(event, prefix="previous_")
                event.deleted_at = deleted_at
                record_calendar_event_change(session, event=event, operation="deleted", user_id=user_id, previous=previous)
                count += 1
        session.flush()
    return count


def sync_due_sources(
    session: Session,
    *,
    tenant_id: str | None = None,
    user_id: str | None = None,
    limit: int = 50,
    now: datetime | None = None,
    client_factory: Callable[[CalendarSyncSource], CalDAVClient] | None = None,
) -> list[CalendarCalDavDueSyncResult]:
    return _sync_due_sources(session, source_kinds=SYNC_SOURCE_KINDS, **_due_sync_options(tenant_id, user_id, limit, now, client_factory))


def sync_due_caldav_sources(
    session: Session,
    *,
    tenant_id: str | None = None,
    user_id: str | None = None,
    limit: int = 50,
    now: datetime | None = None,
    client_factory: Callable[[CalendarSyncSource], CalDAVClient] | None = None,
) -> list[CalendarCalDavDueSyncResult]:
    return _sync_due_sources(session, source_kinds={"caldav"}, **_due_sync_options(tenant_id, user_id, limit, now, client_factory))


def _due_sync_options(
    tenant_id: str | None,
    user_id: str | None,
    limit: int,
    now: datetime | None,
    client_factory: Callable[[CalendarSyncSource], CalDAVClient] | None,
) -> dict[str, object]:
    return {
        "tenant_id": tenant_id,
        "user_id": user_id,
        "limit": limit,
        "now": now,
        "client_factory": client_factory,
    }


def _sync_due_sources(
    session: Session,
    *,
    tenant_id: str | None,
    user_id: str | None,
    limit: int,
    now: datetime | None,
    source_kinds: set[str],
    client_factory: Callable[[CalendarSyncSource], CalDAVClient] | None,
) -> list[CalendarCalDavDueSyncResult]:
    due_at = normalize_datetime(now or utcnow())
    query = session.query(CalendarSyncSource).filter(
        CalendarSyncSource.source_kind.in_(source_kinds),
        CalendarSyncSource.sync_enabled.is_(True),
        CalendarSyncSource.deleted_at.is_(None),
        or_(CalendarSyncSource.next_sync_at.is_(None), CalendarSyncSource.next_sync_at <= due_at),
    )
    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(),
            CalendarSyncSource.created_at.asc(),
        )
        .with_for_update(skip_locked=True)
        .limit(max(1, min(int(limit), 200)))
        .all()
    )
    # next_sync_at doubles as a recoverable lease. A worker that dies during
    # remote I/O leaves the source eligible again after this timeout.
    lease_until = due_at + timedelta(minutes=10)
    for source in sources:
        source.next_sync_at = lease_until
        source.last_attempt_at = due_at
    session.commit()

    results: list[CalendarCalDavDueSyncResult] = []
    for source in sources:
        previous_status = source.last_status
        try:
            refreshed, stats = _sync_due_source(session, source=source, user_id=user_id, client_factory=client_factory)
            results.append(CalendarCalDavDueSyncResult(source_id=refreshed.id, calendar_id=refreshed.calendar_id, status="ok", stats=stats))
            _emit_calendar_sync_notification(session, source=refreshed, status="ok", previous_status=previous_status, stats=stats)
            session.commit()
        except Exception as exc:
            session.rollback()
            refreshed = get_sync_source(
                session,
                tenant_id=source.tenant_id,
                source_id=source.id,
            )
            results.append(CalendarCalDavDueSyncResult(source_id=source.id, calendar_id=source.calendar_id, status="error", error=str(exc)))
            _emit_calendar_sync_notification(session, source=refreshed, status="error", previous_status=previous_status, error=str(exc))
            session.commit()
    return results


def _sync_due_source(
    session: Session,
    *,
    source: CalendarSyncSource,
    user_id: str | None,
    client_factory: Callable[[CalendarSyncSource], CalDAVClient] | None,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
    if source.source_kind == "caldav":
        client = client_factory(source) if client_factory else None
        return sync_caldav_source(session, tenant_id=source.tenant_id, user_id=user_id, source_id=source.id, client=client)
    return sync_source(session, tenant_id=source.tenant_id, user_id=user_id, source_id=source.id)


def schedule_next_caldav_sync(source: CalendarSyncSource, *, now: datetime | None = None) -> None:
    if not source.sync_enabled:
        source.next_sync_at = None
        return
    interval = max(int(source.sync_interval_seconds or CALDAV_DEFAULT_SYNC_INTERVAL_SECONDS), 60)
    source.next_sync_at = normalize_datetime(now or utcnow()) + timedelta(seconds=interval)


def apply_caldav_report(
    session: Session,
    *,
    source: CalendarSyncSource,
    client: CalDAVClient | None = None,
    report: CalDAVReportResult,
    stats: CalendarCalDavSyncStats,
    user_id: str | None,
) -> None:
    seen_hrefs: set[str] = set()
    for item in report.objects:
        href = normalize_caldav_href(source.collection_url, item.href)
        from govoplan_calendar.backend.outbox import calendar_outbox_has_active_desired_state

        if calendar_outbox_has_active_desired_state(session, source_id=source.id, href=href):
            # Do not let an older remote representation erase a committed
            # local desired state while its outbound operation is pending.
            seen_hrefs.add(href)
            stats.unchanged += 1
            continue
        if item.deleted:
            stats.deleted += soft_delete_caldav_href(session, source=source, href=href)
            continue
        seen_hrefs.add(href)
        calendar_data = item.calendar_data
        if calendar_data is None:
            raise CalendarError(
                "CalDAV reports must be hydrated before database application."
            )
        created, updated, unchanged, deleted = import_caldav_resource(
            session,
            source=source,
            href=href,
            etag=item.etag,
            ics=calendar_data,
            user_id=user_id,
        )
        stats.created += created
        stats.updated += updated
        stats.unchanged += unchanged
        stats.deleted += deleted

    if stats.full_sync:
        stats.deleted += soft_delete_unseen_source_events(
            session,
            source=source,
            seen_hrefs=seen_hrefs,
            source_kind="caldav",
            href_prefix=source.collection_url,
            user_id=user_id,
        )


def import_caldav_resource(
    session: Session,
    *,
    source: CalendarSyncSource,
    href: str,
    etag: str | None,
    ics: str,
    user_id: str | None,
) -> tuple[int, int, int, int]:
    parsed_events = parse_vevents(ics)
    remaining = {
        (event.uid, event.recurrence_id): event
        for event in session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == "caldav",
            CalendarEvent.source_href == href,
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    }
    created = updated = unchanged = deleted = 0
    changes: list[tuple[CalendarEvent, str, dict[str, Any] | None]] = []
    raw_ics = str(parsed_events[0].get("raw_ics") or ics)
    for parsed in parsed_events:
        key = (parsed["uid"], parsed.get("recurrence_id"))
        event = find_event_for_caldav_component(session, source=source, href=href, parsed=parsed)
        previous = calendar_event_change_payload(event, prefix="previous_") if event else None
        was_deleted = bool(event and event.deleted_at is not None)
        if event is None:
            event = CalendarEvent(
                tenant_id=source.tenant_id,
                calendar_id=source.calendar_id,
                uid=parsed["uid"],
                recurrence_id=parsed.get("recurrence_id"),
                created_by_user_id=user_id,
            )
            session.add(event)
            created += 1
            changes.append((event, "created", previous))
        else:
            if was_deleted:
                created += 1
                changes.append((event, "created", previous))
            elif event.etag == etag and event.raw_ics == raw_ics:
                unchanged += 1
            else:
                updated += 1
                changes.append((event, "updated", previous))
        apply_parsed_event_to_model(event, parsed, source=source, href=href, etag=etag, raw_ics=raw_ics, user_id=user_id)
        remaining.pop(key, None)
    deleted_at = utcnow()
    for event in remaining.values():
        previous = calendar_event_change_payload(event, prefix="previous_")
        event.deleted_at = deleted_at
        deleted += 1
        changes.append((event, "deleted", previous))
    session.flush()
    for event, operation, previous in changes:
        record_calendar_event_change(session, event=event, operation=operation, user_id=user_id, previous=previous)
    return created, updated, unchanged, deleted


def find_event_for_caldav_component(session: Session, *, source: CalendarSyncSource, href: str, parsed: dict[str, Any]) -> CalendarEvent | None:
    return (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.uid == parsed["uid"],
            CalendarEvent.recurrence_id == parsed.get("recurrence_id"),
        )
        .order_by((CalendarEvent.source_href == href).desc(), CalendarEvent.deleted_at.is_(None).desc())
        .first()
    )


def apply_parsed_event_to_model(
    event: CalendarEvent,
    parsed: dict[str, Any],
    *,
    source: CalendarSyncSource,
    href: str,
    etag: str | None,
    raw_ics: str,
    user_id: str | None,
) -> None:
    for field_name in (
        "sequence",
        "summary",
        "description",
        "location",
        "status",
        "transparency",
        "classification",
        "start_at",
        "end_at",
        "duration_seconds",
        "all_day",
        "timezone",
        "organizer",
        "attendees",
        "categories",
        "rrule",
        "rdate",
        "exdate",
        "reminders",
        "attachments",
        "related_to",
        "icalendar",
    ):
        setattr(event, field_name, parsed.get(field_name))
    event.calendar_id = source.calendar_id
    event.source_kind = source.source_kind
    event.source_href = href
    event.etag = etag
    event.raw_ics = raw_ics
    event.updated_by_user_id = user_id
    event.deleted_at = None
    metadata = dict(event.metadata_ or {})
    metadata[source.source_kind] = {"source_id": source.id, "collection_url": source.collection_url, "href": href, "etag": etag}
    event.metadata_ = metadata
    validate_event_time(event)


def soft_delete_caldav_href(session: Session, *, source: CalendarSyncSource, href: str) -> int:
    deleted_at = utcnow()
    count = 0
    for event in (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == "caldav",
            CalendarEvent.source_href == href,
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    ):
        previous = calendar_event_change_payload(event, prefix="previous_")
        event.deleted_at = deleted_at
        record_calendar_event_change(session, event=event, operation="deleted", user_id=None, previous=previous)
        count += 1
    return count


def normalize_caldav_href(collection_url: str, href: str) -> str:
    collection = ensure_collection_url(collection_url)
    candidate = urllib.parse.urljoin(collection, href)
    collection_parts = urllib.parse.urlparse(collection)
    candidate_parts = urllib.parse.urlparse(candidate)
    if (
        candidate_parts.username
        or candidate_parts.password
        or (
            candidate_parts.scheme.lower(),
            candidate_parts.hostname,
            candidate_parts.port,
        )
        != (
            collection_parts.scheme.lower(),
            collection_parts.hostname,
            collection_parts.port,
        )
    ):
        raise CalendarError("CalDAV resource href must use the configured collection origin")
    collection_path = posixpath.normpath(urllib.parse.unquote(collection_parts.path))
    candidate_path = posixpath.normpath(urllib.parse.unquote(candidate_parts.path))
    collection_prefix = collection_path.rstrip("/") + "/"
    if not candidate_path.startswith(collection_prefix) or candidate_path == collection_path:
        raise CalendarError("CalDAV resource href must remain inside the configured collection path")
    if candidate_parts.query or candidate_parts.fragment:
        raise CalendarError("CalDAV resource href must not contain a query or fragment")
    return urllib.parse.urlunparse(candidate_parts)


def mark_calendar_caldav(calendar: CalendarCollection, source: CalendarSyncSource) -> None:
    mark_calendar_sync_source(calendar, source)


def mark_calendar_sync_source(calendar: CalendarCollection, source: CalendarSyncSource) -> None:
    metadata = dict(calendar.metadata_ or {})
    metadata.setdefault("source_kind", source.source_kind)
    metadata["connector_kind"] = source.source_kind
    metadata["sync_source_id"] = source.id
    metadata["sync_href"] = source.collection_url
    metadata["sync_direction"] = source.sync_direction
    source_metadata = source.metadata_ if isinstance(source.metadata_, dict) else {}
    if source_metadata.get("integration_profile") == "open_xchange":
        for key in (
            "integration_profile",
            "transport_kind",
            "connector_profile_ref",
            "identity_mapping_ref",
            "resource_calendar_ref",
        ):
            if source_metadata.get(key):
                metadata[key] = source_metadata[key]
            else:
                metadata.pop(key, None)
        resource_ref = source_metadata.get("resource_calendar_ref")
        if resource_ref:
            calendar.owner_type = "resource"
            calendar.owner_id = str(resource_ref)
        elif calendar.owner_type == "resource":
            calendar.owner_type = "tenant"
            calendar.owner_id = None
    calendar.metadata_ = metadata


def caldav_source_response(source: CalendarSyncSource) -> dict[str, Any]:
    has_server_owned_credential = bool(
        source.credential_ref
        and (
            source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX)
            or source.credential_ref.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX)
        )
    )
    return {
        "id": source.id,
        "tenant_id": source.tenant_id,
        "calendar_id": source.calendar_id,
        "source_kind": source.source_kind,
        "collection_url": source.collection_url,
        "display_name": source.display_name,
        "auth_type": source.auth_type,
        "username": source.username,
        "credential_ref": None,
        "credential_envelope_id": _core_credential_id(source.credential_ref),
        "has_credential": has_server_owned_credential,
        "sync_enabled": bool(source.sync_enabled),
        "sync_interval_seconds": int(source.sync_interval_seconds or CALDAV_DEFAULT_SYNC_INTERVAL_SECONDS),
        "sync_direction": source.sync_direction,
        "conflict_policy": source.conflict_policy,
        "sync_token": source.sync_token,
        "ctag": source.ctag,
        "last_attempt_at": response_datetime(source.last_attempt_at),
        "last_synced_at": response_datetime(source.last_synced_at),
        "next_sync_at": response_datetime(source.next_sync_at),
        "last_status": source.last_status,
        "last_error": source.last_error,
        "created_at": response_datetime(source.created_at),
        "updated_at": response_datetime(source.updated_at),
        "metadata": source.metadata_ or {},
    }


def caldav_sync_response(source: CalendarSyncSource, stats: CalendarCalDavSyncStats) -> dict[str, Any]:
    return {
        "source": caldav_source_response(source),
        "created": stats.created,
        "updated": stats.updated,
        "deleted": stats.deleted,
        "unchanged": stats.unchanged,
        "fetched": stats.fetched,
        "full_sync": stats.full_sync,
        "used_sync_token": stats.used_sync_token,
        "sync_token": stats.sync_token,
        "ctag": stats.ctag,
        "errors": stats.errors,
    }


def list_events(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str | None = None,
    start_at: datetime | None = None,
    end_at: datetime | None = None,
) -> list[CalendarEvent]:
    query = session.query(CalendarEvent).filter(CalendarEvent.tenant_id == tenant_id, CalendarEvent.deleted_at.is_(None))
    if calendar_id:
        query = query.filter(CalendarEvent.calendar_id == calendar_id)
    if start_at is not None:
        query = query.filter(or_(CalendarEvent.end_at.is_(None), CalendarEvent.end_at >= normalize_datetime(start_at)))
    if end_at is not None:
        query = query.filter(CalendarEvent.start_at <= normalize_datetime(end_at))
    return query.order_by(CalendarEvent.start_at.asc(), CalendarEvent.summary.asc()).all()


def list_event_occurrences(
    session: Session,
    *,
    tenant_id: str,
    start_at: datetime,
    end_at: datetime,
    calendar_id: str | None = None,
) -> list[dict[str, Any]]:
    """Return range-bounded events with recurring series fully reconciled."""

    range_start = normalize_datetime(start_at)
    range_end = normalize_datetime(end_at)
    if range_end <= range_start:
        raise CalendarError("Event range end must be after start")
    if range_end - range_start > timedelta(days=MAX_OCCURRENCE_RANGE_DAYS):
        raise CalendarError(
            f"Expanded event ranges are limited to {MAX_OCCURRENCE_RANGE_DAYS} days"
        )

    base_query = session.query(CalendarEvent).filter(
        CalendarEvent.tenant_id == tenant_id,
        CalendarEvent.deleted_at.is_(None),
    )
    if calendar_id:
        base_query = base_query.filter(CalendarEvent.calendar_id == calendar_id)

    recurring_masters = (
        base_query.filter(
            CalendarEvent.recurrence_id.is_(None),
            or_(
                CalendarEvent.rrule.is_not(None),
                func.json_array_length(CalendarEvent.rdate) > 0,
            ),
        )
        .order_by(CalendarEvent.start_at.asc(), CalendarEvent.id.asc())
        .all()
    )
    direct_events = (
        base_query.filter(
            or_(
                CalendarEvent.end_at.is_(None),
                CalendarEvent.end_at >= range_start,
            ),
            CalendarEvent.start_at <= range_end,
        )
        .order_by(CalendarEvent.start_at.asc(), CalendarEvent.id.asc())
        .all()
    )

    series_keys = {
        (event.calendar_id, event.uid) for event in recurring_masters
    }
    overrides: list[CalendarEvent] = []
    if series_keys:
        series_uids = {uid for _calendar_id, uid in series_keys}
        overrides = [
            event
            for event in base_query.filter(
                CalendarEvent.recurrence_id.is_not(None),
                CalendarEvent.uid.in_(series_uids),
            ).all()
            if (event.calendar_id, event.uid) in series_keys
        ]

    overrides_by_series: dict[
        tuple[str, str], dict[str, CalendarEvent]
    ] = {}
    for override in overrides:
        recurrence_key = normalized_recurrence_id(override.recurrence_id)
        if recurrence_key:
            overrides_by_series.setdefault(
                (override.calendar_id, override.uid), {}
            )[recurrence_key] = override

    results: list[dict[str, Any]] = []
    consumed_event_ids: set[str] = set()
    recurring_master_ids = {event.id for event in recurring_masters}
    recurring_master_by_series = {
        (event.calendar_id, event.uid): event for event in recurring_masters
    }
    for master in recurring_masters:
        series_key = (master.calendar_id, master.uid)
        series_overrides = overrides_by_series.get(series_key, {})
        for occurrence in expand_event_occurrences(
            master,
            range_start,
            range_end,
        ):
            occurrence_recurrence_id = str(
                occurrence.get("recurrence_id") or ""
            )
            recurrence_key = normalized_recurrence_id(
                occurrence_recurrence_id
            )
            override = (
                series_overrides.get(recurrence_key)
                if recurrence_key
                else None
            )
            if override is not None:
                consumed_event_ids.add(override.id)
                if override.status.upper() == "CANCELLED":
                    continue
                if event_overlaps_range(
                    override,
                    range_start=range_start,
                    range_end=range_end,
                ):
                    results.append(
                        expanded_event_response(
                            override,
                            series_event_id=master.id,
                            recurrence_id=(
                                override.recurrence_id
                                or occurrence_recurrence_id
                            ),
                            is_override=True,
                        )
                    )
                continue
            results.append(
                expanded_event_response(
                    master,
                    series_event_id=master.id,
                    recurrence_id=occurrence_recurrence_id,
                    occurrence=occurrence,
                )
            )

    for event in direct_events:
        if event.id in recurring_master_ids or event.id in consumed_event_ids:
            continue
        series_event_id = None
        is_override = False
        if event.recurrence_id:
            master = recurring_master_by_series.get(
                (event.calendar_id, event.uid)
            )
            if master is not None:
                series_event_id = master.id
                is_override = True
            if event.status.upper() == "CANCELLED":
                continue
        results.append(
            expanded_event_response(
                event,
                series_event_id=series_event_id,
                recurrence_id=normalized_recurrence_id(event.recurrence_id),
                is_override=is_override,
            )
        )
    return sorted(
        results,
        key=lambda item: (
            item["start_at"],
            item["calendar_id"],
            item["summary"],
            item["instance_id"],
        ),
    )


def event_overlaps_range(
    event: CalendarEvent,
    *,
    range_start: datetime,
    range_end: datetime,
) -> bool:
    event_start = normalize_datetime(event.start_at)
    event_end = (
        normalize_datetime(event.end_at)
        if event.end_at is not None
        else event_start
    )
    return event_end >= range_start and event_start <= range_end


def expanded_event_response(
    event: CalendarEvent,
    *,
    series_event_id: str | None,
    recurrence_id: str | None,
    occurrence: dict[str, Any] | None = None,
    is_override: bool = False,
) -> dict[str, Any]:
    payload = event_response(event)
    if occurrence is not None:
        payload["start_at"] = response_datetime(occurrence["start_at"])
        payload["end_at"] = response_datetime(occurrence.get("end_at"))
        payload["all_day"] = bool(occurrence.get("all_day"))
    payload["recurrence_id"] = recurrence_id
    payload["series_event_id"] = series_event_id
    payload["is_occurrence"] = series_event_id is not None
    payload["is_override"] = is_override
    payload["instance_id"] = (
        f"{series_event_id}:{recurrence_id}"
        if series_event_id and recurrence_id
        else event.id
    )
    return payload


def list_freebusy(
    session: Session,
    *,
    tenant_id: str,
    start_at: datetime,
    end_at: datetime,
    calendar_ids: list[str] | None = None,
) -> list[dict[str, Any]]:
    range_start = normalize_datetime(start_at)
    range_end = normalize_datetime(end_at)
    if range_end < range_start:
        raise CalendarError("Free/busy end must be after start")
    events: list[dict[str, Any]] = []
    if calendar_ids:
        for calendar_id in dict.fromkeys(calendar_ids):
            events.extend(
                list_event_occurrences(
                    session,
                    tenant_id=tenant_id,
                    calendar_id=calendar_id,
                    start_at=range_start,
                    end_at=range_end,
                )
            )
    else:
        events = list_event_occurrences(
            session,
            tenant_id=tenant_id,
            start_at=range_start,
            end_at=range_end,
        )
    busy = [
        {
            "calendar_id": event["calendar_id"],
            "event_id": event["id"],
            "uid": event["uid"],
            "recurrence_id": event["recurrence_id"],
            "start_at": event["start_at"],
            "end_at": event["end_at"],
            "all_day": event["all_day"],
            "transparency": event["transparency"],
            "status": event["status"],
        }
        for event in events
        if event["status"].upper() != "CANCELLED"
        and event["transparency"].upper() != "TRANSPARENT"
    ]
    return sorted(
        busy,
        key=lambda item: (
            item["start_at"],
            item["calendar_id"],
            item["uid"],
        ),
    )


def get_event(session: Session, *, tenant_id: str, event_id: str) -> CalendarEvent:
    event = (
        session.query(CalendarEvent)
        .filter(CalendarEvent.tenant_id == tenant_id, CalendarEvent.id == event_id, CalendarEvent.deleted_at.is_(None))
        .first()
    )
    if not event:
        raise CalendarError("Calendar event not found")
    return event


def get_calendar_view_preferences(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
) -> dict[str, Any]:
    preference = (
        session.query(CalendarViewPreference)
        .filter(
            CalendarViewPreference.tenant_id == tenant_id,
            CalendarViewPreference.user_id == user_id,
        )
        .first()
    )
    effective = dict(DEFAULT_CALENDAR_VIEW_PREFERENCES)
    overridden_fields: list[str] = []
    if preference is not None:
        for field_name in DEFAULT_CALENDAR_VIEW_PREFERENCES:
            value = getattr(preference, field_name)
            if value is not None:
                effective[field_name] = value
                overridden_fields.append(field_name)
    return {
        **effective,
        "overridden_fields": overridden_fields,
        "defaults": dict(DEFAULT_CALENDAR_VIEW_PREFERENCES),
    }


def update_calendar_view_preferences(
    session: Session,
    *,
    tenant_id: str,
    user_id: str,
    payload: CalendarViewPreferencesUpdateRequest,
) -> dict[str, Any]:
    preference = (
        session.query(CalendarViewPreference)
        .filter(
            CalendarViewPreference.tenant_id == tenant_id,
            CalendarViewPreference.user_id == user_id,
        )
        .first()
    )
    if preference is None:
        preference = CalendarViewPreference(
            tenant_id=tenant_id,
            user_id=user_id,
        )
        session.add(preference)
    for field_name in payload.model_fields_set:
        setattr(preference, field_name, getattr(payload, field_name))

    effective = dict(DEFAULT_CALENDAR_VIEW_PREFERENCES)
    for field_name in DEFAULT_CALENDAR_VIEW_PREFERENCES:
        value = getattr(preference, field_name)
        if value is not None:
            effective[field_name] = value
    if int(effective["workday_end_hour"]) <= int(
        effective["workday_start_hour"]
    ):
        raise CalendarError(
            "Workday end hour must be after workday start hour"
        )
    session.flush()
    return get_calendar_view_preferences(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
    )


def create_event(session: Session, *, tenant_id: str, user_id: str | None, payload: CalendarEventCreateRequest) -> CalendarEvent:
    default_calendar = None if payload.calendar_id else get_default_calendar(session, tenant_id=tenant_id)
    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")
    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,
        calendar_id=calendar_id,
        for_update=True,
    )
    assert_sync_mutation_allowed(source)
    supplied_sync_fields = bool(
        payload.source_href is not None
        or payload.etag is not None
        or (
            "source_kind" in payload.model_fields_set
            and payload.source_kind != "local"
        )
    )
    if source is not None and supplied_sync_fields:
        raise CalendarError(
            "source_kind, source_href, and etag are sync-owned fields on synchronized calendars"
        )
    event = CalendarEvent(
        tenant_id=tenant_id,
        calendar_id=calendar_id,
        uid=payload.uid or f"{uuid.uuid4()}@govoplan.local",
        recurrence_id=payload.recurrence_id,
        sequence=payload.sequence,
        summary=payload.summary,
        description=payload.description,
        location=payload.location,
        status=payload.status.upper(),
        transparency=payload.transparency.upper(),
        classification=payload.classification.upper(),
        start_at=normalize_datetime(payload.start_at),
        end_at=normalize_datetime(payload.end_at) if payload.end_at else None,
        duration_seconds=payload.duration_seconds,
        all_day=payload.all_day,
        timezone=payload.timezone,
        organizer=payload.organizer,
        attendees=payload.attendees,
        categories=payload.categories,
        rrule=payload.rrule,
        rdate=payload.rdate,
        exdate=payload.exdate,
        reminders=payload.reminders,
        attachments=payload.attachments,
        related_to=payload.related_to,
        source_kind=payload.source_kind,
        source_href=payload.source_href,
        etag=payload.etag,
        icalendar=payload.icalendar,
        created_by_user_id=user_id,
        updated_by_user_id=user_id,
        metadata_=payload.metadata,
    )
    validate_event_time(event)
    session.add(event)
    session.flush()
    if should_push_event_to_caldav(source=source, event=event):
        from govoplan_calendar.backend.outbox import enqueue_caldav_put

        enqueue_caldav_put(session, source=source, event_model=event)
    record_calendar_event_change(session, event=event, operation="created", user_id=user_id)
    return event


def _event_update_calendar_ids(
    session: Session,
    *,
    tenant_id: str,
    event_id: str,
    payload: CalendarEventUpdateRequest,
) -> tuple[str, str]:
    event_locator = (
        session.query(CalendarEvent.calendar_id)
        .filter(
            CalendarEvent.tenant_id == tenant_id,
            CalendarEvent.id == event_id,
            CalendarEvent.deleted_at.is_(None),
        )
        .first()
    )
    if event_locator is None:
        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


def _lock_event_update_sources(
    session: Session,
    *,
    tenant_id: str,
    original_calendar_id: str,
    target_calendar_id: str,
) -> tuple[CalendarSyncSource | None, CalendarSyncSource | None]:
    original_source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=original_calendar_id,
    )
    target_source = (
        original_source
        if target_calendar_id == original_calendar_id
        else active_sync_source_for_calendar(
            session,
            tenant_id=tenant_id,
            calendar_id=target_calendar_id,
        )
    )
    locked_sources = lock_sync_sources(session, original_source, target_source)
    if original_source is not None:
        original_source = locked_sources[original_source.id]
    if target_source is not None:
        target_source = locked_sources[target_source.id]
    return original_source, target_source


def _assert_event_update_sync_fields(
    payload: CalendarEventUpdateRequest,
    *,
    original_source: CalendarSyncSource | None,
    target_source: CalendarSyncSource | None,
) -> None:
    supplied_sync_fields = {"source_kind", "source_href", "etag"} & payload.model_fields_set
    if (original_source is not None or target_source is not None) and supplied_sync_fields:
        raise CalendarError(
            "source_kind, source_href, and etag are sync-owned fields on synchronized calendars"
        )


def _move_event_for_update(
    session: Session,
    *,
    event: CalendarEvent,
    payload: CalendarEventUpdateRequest,
    original_source: CalendarSyncSource | None,
    target_source: CalendarSyncSource | None,
) -> None:
    if payload.calendar_id is None:
        assert_sync_mutation_allowed(original_source)
        return
    assert_sync_mutation_allowed(target_source)
    original_caldav = event.source_kind == "caldav" and bool(event.source_href)
    if payload.calendar_id != event.calendar_id and original_source and original_caldav:
        assert_sync_mutation_allowed(original_source)
        from govoplan_calendar.backend.outbox import enqueue_caldav_delete

        enqueue_caldav_delete(session, source=original_source, event_model=event)
        event.source_kind = "local"
        event.source_href = None
        event.etag = None
    event.calendar_id = payload.calendar_id


def _apply_event_update_values(
    event: CalendarEvent,
    *,
    user_id: str | None,
    payload: CalendarEventUpdateRequest,
) -> None:
    scalar_fields = (
        "sequence",
        "summary",
        "description",
        "location",
        "status",
        "transparency",
        "classification",
        "duration_seconds",
        "all_day",
        "timezone",
        "organizer",
        "attendees",
        "categories",
        "rrule",
        "rdate",
        "exdate",
        "reminders",
        "attachments",
        "related_to",
        "source_kind",
        "source_href",
        "etag",
        "icalendar",
    )
    for attr in scalar_fields:
        value = getattr(payload, attr)
        if value is not None:
            value = (
                value.upper()
                if attr in {"status", "transparency", "classification"}
                and isinstance(value, str)
                else value
            )
            setattr(event, attr, value)
    if payload.start_at is not None:
        event.start_at = normalize_datetime(payload.start_at)
    if "end_at" in payload.model_fields_set:
        event.end_at = normalize_datetime(payload.end_at) if payload.end_at else None
    if payload.metadata is not None:
        event.metadata_ = payload.metadata
    event.updated_by_user_id = user_id
    if payload.sequence is None:
        event.sequence += 1


def update_event(session: Session, *, tenant_id: str, user_id: str | None, event_id: str, payload: CalendarEventUpdateRequest) -> CalendarEvent:
    original_calendar_id, target_calendar_id = _event_update_calendar_ids(
        session,
        tenant_id=tenant_id,
        event_id=event_id,
        payload=payload,
    )
    original_source, target_source = _lock_event_update_sources(
        session,
        tenant_id=tenant_id,
        original_calendar_id=original_calendar_id,
        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,
        original_source=original_source,
        target_source=target_source,
    )
    _move_event_for_update(
        session,
        event=event,
        payload=payload,
        original_source=original_source,
        target_source=target_source,
    )
    _apply_event_update_values(event, user_id=user_id, payload=payload)
    validate_event_time(event)
    event.raw_ics = None
    session.flush()
    if should_push_event_to_caldav(source=target_source, event=event):
        from govoplan_calendar.backend.outbox import enqueue_caldav_put

        enqueue_caldav_put(session, source=target_source, event_model=event)
    record_calendar_event_change(session, event=event, operation="updated", user_id=user_id, previous=previous)
    return event


def update_event_occurrence(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    series_event_id: str,
    payload: CalendarEventOccurrenceUpdateRequest,
) -> CalendarEvent:
    master = get_event(
        session,
        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)
    recurrence_id = str(occurrence["recurrence_id"])
    existing = find_event_override(
        session,
        tenant_id=tenant_id,
        master=master,
        recurrence_id=recurrence_id,
    )
    update_values = payload.model_dump(
        exclude_unset=True,
        exclude={"recurrence_id", "rrule", "rdate", "exdate"},
    )
    if (
        "calendar_id" in update_values
        and update_values["calendar_id"] != master.calendar_id
    ):
        raise CalendarError(
            "A recurring occurrence cannot be moved to another calendar"
        )
    update_values.pop("calendar_id", None)
    update_payload = CalendarEventUpdateRequest(**update_values)
    if existing is not None:
        return update_event(
            session,
            tenant_id=tenant_id,
            user_id=user_id,
            event_id=existing.id,
            payload=update_payload,
        )

    source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=master.calendar_id,
        for_update=True,
    )
    assert_sync_mutation_allowed(source)
    event = CalendarEvent(
        tenant_id=tenant_id,
        calendar_id=master.calendar_id,
        uid=master.uid,
        recurrence_id=recurrence_id,
        sequence=max(0, int(master.sequence or 0)),
        summary=master.summary,
        description=master.description,
        location=master.location,
        status=master.status,
        transparency=master.transparency,
        classification=master.classification,
        start_at=normalize_datetime(occurrence["start_at"]),
        end_at=(
            normalize_datetime(occurrence["end_at"])
            if occurrence.get("end_at") is not None
            else None
        ),
        duration_seconds=master.duration_seconds,
        all_day=master.all_day,
        timezone=master.timezone,
        organizer=copy.deepcopy(master.organizer),
        attendees=copy.deepcopy(master.attendees or []),
        categories=copy.deepcopy(master.categories or []),
        rrule=None,
        rdate=[],
        exdate=[],
        reminders=copy.deepcopy(master.reminders or []),
        attachments=copy.deepcopy(master.attachments or []),
        related_to=copy.deepcopy(master.related_to or []),
        source_kind=master.source_kind,
        source_href=master.source_href,
        etag=master.etag,
        icalendar=copy.deepcopy(master.icalendar or {}),
        raw_ics=None,
        created_by_user_id=user_id,
        updated_by_user_id=user_id,
        metadata_=copy.deepcopy(master.metadata_ or {}),
    )
    _apply_event_update_values(
        event,
        user_id=user_id,
        payload=update_payload,
    )
    validate_event_time(event)
    session.add(event)
    session.flush()
    if should_push_event_to_caldav(source=source, event=event):
        from govoplan_calendar.backend.outbox import enqueue_caldav_put

        enqueue_caldav_put(session, source=source, event_model=event)
    record_calendar_event_change(
        session,
        event=event,
        operation="created",
        user_id=user_id,
    )
    return event


def delete_event_occurrence(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    series_event_id: str,
    recurrence_id: str,
) -> CalendarEvent:
    return update_event_occurrence(
        session,
        tenant_id=tenant_id,
        user_id=user_id,
        series_event_id=series_event_id,
        payload=CalendarEventOccurrenceUpdateRequest(
            recurrence_id=recurrence_id,
            status="CANCELLED",
        ),
    )


def recurrence_occurrence(
    master: CalendarEvent,
    recurrence_id: str,
) -> dict[str, Any]:
    recurrence_start = recurrence_id_datetime(recurrence_id)
    if recurrence_start is None:
        raise CalendarError("Invalid recurrence_id")
    recurrence_key = normalized_recurrence_id(recurrence_id)
    candidates = expand_event_occurrences(
        master,
        recurrence_start - timedelta(seconds=1),
        recurrence_start + timedelta(seconds=1),
    )
    occurrence = next(
        (
            item
            for item in candidates
            if normalized_recurrence_id(item.get("recurrence_id"))
            == recurrence_key
        ),
        None,
    )
    if occurrence is None:
        raise CalendarError(
            "recurrence_id is not an occurrence in this event series"
        )
    return occurrence


def find_event_override(
    session: Session,
    *,
    tenant_id: str,
    master: CalendarEvent,
    recurrence_id: str,
) -> CalendarEvent | None:
    recurrence_key = normalized_recurrence_id(recurrence_id)
    candidates = (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == tenant_id,
            CalendarEvent.calendar_id == master.calendar_id,
            CalendarEvent.uid == master.uid,
            CalendarEvent.recurrence_id.is_not(None),
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    )
    return next(
        (
            event
            for event in candidates
            if normalized_recurrence_id(event.recurrence_id)
            == recurrence_key
        ),
        None,
    )


def delete_event(session: Session, *, tenant_id: str, event_id: str, user_id: str | None = None) -> None:
    event_locator = (
        session.query(CalendarEvent.calendar_id)
        .filter(
            CalendarEvent.tenant_id == tenant_id,
            CalendarEvent.id == event_id,
            CalendarEvent.deleted_at.is_(None),
        )
        .first()
    )
    if event_locator is None:
        raise CalendarError("Calendar event not found")
    source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=event_locator[0],
        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):
        series_events = (
            session.query(CalendarEvent)
            .filter(
                CalendarEvent.tenant_id == tenant_id,
                CalendarEvent.calendar_id == event.calendar_id,
                CalendarEvent.uid == event.uid,
                CalendarEvent.deleted_at.is_(None),
            )
            .all()
        )
    previous_by_id = {
        item.id: calendar_event_change_payload(
            item,
            prefix="previous_",
        )
        for item in series_events
    }
    push_delete = should_push_event_to_caldav(source=source, event=event)
    deleted_at = utcnow()
    for item in series_events:
        item.deleted_at = deleted_at
    session.flush()
    if push_delete:
        from govoplan_calendar.backend.outbox import enqueue_caldav_delete

        enqueue_caldav_delete(session, source=source, event_model=event)
    for item in series_events:
        record_calendar_event_change(
            session,
            event=item,
            operation="deleted",
            user_id=user_id,
            previous=previous_by_id[item.id],
        )


def active_sync_source_for_calendar(
    session: Session,
    *,
    tenant_id: str,
    calendar_id: str,
    for_update: bool = False,
) -> CalendarSyncSource | None:
    query = (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.calendar_id == calendar_id,
            CalendarSyncSource.deleted_at.is_(None),
        )
        .order_by((CalendarSyncSource.sync_direction == "two_way").desc(), CalendarSyncSource.created_at.asc())
    )
    if for_update:
        query = query.populate_existing().with_for_update()
    return query.first()


def lock_sync_sources(
    session: Session,
    *sources: CalendarSyncSource | None,
) -> dict[str, CalendarSyncSource]:
    """Lock source rows in stable order before producing desired snapshots."""

    source_ids = sorted({source.id for source in sources if source is not None})
    if not source_ids:
        return {}
    locked_sources = (
        session.query(CalendarSyncSource)
        .filter(CalendarSyncSource.id.in_(source_ids))
        .order_by(CalendarSyncSource.id.asc())
        .populate_existing()
        .with_for_update()
        .all()
    )
    return {source.id: source for source in locked_sources}


def active_caldav_source_for_calendar(session: Session, *, tenant_id: str, calendar_id: str) -> CalendarSyncSource | None:
    return (
        session.query(CalendarSyncSource)
        .filter(
            CalendarSyncSource.tenant_id == tenant_id,
            CalendarSyncSource.calendar_id == calendar_id,
            CalendarSyncSource.source_kind == "caldav",
            CalendarSyncSource.deleted_at.is_(None),
        )
        .order_by((CalendarSyncSource.sync_direction == "two_way").desc(), CalendarSyncSource.created_at.asc())
        .first()
    )


def assert_sync_mutation_allowed(source: CalendarSyncSource | None) -> None:
    if source is None:
        return
    if source.deleted_at is not None:
        raise CalendarError("This calendar synchronization source is no longer active")
    if not source.sync_enabled:
        raise CalendarError(
            "This calendar synchronization source is disabled and cannot accept local changes"
        )
    if source.source_kind != "caldav" or source.sync_direction != "two_way":
        raise CalendarError(f"This {sync_source_label(source.source_kind)} calendar is inbound-only and cannot be changed locally")


def assert_caldav_mutation_allowed(source: CalendarSyncSource | None) -> None:
    assert_sync_mutation_allowed(source)


def should_push_event_to_caldav(*, source: CalendarSyncSource | None, event: CalendarEvent) -> bool:
    return bool(
        source
        and source.deleted_at is None
        and source.sync_enabled
        and source.source_kind == "caldav"
        and source.sync_direction == "two_way"
        and event.deleted_at is None
    )


def caldav_resource_events(session: Session, *, source: CalendarSyncSource, href: str) -> list[CalendarEvent]:
    return (
        session.query(CalendarEvent)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_kind == "caldav",
            CalendarEvent.source_href == href,
            CalendarEvent.deleted_at.is_(None),
        )
        .all()
    )


def caldav_resource_sort_key(event: CalendarEvent) -> tuple[int, str, datetime]:
    return (1 if event.recurrence_id else 0, event.recurrence_id or "", event.start_at)


def resource_etag(events: list[CalendarEvent]) -> str | None:
    return next((event.etag for event in events if event.etag), None)


def generated_caldav_href(session: Session, *, source: CalendarSyncSource, event: CalendarEvent) -> str:
    safe_uid = urllib.parse.quote(event.uid, safe="-_.~@")
    candidate = normalize_caldav_href(source.collection_url, f"{safe_uid}.ics")
    existing = (
        session.query(CalendarEvent.id)
        .filter(
            CalendarEvent.tenant_id == source.tenant_id,
            CalendarEvent.calendar_id == source.calendar_id,
            CalendarEvent.source_href == candidate,
            CalendarEvent.deleted_at.is_(None),
            CalendarEvent.id != event.id,
        )
        .first()
    )
    if existing is None:
        return candidate
    return normalize_caldav_href(source.collection_url, f"{safe_uid}-{event.id}.ics")


def import_ics_event(
    session: Session,
    *,
    tenant_id: str,
    user_id: str | None,
    calendar_id: str | None,
    ics: str,
    upsert: bool,
    source_kind: str,
    source_href: str | None,
    etag: str | None,
) -> CalendarEvent:
    parsed = parse_vevent(ics)
    raw_ics = str(parsed.pop("raw_ics", ""))
    default_calendar = None if calendar_id else get_default_calendar(session, tenant_id=tenant_id)
    target_calendar_id = calendar_id or (default_calendar.id if default_calendar else None)
    if not target_calendar_id:
        raise CalendarError("calendar_id is required because no default calendar exists")
    target_calendar = get_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=target_calendar_id,
    )
    # Source creation also locks the calendar row.  Holding it here makes the
    # public import's provenance decision stable until create/update has queued
    # its desired state.
    (
        session.query(CalendarCollection)
        .filter(
            CalendarCollection.id == target_calendar.id,
            CalendarCollection.tenant_id == tenant_id,
            CalendarCollection.deleted_at.is_(None),
        )
        .populate_existing()
        .with_for_update()
        .one()
    )
    target_source = active_sync_source_for_calendar(
        session,
        tenant_id=tenant_id,
        calendar_id=target_calendar_id,
        for_update=True,
    )
    synchronized_target = target_source is not None
    existing = None
    if upsert:
        existing = (
            session.query(CalendarEvent)
            .filter(
                CalendarEvent.tenant_id == tenant_id,
                CalendarEvent.calendar_id == target_calendar_id,
                CalendarEvent.uid == parsed["uid"],
                CalendarEvent.recurrence_id == parsed.get("recurrence_id"),
                CalendarEvent.deleted_at.is_(None),
            )
            .first()
        )
    payload_values: dict[str, Any] = {
        "calendar_id": target_calendar_id,
        "metadata": {},
        **parsed,
    }
    if not synchronized_target:
        payload_values.update(
            {
                "source_kind": source_kind,
                "source_href": source_href,
                "etag": etag,
            }
        )
    payload = CalendarEventCreateRequest(**payload_values)
    if existing:
        excluded_fields = {"uid", "recurrence_id"}
        if synchronized_target:
            excluded_fields.update({"source_kind", "source_href", "etag"})
        update_payload = CalendarEventUpdateRequest(
            **payload.model_dump(exclude=excluded_fields)
        )
        event = update_event(session, tenant_id=tenant_id, user_id=user_id, event_id=existing.id, payload=update_payload)
        event.raw_ics = raw_ics
        return event
    event = create_event(session, tenant_id=tenant_id, user_id=user_id, payload=payload)
    event.raw_ics = raw_ics
    return event


def normalize_datetime(value: datetime) -> datetime:
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def response_datetime(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    return normalize_datetime(value)


def validate_event_time(event: CalendarEvent) -> None:
    if event.end_at is not None and event.end_at < event.start_at:
        raise CalendarError("Event end must be after start")


def calendar_response(calendar: CalendarCollection) -> dict[str, Any]:
    return {
        "id": calendar.id,
        "tenant_id": calendar.tenant_id,
        "slug": calendar.slug,
        "name": calendar.name,
        "description": calendar.description,
        "timezone": calendar.timezone,
        "color": calendar.color,
        "owner_type": calendar.owner_type,
        "owner_id": calendar.owner_id,
        "visibility": calendar.visibility,
        "is_default": calendar.is_default,
        "created_at": response_datetime(calendar.created_at),
        "updated_at": response_datetime(calendar.updated_at),
        "metadata": calendar.metadata_ or {},
    }


def event_response(event: CalendarEvent) -> dict[str, Any]:
    return {
        "id": event.id,
        "tenant_id": event.tenant_id,
        "calendar_id": event.calendar_id,
        "uid": event.uid,
        "recurrence_id": event.recurrence_id,
        "sequence": event.sequence,
        "summary": event.summary,
        "description": event.description,
        "location": event.location,
        "status": event.status,
        "transparency": event.transparency,
        "classification": event.classification,
        "start_at": response_datetime(event.start_at),
        "end_at": response_datetime(event.end_at),
        "duration_seconds": event.duration_seconds,
        "all_day": event.all_day,
        "timezone": event.timezone,
        "organizer": event.organizer,
        "attendees": event.attendees or [],
        "categories": event.categories or [],
        "rrule": event.rrule,
        "rdate": event.rdate or [],
        "exdate": event.exdate or [],
        "reminders": event.reminders or [],
        "attachments": event.attachments or [],
        "related_to": event.related_to or [],
        "source_kind": event.source_kind,
        "source_href": event.source_href,
        "etag": event.etag,
        "icalendar": event.icalendar or {},
        "created_at": response_datetime(event.created_at),
        "updated_at": response_datetime(event.updated_at),
        "metadata": event.metadata_ or {},
        "instance_id": event.id,
        "series_event_id": None,
        "is_occurrence": False,
        "is_override": False,
    }