"""HEL-972: what the Meetings page and Meeting history show."""

from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy.exc import IntegrityError

from app.meetings.models import (
    Meeting,
    MeetingMember,
    MeetingStatus,
    MemberRole,
    Room,
    RoomStandingCoHost,
    RoomStatus,
)
from app.meetings.service import HISTORY_PAGE_SIZE
from tests.conftest import session_headers

ME = "674"  # the test token's user (sub), organisation 26
T0 = datetime(2026, 11, 3, 9, 0, tzinfo=UTC)


@pytest.fixture
async def headers(client, db):
    return await session_headers(client)


def _meeting(db, meeting_id, *, org="26", product="helixtahr", minutes=0, **fields):
    defaults = {
        "title": "Instant meeting",
        "owner_user_id": "900",
        "host_user_id": "900",
        "host_name": "Someone Else",
        "created_at": T0 + timedelta(minutes=minutes),
        # A meeting that ran: its Host joined (unjoined ones are HEL-973's tests).
        "host_joined_at": T0 + timedelta(minutes=minutes),
    }
    meeting = Meeting(id=meeting_id, product=product, org_id=org, **(defaults | fields))
    db.add(meeting)
    return meeting


def _room(
    db,
    room_id,
    url,
    *,
    owner="900",
    name=None,
    org="26",
    product="helixtahr",
    standing=(),
    disabled=False,
):
    """A Permanent Meeting Room (HEL-1136), with standing Co-hosts."""
    room = Room(
        id=room_id,
        product=product,
        org_id=org,
        url=url,
        name=name or url,
        owner_user_id=owner,
        owner_name="Someone Else",
        created_from_meeting_id="m-origin",
        status=RoomStatus.DISABLED if disabled else RoomStatus.ACTIVE,
        disabled_at=T0 if disabled else None,
    )
    db.add(room)
    for user_id in standing:
        db.add(
            RoomStandingCoHost(
                room_id=room_id, product=product, org_id=org, user_id=user_id, user_name="Test"
            )
        )
    return room


def _session(db, meeting_id, room_id, **fields):
    """A session of a room: a meeting of kind room_session."""
    return _meeting(db, meeting_id, kind="room_session", room_id=room_id, **fields)


def _member(db, meeting_id, *, user=ME, role=MemberRole.PARTICIPANT, org="26", **fields):
    product = fields.pop("product", "helixtahr")
    db.add(
        MeetingMember(
            meeting_id=meeting_id,
            user_id=user,
            user_name="Test User",
            role=role,
            product=product,
            org_id=org,
            **fields,
        )
    )


def _ended(minutes=60):
    return {"status": MeetingStatus.ENDED, "ended_at": T0 + timedelta(minutes=minutes)}


async def _ids(client, headers, path="/v1/meetings"):
    response = await client.get(path, headers=headers)
    assert response.status_code == 200, response.text
    return [row["id"] for row in response.json()["meetings"]]


# --- Meetings page --------------------------------------------------------------


async def test_no_meetings_gives_an_empty_list(client, headers):
    response = await client.get("/v1/meetings", headers=headers)

    assert response.json() == {"meetings": []}


async def test_row_shows_title_host_and_what_the_link_needs(client, db, headers):
    _meeting(db, "m-host", host_user_id=ME, owner_user_id=ME, host_name="Test User")
    _member(db, "m-host", role=MemberRole.HOST)
    await db.commit()

    response = await client.get("/v1/meetings", headers=headers)

    assert response.json() == {
        "meetings": [
            {
                "id": "m-host",
                "title": "Instant meeting",
                "host_name": "Test User",
                "kind": "meeting",
                "room_name": None,
            }
        ]
    }


async def test_meetings_the_user_hosts_co_hosts_or_joined_are_listed(client, db, headers):
    for meeting_id, role, admitted in [
        ("m-host", MemberRole.HOST, None),
        ("m-cohost", MemberRole.CO_HOST, None),
        ("m-joined", MemberRole.PARTICIPANT, T0),
    ]:
        _meeting(db, meeting_id)
        _member(db, meeting_id, role=role, first_admitted_at=admitted)
    await db.commit()

    assert set(await _ids(client, headers)) == {"m-host", "m-cohost", "m-joined"}


