flowdeck/app/services/calendar_sync.py

"""FlowDeck — external calendar sync (v7.1.0).

Bidirectional sync between a collection (date property) and an external
calendar: Google Calendar (REST) or any CalDAV server (raw REPORT/PUT, no
extra dependency). Tokens are Fernet-encrypted at rest.

Matching: ``collection_pages.external_event_id`` ↔ remote event id.
Conflicts (both sides changed since ``last_sync``): last-write-wins +
in-app ``calendar.conflict`` notification (manual edit resolves).

See ``docs/V71_Calendar_Meetings.md``.
"""
from __future__ import annotations

import asyncio
import json
import logging
import time
import uuid
from datetime import UTC, datetime, timedelta

import httpx

from app.db import get_conn

logger = logging.getLogger(__name__)

PROVIDERS = ("google", "caldav")
SYNC_LOOKBACK_DAYS = 30
SYNC_LOOKAHEAD_DAYS = 90


class SyncError(RuntimeError):
    """Raised when the remote calendar cannot be reached/authorized."""


# ── links ──────────────────────────────────────────────────────────────────

def _encrypt_tokens(creds: dict) -> str:
    from app.services.sso_provisioning import encrypt_secret
    return encrypt_secret(json.dumps(creds or {}))


def _decrypt_tokens(tokens_enc: str) -> dict:
    if not tokens_enc:
        return {}
    try:
        from app.services.sso_provisioning import decrypt_secret
        raw = decrypt_secret(tokens_enc)
        if raw:
            return json.loads(raw)
    except Exception:  # noqa: BLE001
        pass
    try:  # legacy plaintext (tests)
        data = json.loads(tokens_enc)
        return data if isinstance(data, dict) else {}
    except Exception:  # noqa: BLE001
        return {}


def save_link(user_id: int, provider: str, collection_id: int,
              credentials: dict, calendar_id: str = "primary",
              date_property: str = "") -> dict:
    if provider not in PROVIDERS:
        raise ValueError(f"provider must be google|caldav, got {provider!r}")
    with get_conn() as conn:
        if not conn.execute("SELECT id FROM collections WHERE id=?",
                            (collection_id,)).fetchone():
            raise ValueError("collection not found")
        cur = conn.execute(
            """INSERT INTO calendar_links
               (user_id, provider, tokens_enc, calendar_id, collection_id, date_property)
               VALUES (?,?,?,?,?,?)
               ON CONFLICT(user_id, provider, calendar_id) DO UPDATE SET
                 tokens_enc=excluded.tokens_enc, collection_id=excluded.collection_id,
                 date_property=excluded.date_property""",
            (user_id, provider, _encrypt_tokens(credentials),
             calendar_id or "primary", collection_id, date_property or ""))
        conn.commit()
        row = conn.execute(
            "SELECT * FROM calendar_links WHERE user_id=? AND provider=? AND calendar_id=?",
            (user_id, provider, calendar_id or "primary")).fetchone()
        _ = cur
    out = dict(row)
    out.pop("tokens_enc", None)
    return out


def list_links(user_id: int) -> list[dict]:
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT id, user_id, provider, calendar_id, collection_id,"
            " date_property, last_sync, created_at FROM calendar_links WHERE user_id=?"
            " ORDER BY id", (user_id,)).fetchall()
    return [dict(r) for r in rows]


def delete_link(user_id: int, link_id: int) -> bool:
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM calendar_links WHERE id=? AND user_id=?",
                           (link_id, user_id))
        conn.commit()
        return cur.rowcount > 0


def _load_link(link_id: int) -> dict | None:
    with get_conn() as conn:
        row = conn.execute("SELECT * FROM calendar_links WHERE id=?", (link_id,)).fetchone()
    return dict(row) if row else None


# ── remote I/O (module-level = monkeypatchable) ────────────────────────────

def _remote_event(eid: str, title: str, start: str, description: str = "",
                  updated: str = "") -> dict:
    return {"id": str(eid), "title": title or "Untitled", "start": start,
            "description": description or "", "updated": updated or ""}


