"""FlowDeck — v7.1.0 Calendar sync + Meeting Notes.

Covers migration 27, calendar link CRUD (encryption, auth, isolation),
bidirectional Google sync (pull/push/idempotence/conflict LWW + notif),
CalDAV XML parsing, free/busy, meeting audio upload + manual transcript +
offline AI summary firing ``meeting.summarized``.
"""
from __future__ import annotations

import json
import secrets

import pytest
from conftest import anon_csrf

from app.db import get_conn
from app.services import calendar_sync as cal

# ── helpers ────────────────────────────────────────────────────────────────

def _login(client):
    from app.auth.session import SessionManager
    login = f"v71_{secrets.token_hex(4)}"
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'V71', ?, 0)",
            (login, f"{login}@test.com"),
        )
        uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
        conn.commit()
    return SessionManager.create_session({"id": uid, "login": login}), uid


def _cookies(session):
    return {"flowdeck_session": session}


def _mkcollection(client, name="Sprint Cal"):
    r = client.post("/db/api", json={"name": name,
                                     "schema": [{"name": "Due", "type": "date"}]})
    assert r.status_code == 200, r.text
    return r.json()["id"]


def _date_prop_id(cid):
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM collection_properties WHERE collection_id=?"
                           " AND prop_type='date'", (cid,)).fetchone()
    return str(row["id"])


def _mkrow(cid, title, day, external_id=""):
    pid = _date_prop_id(cid)
    with get_conn() as conn:
        cur = conn.execute(
            """INSERT INTO collection_pages (collection_id, title, position,
                                             property_values_json, external_event_id)
               VALUES (?,?,0,?,?)""",
            (cid, title, json.dumps({pid: day}), external_id))
        conn.commit()
        return cur.lastrowid


def _mkpage(title="Meeting"):
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format)"
            " VALUES ('test', ?, '', 'blocks')", (title,))
        conn.commit()
        return cur.lastrowid


def _mklink(client, session, cid, provider="google", creds=None, **kw):
    body = {"provider": provider, "collection_id": cid,
            "credentials": creds or {"access_token": "tok123"}}
    body.update(kw)
    r = client.post("/api/v2/calendar-links", json=body, cookies=_cookies(session))
    assert r.status_code == 201, r.text
    return r.json()


# ── migration ──────────────────────────────────────────────────────────────