async def test_other_meetings_are_not_listed(client, db, headers):
    # AC-01.9: not hosting, not co-hosting, never admitted.
    _meeting(db, "m-stranger")
    _member(db, "m-stranger", user="901", role=MemberRole.HOST)
    _meeting(db, "m-only-waited")
    _member(db, "m-only-waited")
    await db.commit()

    assert await _ids(client, headers) == []


async def test_blocked_users_lose_the_meeting_but_removed_users_keep_it(client, db, headers):
    _meeting(db, "m-blocked")
    _member(db, "m-blocked", first_admitted_at=T0, blocked_at=T0)
    _meeting(db, "m-removed")
    _member(db, "m-removed", first_admitted_at=T0, removed_at=T0)
    await db.commit()

    assert await _ids(client, headers) == ["m-removed"]


async def test_ended_normal_meeting_leaves_the_list(client, db, headers):
    # AC-01.3
    _meeting(db, "m-ended", **_ended())
    _member(db, "m-ended", role=MemberRole.HOST)
    await db.commit()

    assert await _ids(client, headers) == []


async def test_permanent_rooms_show_for_owner_and_standing_co_hosts(client, db, headers):
    # AC-01.10 / AC-39.9: also between sessions (nothing running).
    _room(db, "r-owned", "harpreet", owner=ME, name="Harpreet's room")
    _room(db, "r-standing", "mgmt", name="Management meet", standing=[ME])
    await db.commit()

    response = await client.get("/v1/meetings", headers=headers)

    rows = response.json()["meetings"]
    assert [row["id"] for row in rows] == ["r-owned", "r-standing"]
    assert rows[1] == {
        "id": "r-standing",
        "title": "Management meet",
        "host_name": "Someone Else",
        "kind": "permanent_room",
        "room_name": "mgmt",
    }


async def test_permanent_rooms_hidden_from_everyone_else(client, db, headers):
    # Joined a session, or even co-hosted one, but not owner or standing Co-host.
    _room(db, "r-other", "standup")
    await db.flush()
    _session(db, "s-joined", "r-other")
    _member(db, "s-joined", first_admitted_at=T0)
    _session(db, "s-cohost", "r-other", **_ended())
    _member(db, "s-cohost", role=MemberRole.CO_HOST)
    await db.commit()

    assert await _ids(client, headers) == []


async def test_room_used_twice_is_one_row(client, db, headers):
    # AC-39.19: one row on Meetings however many sessions.
    _room(db, "r-team", "team", owner=ME)
    await db.flush()
    _session(db, "s-1", "r-team", owner_user_id=ME, host_user_id=ME, **_ended())
    _member(db, "s-1", role=MemberRole.HOST)
    _session(db, "s-2", "r-team", owner_user_id=ME, host_user_id=ME)
    _member(db, "s-2", role=MemberRole.HOST)
    await db.commit()

    assert await _ids(client, headers) == ["r-team"]


async def test_turned_off_room_is_hidden(client, db, headers):
    _room(db, "r-disabled", "old", owner=ME, disabled=True)
    await db.commit()

    assert await _ids(client, headers) == []


async def test_rooms_first_by_name_then_meetings_newest_first(client, db, headers):
    _room(db, "r-zeta", "zeta", owner=ME, name="zeta room")
    _room(db, "r-alpha", "alpha", owner=ME, name="Alpha room")
    for meeting_id, minutes in [("m-old", 0), ("m-new", 30), ("m-mid", 15)]:
        _meeting(db, meeting_id, minutes=minutes)
        _member(db, meeting_id, role=MemberRole.HOST)
    await db.commit()

    assert await _ids(client, headers) == ["r-alpha", "r-zeta", "m-new", "m-mid", "m-old"]


@pytest.mark.parametrize(
    ("org", "product"),
    [("30", "helixtahr"), ("26", "eazeesign")],
    ids=["other-organisation", "other-product"],
)
async def test_other_organisations_meetings_never_show(client, db, headers, org, product):
    # AC-01.5 / AC-35.4: same user ID, but in another organisation or product.
    _meeting(db, "m-elsewhere", org=org, product=product, owner_user_id=ME)
    _member(db, "m-elsewhere", role=MemberRole.HOST, org=org, product=product)
    _room(db, "r-elsewhere", "harpreet", owner=ME, org=org, product=product)
    await db.commit()

    assert await _ids(client, headers) == []
    assert await _ids(client, headers, "/v1/meetings/history") == []