async def google_list_events(tokens: dict, calendar_id: str,
                             time_min: str, time_max: str) -> list[dict]:
    access = tokens.get("access_token", "")
    if not access:
        raise SyncError("google link has no access_token — relink the calendar")
    url = (f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}"
           f"/events?singleEvents=true&orderBy=startTime"
           f"&timeMin={time_min}&timeMax={time_max}")
    async with httpx.AsyncClient(timeout=15) as client:
        resp = await client.get(url, headers={"Authorization": f"Bearer {access}"})
    if resp.status_code == 401:
        raise SyncError("google token expired — relink the calendar")
    if resp.status_code >= 400:
        raise SyncError(f"google returned HTTP {resp.status_code}")
    out = []
    for item in resp.json().get("items", []):
        start = (item.get("start") or {}).get("dateTime") or (item.get("start") or {}).get("date") or ""
        out.append(_remote_event(item.get("id", ""), item.get("summary", ""),
                                 start, item.get("description", ""),
                                 item.get("updated", "")))
    return out


async def google_push_event(tokens: dict, calendar_id: str, event: dict,
                            remote_id: str = "") -> str:
    access = tokens.get("access_token", "")
    if not access:
        raise SyncError("google link has no access_token — relink the calendar")
    body = {"summary": event.get("title", ""),
            "description": event.get("description", ""),
            "start": {"date": event.get("start", "")[:10]},
            "end": {"date": event.get("start", "")[:10]}}
    base = f"https://www.googleapis.com/calendar/v3/calendars/{calendar_id}/events"
    async with httpx.AsyncClient(timeout=15) as client:
        if remote_id:
            resp = await client.patch(f"{base}/{remote_id}",
                                      headers={"Authorization": f"Bearer {access}"}, json=body)
        else:
            resp = await client.post(base, headers={"Authorization": f"Bearer {access}"},
                                     json=body)
    if resp.status_code == 401:
        raise SyncError("google token expired — relink the calendar")
    if resp.status_code >= 400:
        raise SyncError(f"google returned HTTP {resp.status_code}")
    return str(resp.json().get("id", remote_id or ""))


_CALDAV_REPORT = """<?xml version="1.0" encoding="utf-8" ?>
<C:calendar-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
  <D:prop><D:getetag/><C:calendar-data/></D:prop>
  <C:filter><C:comp-filter name="VCALENDAR"><C:comp-filter name="VEVENT">
    <C:time-range start="{start}" end="{end}"/>
  </C:comp-filter></C:comp-filter></C:filter>
</C:calendar-query>"""


def _parse_caldav_events(xml_text: str) -> list[dict]:
    """Minimal multistatus → event parser (UID/SUMMARY/DTSTART/DESCRIPTION)."""
    import re
    import xml.etree.ElementTree as ET
    events = []
    try:
        root = ET.fromstring(xml_text)
    except ET.ParseError:
        return []
    ns = {"D": "DAV:", "C": "urn:ietf:params:xml:ns:caldav"}
    for resp in root.findall("D:response", ns):
        href = resp.findtext("D:href", default="", namespaces=ns)
        data_el = resp.find(".//{urn:ietf:params:xml:ns:caldav}calendar-data")
        if data_el is None or not data_el.text:
            continue
        ics = data_el.text
        uid = re.search(r"^UID:(.+)$", ics, re.M)
        summary = re.search(r"^SUMMARY:(.+)$", ics, re.M)
        dtstart = re.search(r"^DTSTART(?:;[^:]*)?:(.+)$", ics, re.M)
        desc = re.search(r"^DESCRIPTION:(.+)$", ics, re.M)
        events.append(_remote_event(
            (uid.group(1).strip() if uid else href.strip("/").split("/")[-1]),
            summary.group(1).strip() if summary else "Untitled",
            _ics_to_date(dtstart.group(1).strip()) if dtstart else "",
            desc.group(1).strip() if desc else ""))
    return events


def _ics_to_date(value: str) -> str:
    value = value.strip()
    if len(value) >= 8 and value[:8].isdigit():
        return f"{value[:4]}-{value[4:6]}-{value[6:8]}"
    return value[:10]


def _event_to_ics(uid: str, title: str, date: str, description: str = "") -> str:
    stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    day = (date or "")[:10].replace("-", "")
    return (f"BEGIN:VCALENDAR\r\nVERSION:2.0\r\nPRODID:-//FlowDeck//Sync//EN\r\n"
            f"BEGIN:VEVENT\r\nUID:{uid}\r\nDTSTAMP:{stamp}\r\nDTSTART;VALUE=DATE:{day}\r\n"
            f"SUMMARY:{title}\r\nDESCRIPTION:{description}\r\nEND:VEVENT\r\nEND:VCALENDAR\r\n")