def test_migration_27_tables(client):
    with get_conn() as conn:
        tables = {r[0] for r in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
    for t in ("calendar_links", "meeting_transcripts"):
        assert t in tables
    with get_conn() as conn:
        cols = {r[1] for r in conn.execute(
            "PRAGMA table_info(collection_pages)").fetchall()}
    assert "external_event_id" in cols
    with get_conn() as conn:
        v = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
    assert v >= 27


# ── links CRUD ─────────────────────────────────────────────────────────────

def test_link_crud_and_encryption(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    link = _mklink(client, session, cid)
    lid = link["id"]
    assert "tokens_enc" not in link  # never leaked
    with get_conn() as conn:
        stored = conn.execute("SELECT tokens_enc FROM calendar_links WHERE id=?",
                              (lid,)).fetchone()[0]
    assert "tok123" not in stored  # encrypted at rest
    assert cal._decrypt_tokens(stored)["access_token"] == "tok123"
    r = client.get("/api/v2/calendar-links", cookies=_cookies(session))
    assert any(x["id"] == lid for x in r.json()["links"])
    r = client.delete(f"/api/v2/calendar-links/{lid}", cookies=_cookies(session))
    assert r.status_code == 200
    r = client.get("/api/v2/calendar-links", cookies=_cookies(session))
    assert r.json()["links"] == []


def test_link_validation_and_auth(client):
    anon_csrf(client)
    session, _ = _login(client)
    cid = _mkcollection(client)
    r = client.post("/api/v2/calendar-links",
                    json={"provider": "exchange", "collection_id": cid,
                          "credentials": {}},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post("/api/v2/calendar-links",
                    json={"provider": "caldav", "collection_id": cid,
                          "credentials": {}},
                    cookies=_cookies(session))
    assert r.status_code == 400  # url required
    r = client.post("/api/v2/calendar-links",
                    json={"provider": "google", "collection_id": 999999,
                          "credentials": {"access_token": "x"}},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post("/api/v2/calendar-links",
                    json={"provider": "google", "collection_id": cid,
                          "credentials": {"access_token": "x"}})
    assert r.status_code == 401
    # isolation: another user cannot delete the link
    link = _mklink(client, session, cid)
    s2, _ = _login(client)
    r = client.delete(f"/api/v2/calendar-links/{link['id']}", cookies=_cookies(s2))
    assert r.status_code == 404


# ── sync: pull / push / idempotence / conflicts ────────────────────────────

@pytest.mark.asyncio
async def test_sync_pull_creates_rows(client, monkeypatch):
    session, _ = _login(client)
    cid = _mkcollection(client)
    link = _mklink(client, session, cid)

    async def fake_list(tokens, calendar_id, tmin, tmax):
        assert tokens["access_token"] == "tok123"
        return [{"id": "g1", "title": "Kickoff", "start": "2026-10-06",
                 "description": "", "updated": "2026-09-28T10:00:00Z"},
                {"id": "g2", "title": "Demo", "start": "2026-10-07T14:00:00",
                 "description": "", "updated": "2026-09-28T10:00:00Z"}]
    monkeypatch.setattr(cal, "google_list_events", fake_list)
    stats = await cal.sync_link(link["id"])
    assert stats == {"pulled": 2, "pushed": 0, "conflicts": 0}
    with get_conn() as conn:
        rows = conn.execute("SELECT title, external_event_id, property_values_json"
                            " FROM collection_pages WHERE collection_id=?", (cid,)).fetchall()
    assert {r["external_event_id"] for r in rows} == {"g1", "g2"}
    pid = _date_prop_id(cid)
    vals = json.loads([r for r in rows if r["title"] == "Kickoff"][0]["property_values_json"])
    assert vals[pid] == "2026-10-06"
    # second pass: idempotent
    stats = await cal.sync_link(link["id"])
    assert stats["pulled"] == 0 and stats["conflicts"] == 0


@pytest.mark.asyncio
async def test_sync_push_new_local_row(client, monkeypatch):
    session, _ = _login(client)
    cid = _mkcollection(client)
    link = _mklink(client, session, cid)
    pushed = []

    async def fake_list(tokens, calendar_id, tmin, tmax):
        return []
    async def fake_push(tokens, calendar_id, event, remote_id=""):
        pushed.append((event, remote_id))
        return "g9"
    monkeypatch.setattr(cal, "google_list_events", fake_list)
    monkeypatch.setattr(cal, "google_push_event", fake_push)
    _mkrow(cid, "Local task", "2026-10-08")
    stats = await cal.sync_link(link["id"])
    assert stats["pushed"] == 1
    assert pushed[0][0]["title"] == "Local task"
    with get_conn() as conn:
        xid = conn.execute("SELECT external_event_id FROM collection_pages"
                           " WHERE collection_id=? AND title='Local task'",
                           (cid,)).fetchone()[0]
    assert xid == "g9"


@pytest.mark.asyncio
async def test_sync_conflict_lww_and_notif(client, monkeypatch):
    session, uid = _login(client)
    cid = _mkcollection(client)
    link = _mklink(client, session, cid)
    rid = _mkrow(cid, "Planning", "2026-10-05", external_id="g5")
    # local edit after link's last_sync
    pid = _date_prop_id(cid)
    with get_conn() as conn:
        conn.execute("UPDATE collection_pages SET property_values_json=?,"
                     " updated_at='2026-09-28 12:00:00' WHERE id=?",
                     (json.dumps({pid: "2026-10-09"}), rid))
        conn.execute("UPDATE calendar_links SET last_sync='2026-09-28 11:00:00' WHERE id=?",
                     (link["id"],))
        conn.commit()

    async def fake_list(tokens, calendar_id, tmin, tmax):
        return [{"id": "g5", "title": "Planning", "start": "2026-10-06",
                 "description": "", "updated": "2026-09-28T13:00:00Z"}]  # remote newer
    monkeypatch.setattr(cal, "google_list_events", fake_list)
    stats = await cal.sync_link(link["id"])
    assert stats["conflicts"] == 1
    with get_conn() as conn:
        vals = json.loads(conn.execute("SELECT property_values_json FROM collection_pages"
                                       " WHERE id=?", (rid,)).fetchone()[0])
        notif = conn.execute("SELECT * FROM notifications WHERE user_id=? AND ntype='calendar'",
                             (uid,)).fetchone()
    assert vals[pid] == "2026-10-06"  # remote (newer) won
    assert notif is not None


@pytest.mark.asyncio
async def test_sync_expired_token_maps_502(client, monkeypatch):
    session, _ = _login(client)
    cid = _mkcollection(client)
    link = _mklink(client, session, cid)

    async def fake_list(tokens, calendar_id, tmin, tmax):
        raise cal.SyncError("google token expired — relink the calendar")
    monkeypatch.setattr(cal, "google_list_events", fake_list)
    r = client.post(f"/api/v2/calendar-links/{link['id']}/sync",
                    cookies=_cookies(session))
    assert r.status_code == 502


def test_caldav_parser_unit(client):
    xml = """<D:multistatus xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
<D:response><D:href>/cal/abc.ics</D:href>
<D:propstat><D:prop><C:calendar-data>BEGIN:VCALENDAR
UID:evt-1
DTSTART:20261006T090000Z
SUMMARY:Standup
DESCRIPTION:daily sync
END:VCALENDAR</C:calendar-data></D:prop></D:propstat></D:response>
</D:multistatus>"""
    events = cal._parse_caldav_events(xml)
    assert len(events) == 1
    assert events[0]["id"] == "evt-1"
    assert events[0]["title"] == "Standup"
    assert events[0]["start"] == "2026-10-06"


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

def test_freebusy_basic(client):
    session, _ = _login(client)
    cid = _mkcollection(client)
    _mkrow(cid, "Busy task", "2026-10-05")  # a Monday
    r = client.get(f"/db/{cid}/calendar/freebusy?from=2026-10-05&to=2026-10-07",
                   cookies=_cookies(session))
    assert r.status_code == 200, r.text
    body = r.json()
    by_date = {d["date"]: d for d in body["days"]}
    assert by_date["2026-10-05"]["busy"] is True
    assert by_date["2026-10-06"]["busy"] is False
    assert "2026-10-06" in body["free_weekdays"]
    assert "2026-10-05" not in body["free_weekdays"]


def test_freebusy_validation(client):
    anon_csrf(client)
    session, _ = _login(client)
    cid = _mkcollection(client)
    r = client.get(f"/db/{cid}/calendar/freebusy?from=2026-10-07&to=2026-10-01",
                   cookies=_cookies(session))
    assert r.status_code == 400
    r = client.get("/db/999999/calendar/freebusy?from=2026-10-01&to=2026-10-02",
                   cookies=_cookies(session))
    assert r.status_code == 400
    r = client.get(f"/db/{cid}/calendar/freebusy?from=2026-10-01&to=2026-10-02")
    assert r.status_code == 401


# ── meetings ───────────────────────────────────────────────────────────────

def test_meeting_upload_and_manual_flow(client):
    session, uid = _login(client)
    pid = _mkpage()
    # audio only, no STT backend → stored, not transcribed
    r = client.post("/api/v2/meetings/transcribe",
                    files={"audio": ("rec.mp3", b"ID3" + b"\x00" * 100, "audio/mpeg")},
                    data={"page_id": str(pid)},
                    cookies=_cookies(session))
    assert r.status_code == 201, r.text
    tid = r.json()["id"]
    assert r.json()["transcribed"] is False
    # manual transcript from client
    r = client.post(f"/api/v2/meetings/transcripts/{tid}/text",
                    json={"transcript": "We decided to ship on Friday. Alice owns the release."},
                    cookies=_cookies(session))
    assert r.status_code == 200
    # automation catches meeting.summarized
    r = client.post("/workspace/automations",
                    json={"name": "Post-meeting", "trigger_type": "event",
                          "event": "page.created", "actions": []},
                    cookies=_cookies(session))
    aid = r.json()["id"]
    client.post(f"/workspace/automations/{aid}/steps",
                json={"kind": "trigger", "config": {"event": "meeting.summarized"}},
                cookies=_cookies(session))
    client.post(f"/workspace/automations/{aid}/steps",
                json={"kind": "action",
                      "config": {"type": "notify", "message": "recap ready"}},
                cookies=_cookies(session))
    r = client.post(f"/api/v2/meetings/transcripts/{tid}/summarize",
                    cookies=_cookies(session))
    assert r.status_code == 200, r.text
    assert r.json()["summary"]
    assert r.json()["offline"] is True
    with get_conn() as conn:
        runs = conn.execute("SELECT * FROM automation_runs WHERE automation_id=?",
                            (aid,)).fetchall()
    assert len(runs) == 1 and runs[0]["status"] == "fired"


def test_meeting_upload_validation(client):
    session, _ = _login(client)
    pid = _mkpage()
    r = client.post("/api/v2/meetings/transcribe",
                    files={"audio": ("rec.exe", b"data", "application/octet-stream")},
                    data={"page_id": str(pid)},
                    cookies=_cookies(session))
    assert r.status_code == 400
    r = client.post("/api/v2/meetings/transcribe",
                    data={"page_id": str(pid), "transcript": "hello"},
                    cookies=_cookies(session))
    assert r.status_code == 201  # manual-only transcript allowed (client-side STT)


def test_meeting_transcribe_inline_manual(client):
    session, _ = _login(client)
    pid = _mkpage()
    r = client.post("/api/v2/meetings/transcribe",
                    files={"audio": ("rec.wav", b"RIFF" + b"\x00" * 50, "audio/wav")},
                    data={"page_id": str(pid),
                          "transcript": "Inline notes from the call."},
                    cookies=_cookies(session))
    assert r.status_code == 201
    assert r.json()["transcribed"] is True


def test_meeting_summarize_empty_400(client):
    session, _ = _login(client)
    pid = _mkpage()
    r = client.post("/api/v2/meetings/transcribe",
                    files={"audio": ("rec.mp3", b"ID3abc", "audio/mpeg")},
                    data={"page_id": str(pid)},
                    cookies=_cookies(session))
    tid = r.json()["id"]
    r = client.post(f"/api/v2/meetings/transcripts/{tid}/summarize",
                    cookies=_cookies(session))
    assert r.status_code == 400


def test_meeting_page_not_found(client):
    session, _ = _login(client)
    r = client.post("/api/v2/meetings/transcribe",
                    files={"audio": ("rec.mp3", b"ID3abc", "audio/mpeg")},
                    data={"page_id": "999999"},
                    cookies=_cookies(session))
    assert r.status_code == 404