async def test_room_urls_are_unique_within_an_organisation_only(db):
    # BR-39.3 / AC-39.5: among rooms that are on.
    _room(db, "r-1", "standup")
    _room(db, "r-2", "standup", org="30")
    _room(db, "r-old", "standup", disabled=True)
    await db.commit()

    _room(db, "r-3", "standup")
    with pytest.raises(IntegrityError):
        await db.commit()


# --- Meeting history ------------------------------------------------------------


async def test_history_shows_ended_meetings_with_times(client, db, headers):
    # AC-01.11
    _meeting(db, "m-ended", host_name="Akash Jio", **_ended(minutes=45))
    _member(db, "m-ended", first_admitted_at=T0)
    _meeting(db, "m-running")
    _member(db, "m-running", role=MemberRole.HOST)
    await db.commit()

    response = await client.get("/v1/meetings/history", headers=headers)

    assert response.json() == {
        "meetings": [
            {
                "id": "m-ended",
                "title": "Instant meeting",
                "host_name": "Akash Jio",
                "started_at": "2026-11-03T09:00:00Z",
                "ended_at": "2026-11-03T09:45:00Z",
            }
        ],
        "next_cursor": None,
    }


async def test_history_skips_meetings_the_user_was_never_part_of(client, db, headers):
    _meeting(db, "m-only-waited", **_ended())
    _member(db, "m-only-waited")
    _meeting(db, "m-blocked", **_ended())
    _member(db, "m-blocked", first_admitted_at=T0, blocked_at=T0)
    _meeting(db, "m-stranger", **_ended())
    await db.commit()

    assert await _ids(client, headers, "/v1/meetings/history") == []


async def test_history_is_newest_first_one_page_at_a_time(client, db, headers):
    total = HISTORY_PAGE_SIZE + 5
    for i in range(total):
        _meeting(db, f"m-{i:02d}", **_ended(minutes=i))
        _member(db, f"m-{i:02d}", role=MemberRole.HOST)
    await db.commit()

    first = (await client.get("/v1/meetings/history", headers=headers)).json()
    second = (
        await client.get(
            "/v1/meetings/history", params={"cursor": first["next_cursor"]}, headers=headers
        )
    ).json()

    newest_first = [f"m-{i:02d}" for i in reversed(range(total))]
    assert [row["id"] for row in first["meetings"]] == newest_first[:HISTORY_PAGE_SIZE]
    assert [row["id"] for row in second["meetings"]] == newest_first[HISTORY_PAGE_SIZE:]
    assert second["next_cursor"] is None


@pytest.mark.parametrize("cursor", ["not-base64!", "Zm9v", "MjAyNi0xMS0wM3w"])
async def test_bad_history_cursor_is_refused(client, headers, cursor):
    response = await client.get("/v1/meetings/history", params={"cursor": cursor}, headers=headers)

    assert response.status_code == 400
    assert response.json()["error"]["code"] == "invalid_cursor"


async def test_meetings_need_a_session(client):
    for path in ("/v1/meetings", "/v1/meetings/history"):
        response = await client.get(path)
        assert response.status_code == 401


# --- Opening a meeting by its link ("invited") ---------------------------------


async def _member_row(db, meeting_id, user=ME):
    db.expire_all()
    return await db.get(MeetingMember, (meeting_id, user))


async def test_opening_a_link_shows_the_meeting_for_pre_join(client, db, headers):
    _meeting(db, "m-shared", host_name="Akash Jio")
    await db.commit()

    response = await client.get("/v1/meetings/m-shared", headers=headers)

    assert response.status_code == 200
    assert response.json() == {
        "id": "m-shared",
        "title": "Instant meeting",
        "host_name": "Akash Jio",
        "kind": "meeting",
        "room_name": None,
    }