async def caldav_list_events(creds: dict, time_min: str, time_max: str) -> list[dict]:
    url = creds.get("url", "")
    if not url:
        raise SyncError("caldav link needs a calendar url")
    auth = (creds.get("username", ""), creds.get("password", ""))
    body = _CALDAV_REPORT.format(
        start=time_min.replace("-", "").split("T")[0] + "T000000Z",
        end=time_max.replace("-", "").split("T")[0] + "T000000Z")
    async with httpx.AsyncClient(timeout=15, auth=auth if auth[0] else None) as client:
        resp = await client.request("REPORT", url, content=body,
                                    headers={"Depth": "1",
                                             "Content-Type": "application/xml"})
    if resp.status_code == 401:
        raise SyncError("caldav rejected credentials")
    if resp.status_code >= 400:
        raise SyncError(f"caldav returned HTTP {resp.status_code}")
    return _parse_caldav_events(resp.text)


async def caldav_push_event(creds: dict, event: dict, remote_id: str = "") -> str:
    url = (creds.get("url", "") or "").rstrip("/")
    if not url:
        raise SyncError("caldav link needs a calendar url")
    auth = (creds.get("username", ""), creds.get("password", ""))
    uid = remote_id or f"flowdeck-{uuid.uuid4().hex}@flowdeck"
    href = f"{url}/{uid}.ics" if not remote_id else (
        remote_id if remote_id.startswith("http") else f"{url}/{remote_id}")
    ics = _event_to_ics(uid.split("@")[0], event.get("title", ""),
                        event.get("start", ""), event.get("description", ""))
    async with httpx.AsyncClient(timeout=15, auth=auth if auth[0] else None) as client:
        resp = await client.put(href, content=ics, headers={"Content-Type": "text/calendar"})
    if resp.status_code >= 400:
        raise SyncError(f"caldav returned HTTP {resp.status_code}")
    return uid


# ── mapping + sync ─────────────────────────────────────────────────────────

def _date_prop_id(conn, collection_id: int, wanted: str = "") -> tuple[str, str] | None:
    props = conn.execute(
        "SELECT id, name FROM collection_properties WHERE collection_id=? ORDER BY position",
        (collection_id,)).fetchall()
    if wanted:
        for p in props:
            if str(p["id"]) == str(wanted) or p["name"] == wanted:
                return str(p["id"]), p["name"]
        return None
    for p in props:
        # prop_type lives in the row; fetch full rows only when needed
        full = conn.execute("SELECT prop_type FROM collection_properties WHERE id=?",
                            (p["id"],)).fetchone()
        if full and full["prop_type"] == "date":
            return str(p["id"]), p["name"]
    return None


def _row_date(values: dict, prop_id: str, prop_name: str) -> str:
    raw = values.get(prop_id, values.get(prop_name, ""))
    if isinstance(raw, dict):
        raw = raw.get("date") or raw.get("value") or ""
    return str(raw or "")


def _to_epoch(value: str | None) -> float:
    if not value:
        return 0.0
    text = str(value).strip()
    try:
        if text.endswith("Z"):
            dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
        else:
            dt = datetime.fromisoformat(text[:19] if "T" in text else text[:19])
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=UTC)
        return dt.timestamp()
    except Exception:  # noqa: BLE001
        try:
            return time.mktime(time.strptime(text[:10], "%Y-%m-%d"))
        except Exception:  # noqa: BLE001
            return 0.0


def _window() -> tuple[str, str]:
    now = datetime.now(UTC)
    start = (now - timedelta(days=SYNC_LOOKBACK_DAYS)).strftime("%Y-%m-%dT00:00:00Z")
    end = (now + timedelta(days=SYNC_LOOKAHEAD_DAYS)).strftime("%Y-%m-%dT00:00:00Z")
    return start, end


