webmail-client/plugins/webmail_plugin_calendar/webmail_plugin_calendar/sync.py

"""CalDAV pull sync (SPEC §A.3, read side).

Refreshes imported CalDAV calendars from their servers: skip when the collection's
ctag is unchanged, else pull every VEVENT and upsert it, dropping local events that
vanished remotely. Runs on connect (per calendar) and on a periodic cron (all).
Credentials come from the calendar's own vaulted ``secret_ct``.

Write-back (pushing local edits to the server) is a later stage; imported CalDAV
calendars are created ``read_only`` for now, so nothing here has to reconcile local
changes.
"""

from __future__ import annotations

import json
import logging
import uuid
from typing import Any

from sqlalchemy import text

from app.plugins.sdk import sdk
from webmail_plugin_calendar import caldav

logger = logging.getLogger("plg.calendar.sync")

_SDK = sdk("calendar")


def _creds(source_config: dict[str, Any], secret_ct: bytes | None) -> tuple[str, str]:
    username = str(source_config.get("username") or "")
    password = _SDK.vault.decrypt(secret_ct).decode("utf-8") if secret_ct else ""
    return username, password


async def _push_changes(
    session: Any, http: Any, cal_id: str, calendar_url: str, creds: tuple[str, str]
) -> int:
    """Push locally-changed events (``dav_dirty``) to the server. Deletions
    (``dav_deleted``) are DELETEd then dropped; other changes are PUT. Returns the
    number of resources pushed. Runs with ``webmail.dav_sync='on'`` so its own writes
    don't re-mark rows dirty."""
    rows = (
        await session.execute(
            text(
                "SELECT id, ical_uid, sequence, summary, description, location, dtstart, "
                "dtend, all_day, rrule, exdates, dav_href, dav_etag, dav_deleted "
                "FROM plg_calendar_events WHERE calendar_id = :cal AND dav_dirty = true"
            ),
            {"cal": cal_id},
        )
    ).mappings().all()
    pushed = 0
    for r in rows:
        try:
            if r["dav_deleted"]:
                if r["dav_href"]:
                    await caldav.delete_event(http, r["dav_href"], r["dav_etag"], creds)
                await session.execute(
                    text("DELETE FROM plg_calendar_events WHERE id = :id"), {"id": str(r["id"])}
                )
            else:
                href = r["dav_href"] or caldav.event_href(calendar_url, r["ical_uid"])
                ics = caldav.build_ics(dict(r))
                new_etag = await caldav.put_event(http, href, ics, r["dav_etag"], creds)
                await session.execute(
                    text(
                        "UPDATE plg_calendar_events SET dav_dirty = false, dav_href = :href, "
                        "dav_etag = :etag WHERE id = :id"
                    ),
                    {"href": href, "etag": new_etag, "id": str(r["id"])},
                )
            pushed += 1
        except caldav.CalDavError:
            # Conflict/transport error: leave the row dirty and try again next run.
            logger.exception("caldav push failed for event %s", r["id"])
    return pushed


