GovOPlaN/govoplan-calendar
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(calendar): make external mutations durable

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-20 16:58:45 +02:00
parent 371a00aff5
commit a6cd64e3f9
15 changed files with 4726 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

685
src/govoplan_calendar/backend/service.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import json
import posixpath
import re
import uuid
import os
@@ -9,16 +10,16 @@ import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Callable
from typing import Any, Callable, Iterable
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from defusedxml import ElementTree as SafeElementTree
from sqlalchemy import or_
from sqlalchemy.orm import Session
from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarSyncCredential, CalendarSyncSource
from govoplan_calendar.backend.ical import events_to_ics, expand_event_occurrences, parse_vevent, parse_vevents
from govoplan_calendar.backend.ical import expand_event_occurrences, parse_vevent, parse_vevents
from govoplan_calendar.backend.runtime import get_registry
from govoplan_calendar.backend.schemas import (
CalendarCalDavDiscoveryRequest,
@@ -58,12 +59,20 @@ SOURCE_EVENT_CLEANUP_BATCH_SIZE = 500
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
),
}
@@ -324,15 +333,63 @@ def get_default_calendar(session: Session, *, tenant_id: str) -> CalendarCollect
)
def list_calendars(session: Session, *, tenant_id: str, ensure_default: bool = False, user_id: str | None = None) -> list[CalendarCollection]:
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)
return (
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:
@@ -387,14 +444,129 @@ def delete_calendar(session: Session, *, tenant_id: str, calendar_id: str, paylo
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)
for event in calendar.events:
if event.deleted_at is None:
event.calendar_id = target_calendar.id
source = None
target_source = None
if hasattr(session, "query"):
# Source creation takes the collection row lock before linking a
# source. Lock both collections in stable order so the mode 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]
if payload.external_action == "remote_move":
raise CalendarError(
"external_action='remote_move' is not supported; no destructive remote move was performed"
)
if source is not None and target_source is not None:
raise CalendarError(
"Events cannot be bulk-moved between synchronized calendars; "
"external_action='remote_move' remains unsupported"
)
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 {}
)
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,
)
elif 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)
elif payload.external_action is not None:
raise CalendarError(
"external_action is only valid when moving events to or from a synchronized calendar"
)
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
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 calendar.is_default:
calendar.is_default = False
calendar.deleted_at = deleted_at
@@ -515,6 +687,30 @@ def discover_caldav_calendars(session: Session, *, tenant_id: str, payload: Cale
def create_sync_source(session: Session, *, tenant_id: str, user_id: str | None, payload: CalendarSyncSourceCreateRequest) -> CalendarSyncSource:
calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=payload.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()
)
existing_calendar_source = (
session.query(CalendarSyncSource.id)
.filter(
CalendarSyncSource.tenant_id == tenant_id,
CalendarSyncSource.calendar_id == calendar.id,
CalendarSyncSource.deleted_at.is_(None),
)
.first()
)
if existing_calendar_source is not None:
raise CalendarError(
"A calendar can have only one active synchronization source"
)
source_kind = normalize_source_kind(payload.source_kind)
collection_url = normalize_sync_source_url(source_kind, payload.collection_url)
retire_stale_sync_sources_for_url(session, tenant_id=tenant_id, source_kind=source_kind, collection_url=collection_url)
@@ -562,8 +758,123 @@ def create_caldav_source(session: Session, *, tenant_id: str, user_id: str | Non
def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payload: CalendarSyncSourceUpdateRequest) -> CalendarSyncSource:
source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
source = (
session.query(CalendarSyncSource)
.filter(
CalendarSyncSource.id == source.id,
CalendarSyncSource.tenant_id == tenant_id,
CalendarSyncSource.deleted_at.is_(None),
)
.populate_existing()
.with_for_update()
.one()
)
normalized_collection_url = (
normalize_sync_source_url(source.source_kind, payload.collection_url)
if payload.collection_url is not None
else source.collection_url
)
materially_reconfigured = bool(
source.source_kind == "caldav"
and (
(payload.calendar_id is not None and payload.calendar_id != source.calendar_id)
or normalized_collection_url != source.collection_url
or payload.sync_enabled is False
or payload.sync_direction == "inbound"
)
)
endpoint_or_calendar_changed = bool(
source.source_kind == "caldav"
and (
(payload.calendar_id is not None and payload.calendar_id != source.calendar_id)
or normalized_collection_url != source.collection_url
)
)
if materially_reconfigured:
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:
sourced_event_exists = (
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
)
if sourced_event_exists:
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",
)
if payload.calendar_id is not None:
calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=payload.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
else:
calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=source.calendar_id)
@@ -583,7 +894,7 @@ def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payl
if value is not None:
setattr(source, field_name, value)
if payload.collection_url is not None:
source.collection_url = normalize_sync_source_url(source.source_kind, payload.collection_url)
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:
@@ -684,6 +995,31 @@ def active_caldav_source_for_url(session: Session, *, tenant_id: str, collection
def retire_sync_source(session: Session, *, tenant_id: str, source: CalendarSyncSource, deleted_at: datetime) -> 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"
)
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
delete_caldav_credential(session, tenant_id=tenant_id, credential_ref=source.credential_ref)
@@ -933,6 +1269,22 @@ def sync_caldav_source(
force_full: bool = False,
) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
source = get_caldav_source(session, tenant_id=tenant_id, source_id=source_id)
# Serialize inbound application with local desired-state snapshots and
# outbound delivery for this source. The lock intentionally spans the
# REPORT and its application so an older report cannot land after a newer
# outbound write.
source = (
session.query(CalendarSyncSource)
.filter(
CalendarSyncSource.id == source.id,
CalendarSyncSource.tenant_id == tenant_id,
CalendarSyncSource.source_kind == "caldav",
CalendarSyncSource.deleted_at.is_(None),
)
.populate_existing()
.with_for_update()
.one()
)
get_calendar(session, tenant_id=tenant_id, calendar_id=source.calendar_id)
client = client or caldav_client_for_source(session, source, password=password, bearer_token=bearer_token)
stats = CalendarCalDavSyncStats()
@@ -1452,6 +1804,15 @@ def soft_delete_unseen_source_events(
.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
@@ -1714,6 +2075,14 @@ def apply_caldav_report(
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
@@ -1899,7 +2268,33 @@ def soft_delete_caldav_href(session: Session, *, source: CalendarSyncSource, hre
def normalize_caldav_href(collection_url: str, href: str) -> str:
return urllib.parse.urljoin(ensure_collection_url(collection_url), href)
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:
@@ -2044,8 +2439,25 @@ def create_event(session: Session, *, tenant_id: str, user_id: str | None, paylo
if not calendar_id:
raise CalendarError("calendar_id is required because no default calendar exists")
get_calendar(session, tenant_id=tenant_id, calendar_id=calendar_id)
source = active_sync_source_for_calendar(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,
@@ -2084,24 +2496,63 @@ def create_event(session: Session, *, tenant_id: str, user_id: str | None, paylo
session.add(event)
session.flush()
if should_push_event_to_caldav(source=source, event=event):
push_caldav_event_resource(session, source=source, event=event, create=True)
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 update_event(session: Session, *, tenant_id: str, user_id: str | None, event_id: str, payload: CalendarEventUpdateRequest) -> CalendarEvent:
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
if payload.calendar_id is not None:
get_calendar(session, tenant_id=tenant_id, calendar_id=target_calendar_id)
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]
event = get_event(session, tenant_id=tenant_id, event_id=event_id)
previous = calendar_event_change_payload(event, prefix="previous_")
original_calendar_id = event.calendar_id
original_source = active_sync_source_for_calendar(session, tenant_id=tenant_id, calendar_id=event.calendar_id)
original_caldav = event.source_kind == "caldav" and bool(event.source_href)
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"
)
if payload.calendar_id is not None:
get_calendar(session, tenant_id=tenant_id, calendar_id=payload.calendar_id)
target_source = active_sync_source_for_calendar(session, tenant_id=tenant_id, calendar_id=payload.calendar_id)
assert_sync_mutation_allowed(target_source)
if payload.calendar_id != event.calendar_id and original_source and original_caldav:
assert_sync_mutation_allowed(original_source)
push_caldav_delete_for_event(session, source=original_source, event=event)
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
@@ -2149,32 +2600,52 @@ def update_event(session: Session, *, tenant_id: str, user_id: str | None, event
validate_event_time(event)
event.raw_ics = None
session.flush()
target_source = active_sync_source_for_calendar(session, tenant_id=tenant_id, calendar_id=event.calendar_id)
if should_push_event_to_caldav(source=target_source, event=event):
push_caldav_event_resource(
session,
source=target_source,
event=event,
create=event.source_kind != "caldav" or not event.source_href or original_calendar_id != event.calendar_id,
)
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 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)
previous = calendar_event_change_payload(event, prefix="previous_")
source = active_sync_source_for_calendar(session, tenant_id=tenant_id, calendar_id=event.calendar_id)
assert_sync_mutation_allowed(source)
if should_push_event_to_caldav(source=source, event=event):
push_caldav_delete_for_event(session, source=source, event=event)
from govoplan_calendar.backend.outbox import enqueue_caldav_delete
enqueue_caldav_delete(session, source=source, event_model=event)
event.deleted_at = utcnow()
session.flush()
record_calendar_event_change(session, event=event, operation="deleted", user_id=user_id, previous=previous)
def active_sync_source_for_calendar(session: Session, *, tenant_id: str, calendar_id: str) -> CalendarSyncSource | None:
return (
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,
@@ -2182,8 +2653,30 @@ def active_sync_source_for_calendar(session: Session, *, tenant_id: str, calenda
CalendarSyncSource.deleted_at.is_(None),
)
.order_by((CalendarSyncSource.sync_direction == "two_way").desc(), CalendarSyncSource.created_at.asc())
.first()
)
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:
@@ -2203,6 +2696,12 @@ def active_caldav_source_for_calendar(session: Session, *, tenant_id: str, calen
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")
@@ -2212,63 +2711,14 @@ def assert_caldav_mutation_allowed(source: CalendarSyncSource | None) -> None:
def should_push_event_to_caldav(*, source: CalendarSyncSource | None, event: CalendarEvent) -> bool:
return bool(source and source.source_kind == "caldav" and source.sync_direction == "two_way" and event.deleted_at is None)
def push_caldav_event_resource(session: Session, *, source: CalendarSyncSource, event: CalendarEvent, create: bool) -> None:
href = event.source_href or generated_caldav_href(session, source=source, event=event)
full_href = normalize_caldav_href(source.collection_url, href)
resource_events = caldav_resource_events(session, source=source, href=full_href) if event.source_href else [event]
if event not in resource_events:
resource_events.append(event)
resource_events = sorted(resource_events, key=caldav_resource_sort_key)
ics = events_to_ics(resource_events)
client = caldav_client_for_source(session, source)
overwrite = source.conflict_policy == "overwrite"
etag = None if create else resource_etag(resource_events)
try:
result = client.put_object(full_href, ics, etag=etag, create=create, overwrite=overwrite)
except CalDAVPreconditionFailed as exc:
raise CalendarError("CalDAV resource changed remotely; sync the calendar before saving again") from exc
except CalDAVError as exc:
raise CalendarError(f"CalDAV write failed: {exc}") from exc
next_etag = result.etag
for item in resource_events:
apply_caldav_write_metadata(item, source=source, href=full_href, etag=next_etag, raw_ics=ics)
source.last_status = "outbound_ok"
source.last_error = None
schedule_next_caldav_sync(source)
session.flush()
def push_caldav_delete_for_event(session: Session, *, source: CalendarSyncSource, event: CalendarEvent) -> None:
if not event.source_href:
return
full_href = normalize_caldav_href(source.collection_url, event.source_href)
remaining = [
item
for item in caldav_resource_events(session, source=source, href=full_href)
if item.id != event.id
]
client = caldav_client_for_source(session, source)
overwrite = source.conflict_policy == "overwrite"
try:
if remaining:
remaining = sorted(remaining, key=caldav_resource_sort_key)
ics = events_to_ics(remaining)
result = client.put_object(full_href, ics, etag=event.etag or resource_etag(remaining), create=False, overwrite=overwrite)
for item in remaining:
apply_caldav_write_metadata(item, source=source, href=full_href, etag=result.etag, raw_ics=ics)
else:
client.delete_object(full_href, etag=event.etag, overwrite=overwrite)
except CalDAVPreconditionFailed as exc:
raise CalendarError("CalDAV resource changed remotely; sync the calendar before deleting again") from exc
except CalDAVError as exc:
raise CalendarError(f"CalDAV delete failed: {exc}") from exc
source.last_status = "outbound_ok"
source.last_error = None
schedule_next_caldav_sync(source)
session.flush()
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]:
@@ -2312,16 +2762,6 @@ def generated_caldav_href(session: Session, *, source: CalendarSyncSource, event
return normalize_caldav_href(source.collection_url, f"{safe_uid}-{event.id}.ics")
def apply_caldav_write_metadata(event: CalendarEvent, *, source: CalendarSyncSource, href: str, etag: str | None, raw_ics: str) -> None:
event.source_kind = "caldav"
event.source_href = href
event.etag = etag
event.raw_ics = raw_ics
metadata = dict(event.metadata_ or {})
metadata["caldav"] = {"source_id": source.id, "collection_url": source.collection_url, "href": href, "etag": etag}
event.metadata_ = metadata
def import_ics_event(
session: Session,
*,
@@ -2340,6 +2780,32 @@ def import_ics_event(
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 = (
@@ -2353,16 +2819,27 @@ def import_ics_event(
)
.first()
)
payload = CalendarEventCreateRequest(
calendar_id=target_calendar_id,
source_kind=source_kind,
source_href=source_href,
etag=etag,
metadata={},
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:
update_payload = CalendarEventUpdateRequest(**payload.model_dump(exclude={"uid", "recurrence_id"}))
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