async def sync_link(link_id: int) -> dict:
    """One bidirectional sync pass. Returns {pulled, pushed, conflicts}."""
    link = _load_link(link_id)
    if not link:
        raise ValueError("link not found")
    creds = _decrypt_tokens(link.get("tokens_enc") or "")
    collection_id = link.get("collection_id")
    if not collection_id:
        raise ValueError("link has no collection")
    with get_conn() as conn:
        date_prop = _date_prop_id(conn, collection_id, link.get("date_property") or "")
    if not date_prop:
        raise ValueError("collection has no date property")
    prop_id, prop_name = date_prop

    tmin, tmax = _window()
    if link["provider"] == "google":
        remote = await google_list_events(creds, link.get("calendar_id") or "primary",
                                          tmin, tmax)
    else:
        remote = await caldav_list_events(creds, tmin, tmax)

    last_sync = _to_epoch(link.get("last_sync"))
    pulled = pushed = conflicts = 0
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT id, title, property_values_json, updated_at,"
            " COALESCE(external_event_id, '') AS xid FROM collection_pages"
            " WHERE collection_id=?", (collection_id,)).fetchall()
        local = {r["xid"]: dict(r) for r in rows if r["xid"]}
        seen_remote: set[str] = set()
        touched: set[int] = set()  # rows written by this pull pass — never push back

        for ev in remote:
            eid = ev.get("id", "")
            if not eid:
                continue
            seen_remote.add(eid)
            day = (ev.get("start") or "")[:10]
            if eid not in local:
                values: dict = {}
                values[prop_id] = day
                max_pos = conn.execute(
                    "SELECT COALESCE(MAX(position), -1)+1 FROM collection_pages"
                    " WHERE collection_id=?", (collection_id,)).fetchone()[0]
                conn.execute(
                    """INSERT INTO collection_pages
                       (collection_id, title, position, property_values_json, external_event_id)
                       VALUES (?,?,?,?,?)""",
                    (collection_id, ev.get("title") or "Untitled", max_pos,
                     json.dumps(values), eid))
                pulled += 1
                continue
            row = local[eid]
            try:
                values = json.loads(row["property_values_json"] or "{}")
            except (TypeError, json.JSONDecodeError):
                values = {}
            local_day = _row_date(values, prop_id, prop_name)[:10]
            remote_newer = _to_epoch(ev.get("updated")) > _to_epoch(row["updated_at"])
            local_dirty = _to_epoch(row["updated_at"]) > last_sync and local_day != day
            if remote_newer and local_dirty and local_day and day and local_day != day:
                # Conflict: both sides moved → last-write-wins + notify.
                if _to_epoch(ev.get("updated")) >= _to_epoch(row["updated_at"]):
                    values[prop_id] = day
                    conn.execute(
                        "UPDATE collection_pages SET property_values_json=?,"
                        " updated_at=CURRENT_TIMESTAMP WHERE id=?",
                        (json.dumps(values), row["id"]))
                    touched.add(row["id"])
                conflicts += 1
                _notify_conflict(conn, link, row, ev)
            elif day and day != local_day:
                values[prop_id] = day
                conn.execute(
                    "UPDATE collection_pages SET property_values_json=?,"
                    " updated_at=CURRENT_TIMESTAMP WHERE id=?",
                    (json.dumps(values), row["id"]))
                touched.add(row["id"])
                pulled += 1

        # Push local changes (created locally or edited after last_sync).
        for xid, row in local.items():
            if row["id"] in touched:
                continue
            if xid in seen_remote:
                # Edited locally since last sync and remote unchanged → push.
                if last_sync and _to_epoch(row["updated_at"]) > last_sync:
                    await _push(link, creds, row, prop_id, prop_name, xid)
                    pushed += 1
                continue
            # Remote deleted the event → drop the local id (keep the row).
            conn.execute("UPDATE collection_pages SET external_event_id='' WHERE id=?",
                         (row["id"],))
        # Rows never linked and recently touched → create remotely.
        fresh = conn.execute(
            "SELECT id, title, property_values_json, updated_at FROM collection_pages"
            " WHERE collection_id=? AND COALESCE(external_event_id, '')=''",
            (collection_id,)).fetchall()
        for row in fresh:
            try:
                values = json.loads(row["property_values_json"] or "{}")
            except (TypeError, json.JSONDecodeError):
                values = {}
            day = _row_date(values, prop_id, prop_name)[:10]
            if not day:
                continue
            new_id = await _push(link, creds, dict(row), prop_id, prop_name, "")
            conn.execute("UPDATE collection_pages SET external_event_id=? WHERE id=?",
                         (new_id, row["id"]))
            pushed += 1

        conn.execute("UPDATE calendar_links SET last_sync=CURRENT_TIMESTAMP WHERE id=?",
                     (link_id,))
        conn.commit()
    return {"pulled": pulled, "pushed": pushed, "conflicts": conflicts}