async def sync_calendar(session: Any, cal: dict[str, Any]) -> int:
    """Sync one CalDAV calendar: push local changes, then pull the server into
    ``plg_calendar_events``. Returns events upserted on the pull."""
    cfg = cal["source_config"]
    if isinstance(cfg, str):
        cfg = json.loads(cfg)
    calendar_url = str(cfg.get("calendar_url") or "")
    if not calendar_url:
        return 0
    creds = _creds(cfg, cal["secret_ct"])
    http = _SDK.http

    # Suppress the dav_dirty trigger for every write we make in this transaction.
    await session.execute(text("SELECT set_config('webmail.dav_sync', 'on', true)"))
    pushed = await _push_changes(session, http, str(cal["id"]), calendar_url, creds)

    ctag = await caldav.get_ctag(http, calendar_url, creds)
    if not pushed and ctag and ctag == (cal.get("sync_token") or ""):
        return 0  # nothing pushed and collection unchanged since last sync

    events = await caldav.pull_events(http, calendar_url, creds)
    cal_id = str(cal["id"])
    user_id = str(cal["user_id"])
    seen: set[str] = set()
    upserted = 0
    for dav in events:
        fields = caldav.vevent_to_fields(dav.ics)
        if fields is None:
            continue
        seen.add(fields["ical_uid"])
        await session.execute(
            text(
                "INSERT INTO plg_calendar_events "
                "(user_id, calendar_id, ical_uid, sequence, summary, description, location, "
                " dtstart, dtend, all_day, rrule, status, raw_ical, dav_href, dav_etag) VALUES "
                "(:u, :cal, :uid, :seq, :sum, :desc, :loc, :start, :end, :allday, :rrule, "
                " :status, :raw, :href, :etag) "
                "ON CONFLICT (calendar_id, ical_uid, recurrence_id) DO UPDATE SET "
                "sequence = EXCLUDED.sequence, summary = EXCLUDED.summary, "
                "description = EXCLUDED.description, location = EXCLUDED.location, "
                "dtstart = EXCLUDED.dtstart, dtend = EXCLUDED.dtend, all_day = EXCLUDED.all_day, "
                "rrule = EXCLUDED.rrule, status = EXCLUDED.status, raw_ical = EXCLUDED.raw_ical, "
                "dav_href = EXCLUDED.dav_href, dav_etag = EXCLUDED.dav_etag "
                # Don't overwrite a locally-changed row that hasn't been pushed yet.
                "WHERE plg_calendar_events.dav_dirty = false"
            ),
            {
                "u": user_id,
                "cal": cal_id,
                "uid": fields["ical_uid"],
                "seq": fields["sequence"],
                "sum": fields["summary"],
                "desc": fields["description"],
                "loc": fields["location"],
                "start": fields["dtstart"],
                "end": fields["dtend"],
                "allday": fields["all_day"],
                "rrule": fields["rrule"],
                "status": fields["status"],
                "raw": fields["raw_ical"],
                "href": dav.href,
                "etag": dav.etag,
            },
        )
        upserted += 1

    # Drop local events that no longer exist on the server, but never touch rows with
    # unpushed local changes (dav_dirty) — including freshly-created ones not yet PUT.
    if seen:
        await session.execute(
            text(
                "DELETE FROM plg_calendar_events "
                "WHERE calendar_id = :cal AND dav_dirty = false "
                "AND NOT (ical_uid = ANY(:keep))"
            ),
            {"cal": cal_id, "keep": list(seen)},
        )
    else:
        await session.execute(
            text(
                "DELETE FROM plg_calendar_events "
                "WHERE calendar_id = :cal AND dav_dirty = false"
            ),
            {"cal": cal_id},
        )
    await session.execute(
        text("UPDATE plg_calendar_calendars SET sync_token = :ct WHERE id = :id"),
        {"ct": ctag, "id": cal_id},
    )
    return upserted


async def caldav_sync(ctx: dict[str, Any], calendar_id: str | None = None) -> None:
    """Job: sync one CalDAV calendar (``calendar_id`` given) or all of them (cron).

    Enumerates via a non-RLS session (cron spans users), then applies each calendar's
    upserts under that user's RLS session."""
    async with _SDK.public_session() as pub:
        where = "kind = 'caldav'"
        params: dict[str, Any] = {}
        if calendar_id:
            where += " AND id = :id"
            params["id"] = calendar_id
        rows = (
            await pub.execute(
                text(
                    "SELECT id, user_id, source_config, secret_ct, sync_token "
                    f"FROM plg_calendar_calendars WHERE {where}"
                ),
                params,
            )
        ).mappings().all()

    for cal in rows:
        try:
            async with _SDK.session(uuid.UUID(str(cal["user_id"]))) as s:
                n = await sync_calendar(s, dict(cal))
            logger.info("caldav sync: calendar %s upserted %d event(s)", cal["id"], n)
        except Exception:  # noqa: BLE001 - one bad calendar mustn't stop the rest
            logger.exception("caldav sync failed for calendar %s", cal["id"])