refactor: harden calendar sync and add dashboard widget

src/govoplan_calendar/backend/manifest.py

@@ -25,6 +25,7 @@ from govoplan_core.core.modules import (
     PermissionDefinition,
     RoleTemplate,
 )
+from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 
 
@@ -204,6 +205,15 @@ manifest = ModuleManifest(
             ),
         ),
         nav_items=(NavItem(path="/calendar", label="Calendar", icon="calendar", required_any=("calendar:event:read",), order=55),),
+        view_surfaces=(
+            ViewSurface(
+                id="calendar.widget.upcoming",
+                module_id="calendar",
+                kind="section",
+                label="Upcoming events widget",
+                order=40,
+            ),
+        ),
     ),
     migration_spec=MigrationSpec(
         module_id="calendar",

src/govoplan_calendar/backend/service.py

@@ -34,7 +34,7 @@ from govoplan_core.security.credential_envelopes import (
 )
 from govoplan_core.audit.logging import audit_event
 
-from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
+from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url
 from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarSyncCredential, CalendarSyncSource
 from govoplan_calendar.backend.ical import expand_event_occurrences, parse_vevent, parse_vevents
 from govoplan_calendar.backend.runtime import get_registry
@@ -74,6 +74,7 @@ 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
 
 
 def calendar_credential_context(
@@ -169,6 +170,36 @@ class CalendarCalDavDueSyncResult:
     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]
+
+
 def slugify(value: str) -> str:
     slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
     return slug or "calendar"
@@ -1963,8 +1994,16 @@ def sync_source(
     bearer_token: str | None = None,
     force_full: bool = False,
 ) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
-    source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
-    if source.source_kind == "caldav":
+    # 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,
@@ -1974,14 +2013,13 @@ def sync_source(
             bearer_token=bearer_token,
             force_full=force_full,
         )
-    get_calendar(session, tenant_id=tenant_id, calendar_id=source.calendar_id)
-    if source.source_kind in {"ics", "webcal"}:
+    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.source_kind == "graph":
+    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.source_kind == "ews":
+    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.source_kind}")
+    raise CalendarError(f"Unsupported calendar sync source kind: {source_kind}")
 
 
 def http_request(
@@ -2131,37 +2169,26 @@ def sync_caldav_source(
     bearer_token: str | None = None,
     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),
+    session.commit()
+    snapshot, client = _prepare_caldav_sync(
+        session,
+        tenant_id=tenant_id,
+        source_id=source_id,
+        client=client,
+        password=password,
+        bearer_token=bearer_token,
     )
-        .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()
     collection_props = CalDAVReportResult()
-    source.last_attempt_at = utcnow()
     try:
         try:
             collection_props = client.propfind_collection()
         except CalDAVError as exc:
             stats.errors.append(f"Collection PROPFIND failed; continuing with REPORT: {exc}")
 
-        if source.sync_token and not force_full:
+        if snapshot.sync_token and not force_full:
             try:
-                report = client.sync_collection(source.sync_token)
+                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}")
@@ -2171,7 +2198,16 @@ def sync_caldav_source(
             report = client.list_objects()
             stats.full_sync = True
 
-        apply_caldav_report(session, source=source, client=client, report=report, stats=stats, user_id=user_id)
+        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(
@@ -2183,10 +2219,262 @@ def sync_caldav_source(
         )
         return source, stats
     except Exception as exc:
-        _finalize_sync_error(session, source=source, error=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,
+        )
+        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,
+        )
+        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,
     *,
@@ -2197,20 +2485,30 @@ def sync_ics_source(
     bearer_token: str | None = None,
     force_full: bool = False,
 ) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
-    source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
-    if source.source_kind not in {"ics", "webcal"}:
-        raise CalendarError("ICS/webcal sync source not found")
+    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)
-    source.last_attempt_at = utcnow()
     headers = {"Accept": "text/calendar, application/calendar+json;q=0.5, */*;q=0.1", "User-Agent": "govoplan-calendar-sync/0.1"}