async def _push(link: dict, creds: dict, row: dict, prop_id: str,
               prop_name: str, remote_id: str) -> str:
    try:
        values = json.loads(row.get("property_values_json") or "{}")
    except (TypeError, json.JSONDecodeError):
        values = {}
    event = {"title": row.get("title") or "Untitled",
             "start": _row_date(values, prop_id, prop_name),
             "description": ""}
    if link["provider"] == "google":
        return await google_push_event(creds, link.get("calendar_id") or "primary",
                                       event, remote_id)
    return await caldav_push_event(creds, event, remote_id)


def _notify_conflict(conn, link: dict, row: dict, ev: dict) -> None:
    try:
        from app.services.notifications import create_notification
        create_notification(
            link["user_id"], link["user_id"], "calendar",
            "Calendar sync conflict",
            f"« {row.get('title') or 'Untitled'} » changed on both sides;"
            f" kept the newest ({ev.get('start', '')[:10]}). Edit the row to resolve.",
            resource_type="collection", resource_id=link.get("collection_id") or 0,
            url=f"/db/{link.get('collection_id')}", conn=conn, commit=False)
    except Exception:  # noqa: BLE001 — notify must never break sync
        pass


async def calendar_sync_scheduler(interval_seconds: int = 900) -> None:
    """Background loop: sync every link with a collection (15 min default)."""
    while True:
        try:
            with get_conn() as conn:
                ids = [r["id"] for r in conn.execute(
                    "SELECT id FROM calendar_links WHERE collection_id IS NOT NULL"
                ).fetchall()]
            for link_id in ids:
                try:
                    await sync_link(link_id)
                except Exception as exc:  # noqa: BLE001 — one link must not kill the loop
                    logger.debug("calendar sync link %s failed: %s", link_id, exc)
        except Exception as exc:  # noqa: BLE001
            logger.debug("calendar_sync_scheduler: %s", exc)
        await asyncio.sleep(interval_seconds)


# ── free/busy ──────────────────────────────────────────────────────────────

def freebusy(collection_id: int, date_from: str, date_to: str,
             date_property: str = "") -> dict:
    """Busy/free weekdays in [date_from, date_to] (day granularity).

    Expands recurrence rules server-side (``recurrence.expand_rule``).
    """
    from app.services import recurrence as _rec
    try:
        start = datetime.strptime(date_from[:10], "%Y-%m-%d").date()
        end = datetime.strptime(date_to[:10], "%Y-%m-%d").date()
    except ValueError:
        raise ValueError("use YYYY-MM-DD dates") from None
    if end < start or (end - start).days > 370:
        raise ValueError("range must be 1..370 days")
    with get_conn() as conn:
        date_prop = _date_prop_id(conn, collection_id, date_property)
        if not date_prop:
            raise ValueError("collection has no date property")
        prop_id, prop_name = date_prop
        rows = conn.execute(
            "SELECT property_values_json FROM collection_pages WHERE collection_id=?",
            (collection_id,)).fetchall()
    busy: set[str] = set()
    for r in rows:
        try:
            values = json.loads(r["property_values_json"] or "{}")
        except (TypeError, json.JSONDecodeError):
            continue
        base = _row_date(values, prop_id, prop_name)
        if not base:
            continue
        rec = (values.get("__recurrence__") or {})
        rule = rec.get(prop_id) or rec.get(prop_name)
        if rule:
            try:
                for occ in _rec.expand_rule(
                        base, rule, start.isoformat(), end.isoformat()):
                    busy.add(occ[:10])
            except Exception:  # noqa: BLE001 — bad rule, use base date only
                busy.add(base[:10])
        else:
            if start.isoformat() <= base[:10] <= end.isoformat():
                busy.add(base[:10])
    days, free = [], []
    day = start
    while day <= end:
        iso = day.isoformat()
        days.append({"date": iso, "busy": iso in busy,
                     "weekend": day.weekday() >= 5})
        if iso not in busy and day.weekday() < 5:
            free.append(iso)
        day += timedelta(days=1)
    return {"collection_id": collection_id, "from": start.isoformat(),
            "to": end.isoformat(), "days": days, "free_weekdays": free}