async def test_opening_a_link_counts_as_invited(client, db, headers):
    _meeting(db, "m-shared")
    await db.commit()
    assert await _ids(client, headers) == []

    await client.get("/v1/meetings/m-shared", headers=headers)

    member = await _member_row(db, "m-shared")
    assert member.link_opened_at is not None
    assert member.first_admitted_at is None  # opening the link never admits
    assert await _ids(client, headers) == ["m-shared"]


async def test_opening_again_keeps_the_first_time_and_the_role(client, db, headers):
    _meeting(db, "m-mine", host_user_id=ME, owner_user_id=ME)
    _member(db, "m-mine", role=MemberRole.HOST, link_opened_at=T0)
    await db.commit()

    await client.get("/v1/meetings/m-mine", headers=headers)

    member = await _member_row(db, "m-mine")
    assert member.role == MemberRole.HOST
    assert member.link_opened_at == T0


async def test_invited_meeting_moves_to_history_when_it_ends(client, db, headers):
    _meeting(db, "m-was-shared", **_ended())
    _member(db, "m-was-shared", link_opened_at=T0)
    await db.commit()

    assert await _ids(client, headers, "/v1/meetings/history") == ["m-was-shared"]


LINK_INVALID = {
    "error": {"code": "link_invalid", "message": "This meeting link is no longer valid"}
}


@pytest.mark.parametrize("meeting_id", ["no-such-meet", "BAD!id", "x" * 40])
async def test_unknown_link_is_no_longer_valid(client, headers, meeting_id):
    response = await client.get(f"/v1/meetings/{meeting_id}", headers=headers)

    assert response.status_code == 404
    assert response.json() == LINK_INVALID


async def test_ended_meeting_link_is_no_longer_valid(client, db, headers):
    # AC-01.3
    _meeting(db, "m-ended", **_ended())
    _member(db, "m-ended", role=MemberRole.HOST)
    await db.commit()

    response = await client.get("/v1/meetings/m-ended", headers=headers)

    assert response.status_code == 404
    assert response.json() == LINK_INVALID


async def test_turned_off_room_address_is_no_longer_valid(client, db, headers):
    _room(db, "r-off", "old", disabled=True)
    await db.commit()

    response = await client.get("/v1/rooms/26/old", headers=headers)

    assert response.json() == LINK_INVALID


@pytest.mark.parametrize(
    ("org", "product"),
    [("30", "helixtahr"), ("26", "eazeesign")],
    ids=["other-organisation", "other-product"],
)
async def test_other_organisations_link_reveals_nothing(client, db, headers, org, product):
    # HEL-977 check 3, before the link check: even an ended meeting gives this answer.
    _meeting(db, "m-theirs", org=org, product=product, title="Payroll review")
    _meeting(db, "m-theirs-ended", org=org, product=product, **_ended())
    await db.commit()

    for meeting_id in ("m-theirs", "m-theirs-ended"):
        response = await client.get(f"/v1/meetings/{meeting_id}", headers=headers)
        assert response.status_code == 403
        assert response.json() == {
            "error": {
                "code": "cannot_join",
                "message": "You can't join this meeting. Your HelixtaHR account doesn't "
                "have access to this meeting.",
            }
        }
        assert "Payroll" not in response.text
        assert await _member_row(db, meeting_id) is None


async def test_blocked_user_cannot_open_the_link(client, db, headers):
    _meeting(db, "m-blocked")
    _member(db, "m-blocked", first_admitted_at=T0, blocked_at=T0)
    await db.commit()

    response = await client.get("/v1/meetings/m-blocked", headers=headers)

    assert response.status_code == 403
    assert response.json()["error"]["code"] == "blocked"


async def test_opening_a_room_session_does_not_put_the_room_on_the_list(client, db, headers):
    # Rooms show only for their owner and standing Co-hosts.
    _room(db, "r-team", "team", name="Team room")
    await db.flush()
    _session(db, "s-team", "r-team", title="Team room")
    await db.commit()

    response = await client.get("/v1/meetings/s-team", headers=headers)

    assert response.json()["kind"] == "permanent_room"
    assert response.json()["room_name"] == "team"
    assert await _ids(client, headers) == []


async def test_opening_a_link_needs_a_session(client):
    response = await client.get("/v1/meetings/m-anything")

    assert response.status_code == 401