-    headers.update(source_auth_headers(session, source, password=password, bearer_token=bearer_token))
-    if source.ctag and not force_full:
-        headers["If-None-Match"] = source.ctag
-    last_modified = (source.metadata_ or {}).get("last_modified") if isinstance(source.metadata_, dict) else None
+    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(source.collection_url, headers=headers)
+        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
@@ -2230,7 +2528,9 @@ def sync_ics_source(
         _finalize_sync_success(session, source=source, stats=stats, mark_calendar=mark_calendar_sync_source)
         return source, stats
     except Exception as exc:
-        _finalize_sync_error(session, source=source, error=exc)
+        session.rollback()
+        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
+        session.commit()
         raise
 
 
@@ -2329,58 +2629,90 @@ def sync_graph_source(
     bearer_token: str | None = None,
     force_full: bool = False,
 ) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
-    source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
-    if source.source_kind != "graph":
-        raise CalendarError("Microsoft Graph sync source not found")
+    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(source_auth_headers(session, source, bearer_token=bearer_token))
-    stats = CalendarCalDavSyncStats(full_sync=force_full or not bool(source.sync_token), used_sync_token=bool(source.sync_token and not force_full))
-    source.last_attempt_at = utcnow()
+    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 = source.sync_token if source.sync_token and not force_full else source.collection_url
-        seen_hrefs: set[str] = set()
+        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(
-                source.collection_url,
+                snapshot.collection_url,
                 next_url,
                 label="Microsoft Graph continuation URL",
             )
             _status, _headers, body = http_request(
                 next_url,
                 headers=headers,
-                credential_origin=source.collection_url,
+                credential_origin=snapshot.collection_url,
             )
             payload = json.loads(body or "{}")
             for item in payload.get("value", []):
-                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 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(
-                    source.collection_url,
+                    snapshot.collection_url,
                     str(raw_delta_link),
                     label="Microsoft Graph delta URL",
                 )
             if not next_link:
                 break
             next_url = same_origin_http_url(
-                source.collection_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
@@ -2392,11 +2724,9 @@ def sync_graph_source(
         session.flush()
         return source, stats
     except Exception as exc:
-        source.last_attempt_at = utcnow()
-        source.last_status = "error"
-        source.last_error = str(exc)
-        schedule_next_caldav_sync(source)
-        session.flush()
+        session.rollback()
+        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
+        session.commit()
         raise
 
 
@@ -2493,10 +2823,16 @@ def sync_ews_source(
     bearer_token: str | None = None,
     force_full: bool = False,
 ) -> tuple[CalendarSyncSource, CalendarCalDavSyncStats]:
-    source = get_sync_source(session, tenant_id=tenant_id, source_id=source_id)
-    if source.source_kind != "ews":
-        raise CalendarError("Exchange Web Services sync source not found")
-    metadata = source.metadata_ or {}
+    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()
@@ -2507,12 +2843,26 @@ def sync_ews_source(
         "Content-Type": "text/xml; charset=utf-8",
         "User-Agent": "govoplan-calendar-ews/0.1",
     }
-    headers.update(source_auth_headers(session, source, password=password, bearer_token=bearer_token))
+    headers.update(auth_headers)
     stats = CalendarCalDavSyncStats(full_sync=True)
-    source.last_attempt_at = utcnow()
     try:
-        _status, _headers, body = http_request(source.collection_url, method="POST", headers=headers, body=ews_find_item_body(start=start, end=end, mailbox=metadata.get("mailbox") if isinstance(metadata, dict) else None))
+        _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"),
+            ),
+        )
         items = parse_ews_calendar_items(body)
+        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"]
@@ -2532,11 +2882,9 @@ def sync_ews_source(
         session.flush()
         return source, stats
     except Exception as exc:
-        source.last_attempt_at = utcnow()
-        source.last_status = "error"
-        source.last_error = str(exc)
-        schedule_next_caldav_sync(source)
-        session.flush()
+        session.rollback()
+        _record_remote_sync_error(session, snapshot=snapshot, error=exc)
+        session.commit()
         raise
 
 
@@ -2909,7 +3257,23 @@ def _sync_due_sources(
     )
     if tenant_id is not None:
         query = query.filter(CalendarSyncSource.tenant_id == tenant_id)
-    sources = query.order_by(CalendarSyncSource.next_sync_at.asc(), CalendarSyncSource.created_at.asc()).limit(limit).all()
+    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
@@ -2917,10 +3281,17 @@ def _sync_due_sources(
             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=source, status="error", previous_status=previous_status, error=str(exc))
-    session.flush()
+            _emit_calendar_sync_notification(session, source=refreshed, status="error", previous_status=previous_status, error=str(exc))
+            session.commit()
     return results
 
 
@@ -2949,7 +3320,7 @@ def apply_caldav_report(
     session: Session,
     *,
     source: CalendarSyncSource,
-    client: CalDAVClient,
+    client: CalDAVClient | None = None,
     report: CalDAVReportResult,
     stats: CalendarCalDavSyncStats,
     user_id: str | None,
@@ -2971,13 +3342,9 @@ def apply_caldav_report(
         seen_hrefs.add(href)
         calendar_data = item.calendar_data
         if calendar_data is None:
-            try:
-                calendar_data = client.fetch_object(item.href)
-                stats.fetched += 1
-            except CalDAVNotFound:
-                stats.deleted += soft_delete_caldav_href(session, source=source, href=href)
-                stats.errors.append(f"CalDAV object disappeared during sync and was treated as deleted: {href}")
-                continue
+            raise CalendarError(
+                "CalDAV reports must be hydrated before database application."
+            )
         created, updated, unchanged, deleted = import_caldav_resource(
             session,
             source=source,

webui/package.json

@@ -21,7 +21,7 @@
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",
     "react-dom": "^19.0.0",
-    "react-router-dom": "^7.1.1",
+    "react-router-dom": ">=7.18.2 <8",
     "@vitejs/plugin-react": "^4.3.4",
     "typescript": "^5.7.2",
     "vite": "^6.0.6"

webui/src/features/calendar/UpcomingEventsWidget.tsx

@@ -0,0 +1,110 @@
+import { useCallback } from "react";
+import { CalendarDays, MapPin } from "lucide-react";
+import { Link } from "react-router-dom";
+import {
+  DashboardWidgetList,
+  DismissibleAlert,
+  LoadingFrame,
+  useDashboardWidgetData,
+  type ApiSettings,
+  type DashboardWidgetConfiguration
+} from "@govoplan/core-webui";
+import {
+  listCalendarEvents,
+  type CalendarEvent
+} from "../../api/calendar";
+
+export default function UpcomingEventsWidget({
+  settings,
+  refreshKey,
+  configuration
+}: {
+  settings: ApiSettings;
+  refreshKey: number;
+  configuration: DashboardWidgetConfiguration;
+}) {
+  const maxItems = numberSetting(configuration.maxItems, 5, 1, 12);
+  const daysAhead = numberSetting(configuration.daysAhead, 14, 1, 90);
+  const showLocation = configuration.showLocation !== false;
+  const load = useCallback(async () => {
+    const start = new Date();
+    const end = new Date(start);
+    end.setDate(end.getDate() + daysAhead);
+    const response = await listCalendarEvents(settings, {
+      start_at: start.toISOString(),
+      end_at: end.toISOString()
+    });
+    return [...response.events]
+      .filter((event) => event.status !== "cancelled")
+      .sort(
+        (left, right) =>
+          new Date(left.start_at).getTime() - new Date(right.start_at).getTime()
+      )
+      .slice(0, maxItems);
+  }, [daysAhead, maxItems, settings]);
+  const { data: events, loading, error } = useDashboardWidgetData(
+    load,
+    refreshKey
+  );
+
+  return (
+    <LoadingFrame loading={loading} label="Loading upcoming calendar events">
+      {error && (
+        <DismissibleAlert tone="warning" resetKey={error}>
+          {error}
+        </DismissibleAlert>
+      )}
+      <DashboardWidgetList
+        emptyText={`No events in the next ${daysAhead} days.`}
+        items={(events ?? []).map((event) => ({
+          id: event.id,
+          title: event.summary,
+          detail:
+            showLocation && event.location ? (
+              <>
+                <MapPin size={12} aria-hidden="true" /> {event.location}
+              </>
+            ) : undefined,
+          meta: eventTimeLabel(event),
+          leading: <CalendarDays size={17} aria-hidden="true" />,
+          to: "/calendar"
+        }))}
+      />
+      <div className="dashboard-contribution-footer">
+        <Link className="btn btn-secondary" to="/calendar">
+          Open calendar
+        </Link>
+      </div>
+    </LoadingFrame>
+  );
+}
+
+function eventTimeLabel(event: CalendarEvent): string {
+  const start = new Date(event.start_at);
+  if (event.all_day) {
+    return new Intl.DateTimeFormat(undefined, {
+      weekday: "short",
+      day: "2-digit",
+      month: "short"
+    }).format(start);
+  }
+  return new Intl.DateTimeFormat(undefined, {
+    weekday: "short",
+    day: "2-digit",
+    month: "short",
+    hour: "2-digit",
+    minute: "2-digit"
+  }).format(start);
+}
+
+function numberSetting(
+  value: unknown,
+  fallback: number,
+  minimum: number,
+  maximum: number
+): number {
+  const numeric = typeof value === "number" ? value : Number(value);
+  return Number.isFinite(numeric)
+    ? Math.max(minimum, Math.min(maximum, Math.floor(numeric)))
+    : fallback;
+}

webui/src/module.ts

@@ -1,8 +1,13 @@
 import { createElement, lazy } from "react";
-import type { CalendarPickerUiCapability, PlatformWebModule } from "@govoplan/core-webui";
+import type {
+  CalendarPickerUiCapability,
+  DashboardWidgetsUiCapability,
+  PlatformWebModule
+} from "@govoplan/core-webui";
 import "./styles/calendar.css";
 import { generatedTranslations } from "./i18n/generatedTranslations";
 import CalendarPicker from "./features/calendar/CalendarPicker";
+import UpcomingEventsWidget from "./features/calendar/UpcomingEventsWidget";
 
 const CalendarPage = lazy(() => import("./features/calendar/CalendarPage"));
 
@@ -13,6 +18,60 @@ const translations = {
 };
 
 const calendarPicker: CalendarPickerUiCapability = { CalendarPicker };
+const calendarDashboardWidgets: DashboardWidgetsUiCapability = {
+  widgets: [
+    {
+      id: "calendar.upcoming",
+      surfaceId: "calendar.widget.upcoming",
+      title: "Upcoming events",
+      description: "The next events across visible calendars.",
+      moduleId: "calendar",
+      category: "Planning",
+      order: 40,
+      defaultVisible: false,
+      defaultSize: "medium",
+      supportedSizes: ["medium", "wide"],
+      anyOf: eventRead,
+      refreshIntervalMs: 60_000,
+      defaultConfiguration: {
+        maxItems: 5,
+        daysAhead: 14,
+        showLocation: true
+      },
+      configurationFields: [
+        {
+          id: "maxItems",
+          label: "Maximum events",
+          kind: "number",
+          min: 1,
+          max: 12,
+          step: 1,
+          required: true
+        },
+        {
+          id: "daysAhead",
+          label: "Days ahead",
+          kind: "number",
+          min: 1,
+          max: 90,
+          step: 1,
+          required: true
+        },
+        {
+          id: "showLocation",
+          label: "Show locations",
+          kind: "boolean"
+        }
+      ],
+      render: ({ settings, refreshKey, configuration }) =>
+        createElement(UpcomingEventsWidget, {
+          settings,
+          refreshKey,
+          configuration
+        })
+    }
+  ]
+};
 
 export const calendarModule: PlatformWebModule = {
   id: "calendar",
@@ -21,11 +80,21 @@ export const calendarModule: PlatformWebModule = {
   dependencies: ["access"],
   optionalDependencies: ["mail", "tasks", "scheduling", "appointments", "workflow", "notifications", "dms", "connectors"],
   translations,
+  viewSurfaces: [
+    {
+      id: "calendar.widget.upcoming",
+      moduleId: "calendar",
+      kind: "section",
+      label: "Upcoming events widget",
+      order: 40
+    }
+  ],
   navItems: [{ to: "/calendar", label: "i18n:govoplan-calendar.calendar.adab5090", iconName: "calendar", anyOf: eventRead, order: 55 }],
   routes: [
   { path: "/calendar", anyOf: eventRead, order: 55, render: ({ settings, auth }) => createElement(CalendarPage, { settings, auth }) }],
   uiCapabilities: {
-    "calendar.picker": calendarPicker
+    "calendar.picker": calendarPicker,
+    "dashboard.widgets": calendarDashboardWidgets
   }
 
 };