"""HEL-977 (join checks, 25 places), HEL-982 (waiting room, admit, deny) and
HEL-981 (Co-hosts): the backend."""

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

import pytest
from sqlalchemy import func, select, update

from app.audit.models import AuditRecord
from app.core.alerts import Alerts
from app.meetings import admission
from app.meetings.models import (
    AdmissionRequest,
    Meeting,
    MeetingDevice,
    MeetingMember,
    MeetingStatus,
    MemberRole,
)
from app.meetings.rooms import WAITING_TOPIC
from tests.conftest import session_headers
from tests.fakes import FAKE_WEBHOOK_SIGNATURE

HOST = "674"  # the test token's user: starts the meetings
HOST_DEVICE = "host-laptop-01"
FULL = {
    "code": "meeting_full",
    "message": "This meeting is full. The meeting has reached its limit of 25 participants.",
}


def _identity(user: str, device: str) -> str:
    return f"helixtahr:26:{user}:{device}"


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


@pytest.fixture
async def meeting_id(client, host):
    """A started meeting whose Host has joined on one device."""
    meeting_id = (await client.post("/v1/meetings", headers=host)).json()["id"]
    response = await _join(client, host, meeting_id, HOST_DEVICE)
    assert response.status_code == 200
    return meeting_id


async def _person(client, user: str):
    return await session_headers(client, sub=user, name=f"Person {user}")


async def _join(client, headers, meeting_id, device="tab-device-01"):
    return await client.post(
        f"/v1/meetings/{meeting_id}/join", headers=headers, json={"device_id": device}
    )


async def _status(client, headers, meeting_id):
    return await client.get(f"/v1/meetings/{meeting_id}/admission", headers=headers)


async def _waiting(client, headers, meeting_id):
    return await client.get(f"/v1/meetings/{meeting_id}/waiting", headers=headers)


async def _request_id(client, host, meeting_id, user):
    waiting = (await _waiting(client, host, meeting_id)).json()["waiting"]
    return next(row["request_id"] for row in waiting if row["user_id"] == user)


async def _decide(client, headers, meeting_id, request_id, decision):
    return await client.post(
        f"/v1/meetings/{meeting_id}/waiting/{request_id}/{decision}", headers=headers
    )


async def _admitted(client, host, meeting_id, user="700"):
    """A colleague asks, is admitted and joins. Returns their headers."""
    person = await _person(client, user)
    assert (await _join(client, person, meeting_id)).status_code == 202
    request_id = await _request_id(client, host, meeting_id, user)
    assert (await _decide(client, host, meeting_id, request_id, "admit")).status_code == 204
    assert (await _join(client, person, meeting_id)).status_code == 200
    return person


async def _webhook(client, event, meeting_id, identity, connection="conn-1", signature=None):
    body = {"event": event, "room": meeting_id, "identity": identity, "connection_id": connection}
    return await client.post(
        "/livekit/webhook",
        content=json.dumps(body),
        headers={"Authorization": signature or FAKE_WEBHOOK_SIGNATURE},
    )


async def _fill(db, meeting_id, count, *, first_user=9000):
    """`count` admitted participants, each connected on one device."""
    now = datetime.now(UTC)
    for i in range(count):
        user = str(first_user + i)
        db.add(
            MeetingMember(
                meeting_id=meeting_id,
                product="helixtahr",
                org_id="26",
                user_id=user,
                user_name=f"Filler {user}",
                role=MemberRole.PARTICIPANT,
                first_admitted_at=now - timedelta(minutes=30),
                admitted_at=now - timedelta(minutes=30),
            )
        )
        db.add(
            MeetingDevice(
                meeting_id=meeting_id,
                product="helixtahr",
                org_id="26",
                identity=_identity(user, "filler"),
                user_id=user,
                device_id="filler",
                pass_issued_at=now - timedelta(minutes=30),
                connected_at=now - timedelta(minutes=29),
            )
        )
    await db.commit()


async def _all(db, model, *where):
    db.expire_all()
    return list(await db.scalars(select(model).where(*where)))


async def _device_left(db, meeting_id, user, minutes_ago):
    await db.execute(
        update(MeetingDevice)
        .where(MeetingDevice.meeting_id == meeting_id, MeetingDevice.user_id == user)
        .values(
            connected_at=func.now() - timedelta(minutes=minutes_ago + 1),
            left_at=func.now() - timedelta(minutes=minutes_ago),
            pass_issued_at=func.now() - timedelta(minutes=minutes_ago + 2),
        )
    )
    await db.commit()


# --- Asking to join (HEL-977 / HEL-982) ------------------------------------


async def test_colleague_waits_and_hosts_are_told(client, db, meeting_id, fake_livekit):
    colleague = await _person(client, "700")

    response = await _join(client, colleague, meeting_id)

    assert response.status_code == 202
    assert response.json() == {"status": "waiting"}
    # AC-10.1: a data message to the Host's device: pop-up + People badge.
    [sent] = fake_livekit.sent
    assert sent.room == meeting_id
    assert sent.identities == [_identity(HOST, HOST_DEVICE)]
    assert (sent.topic, sent.message) == (WAITING_TOPIC, {"waiting": 1})
    # AC-10.9: waiting people never get a pass.
    assert [issued.identity for issued in fake_livekit.passes] == [_identity(HOST, HOST_DEVICE)]


async def test_host_sees_who_is_waiting_longest_first(client, db, meeting_id, host):
    for user in ("701", "702"):
        await _join(client, await _person(client, user), meeting_id)

    response = await _waiting(client, host, meeting_id)

    assert response.status_code == 200
    rows = response.json()["waiting"]
    assert [(row["user_id"], row["name"]) for row in rows] == [
        ("701", "Person 701"),
        ("702", "Person 702"),
    ]


async def test_asking_again_keeps_one_request(client, db, meeting_id, fake_livekit):
    colleague = await _person(client, "700")

    await _join(client, colleague, meeting_id)
    await _join(client, colleague, meeting_id, device="other-tab-02")

    assert len(await _all(db, AdmissionRequest)) == 1
    assert len(fake_livekit.sent) == 1  # told once


async def test_waiting_screen_sees_waiting(client, db, meeting_id):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)

    response = await _status(client, colleague, meeting_id)

    assert response.json() == {"status": "waiting", "message": None}


async def test_nobody_waiting_has_no_status(client, db, meeting_id):
    response = await _status(client, await _person(client, "700"), meeting_id)

    assert response.json()["status"] == "none"


async def test_people_keep_waiting_when_no_host_is_there(client, db, host):
    # AC-10.4: started but the Host hasn't joined: nobody can admit; they wait.
    meeting_id = (await client.post("/v1/meetings", headers=host)).json()["id"]
    colleague = await _person(client, "700")

    assert (await _join(client, colleague, meeting_id)).status_code == 202
    assert (await _status(client, colleague, meeting_id)).json()["status"] == "waiting"


async def test_leaving_the_waiting_screen_removes_the_entry(
    client, db, meeting_id, host, fake_livekit
):
    # AC-10.7
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)

    response = await client.delete(f"/v1/meetings/{meeting_id}/admission", headers=colleague)

    assert response.status_code == 204
    assert (await _waiting(client, host, meeting_id)).json()["waiting"] == []
    assert fake_livekit.sent[-1].message == {"waiting": 0}


async def test_waiting_screen_that_stops_asking_is_gone_after_30_seconds(
    client, db, meeting_id, host
):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    await db.execute(
        update(AdmissionRequest).values(last_seen_at=func.now() - timedelta(seconds=31))
    )
    await db.commit()

    assert (await _waiting(client, host, meeting_id)).json()["waiting"] == []
    assert (await _status(client, colleague, meeting_id)).json()["status"] == "left"
    # Joining again puts them back in the waiting room.
    assert (await _join(client, colleague, meeting_id)).status_code == 202
    assert len((await _waiting(client, host, meeting_id)).json()["waiting"]) == 1


async def test_background_clean_up_marks_gone_requests(db, client, meeting_id):
    await _join(client, await _person(client, "700"), meeting_id)
    await db.execute(
        update(AdmissionRequest).values(last_seen_at=func.now() - timedelta(seconds=31))
    )
    await db.commit()

    assert await admission.forget_gone(db) == [meeting_id]
    [request] = await _all(db, AdmissionRequest)
    assert request.status == "left"


async def test_waiting_screen_learns_the_meeting_ended(client, db, meeting_id):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    await db.execute(update(Meeting).values(status=MeetingStatus.ENDED, ended_at=func.now()))
    await db.commit()

    response = await _status(client, colleague, meeting_id)

    # AC-24.9: released, with the ended message (not "no longer valid").
    assert response.status_code == 200
    assert response.json() == {"status": "ended", "message": "The meeting has ended"}


# --- Admit and deny (HEL-982) ----------------------------------------------


async def test_admitted_person_joins_with_the_mic_off(client, db, meeting_id, host, fake_livekit):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    request_id = await _request_id(client, host, meeting_id, "700")

    assert (await _decide(client, host, meeting_id, request_id, "admit")).status_code == 204
    assert (await _status(client, colleague, meeting_id)).json()["status"] == "admitted"
    joined = await _join(client, colleague, meeting_id)

    # AC-10.2
    assert joined.status_code == 200
    assert joined.json()["role"] == "participant"
    assert joined.json()["microphone_on"] is False
    issued = fake_livekit.passes[-1]
    assert issued.identity == _identity("700", "tab-device-01")
    assert issued.attributes == {"role": "participant", "user_id": "700"}
    assert fake_livekit.sent[-1].message == {"waiting": 0}
    # Now "joined": the meeting is on their Meetings page.
    listed = (await client.get("/v1/meetings", headers=colleague)).json()["meetings"]
    assert [row["id"] for row in listed] == [meeting_id]


async def test_admit_writes_an_audit_record(client, db, meeting_id, host):
    await _admitted(client, host, meeting_id)

    [record] = await _all(db, AuditRecord, AuditRecord.action == "participant_admitted")
    assert (record.meeting_id, record.actor_user_id) == (meeting_id, HOST)
    assert record.details == {"user_id": "700", "name": "Person 700"}


async def test_denied_person_sees_the_message_and_can_ask_again(client, db, meeting_id, host):
    # AC-10.3 / BR-10.3
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    request_id = await _request_id(client, host, meeting_id, "700")

    assert (await _decide(client, host, meeting_id, request_id, "deny")).status_code == 204
    assert (await _status(client, colleague, meeting_id)).json() == {
        "status": "denied",
        "message": "Your request to join was declined",
    }
    assert (await _waiting(client, host, meeting_id)).json()["waiting"] == []
    [record] = await _all(db, AuditRecord, AuditRecord.action == "participant_denied")
    assert record.details["user_id"] == "700"

    assert (await _join(client, colleague, meeting_id)).status_code == 202
    assert (await _status(client, colleague, meeting_id)).json()["status"] == "waiting"


async def test_ordinary_participants_can_never_see_or_admit(client, db, meeting_id, host):
    # AC-10.5 / BR-10.1
    participant = await _admitted(client, host, meeting_id, "700")
    await _join(client, await _person(client, "701"), meeting_id)
    request_id = await _request_id(client, host, meeting_id, "701")

    listing = await _waiting(client, participant, meeting_id)
    admit = await _decide(client, participant, meeting_id, request_id, "admit")
    deny = await _decide(client, participant, meeting_id, request_id, "deny")

    for response in (listing, admit, deny):
        assert response.status_code == 403
        assert response.json()["error"]["code"] == "not_a_host"


async def test_admitting_someone_no_longer_waiting(client, db, meeting_id, host):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    request_id = await _request_id(client, host, meeting_id, "700")
    await client.delete(f"/v1/meetings/{meeting_id}/admission", headers=colleague)

    response = await _decide(client, host, meeting_id, request_id, "admit")

    assert response.status_code == 404
    assert response.json()["error"]["code"] == "request_gone"


async def test_another_organisation_cannot_see_the_waiting_list(client, db, meeting_id):
    outsider = await session_headers(client, company_id=27)

    response = await _waiting(client, outsider, meeting_id)

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


# --- 25 places (HEL-977) ---------------------------------------------------


async def test_26th_person_is_told_the_meeting_is_full(client, db, meeting_id, host):
    # AC-06.3: Host + 24 connected.
    await _fill(db, meeting_id, 24)

    response = await _join(client, await _person(client, "700"), meeting_id)

    assert response.status_code == 409
    assert response.json() == {"error": FULL}
    assert (await _waiting(client, host, meeting_id)).json()["waiting"] == []


async def test_admit_is_blocked_while_full(client, db, meeting_id, host):
    # AC-06.5 / AC-10.6: they waited before it filled up.
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    request_id = await _request_id(client, host, meeting_id, "700")
    await _fill(db, meeting_id, 24)

    blocked = await _decide(client, host, meeting_id, request_id, "admit")

    assert blocked.status_code == 409
    assert blocked.json() == {"error": FULL}
    assert (await _status(client, colleague, meeting_id)).json()["status"] == "waiting"

    # A participant leaves; their place is held for 2 minutes, then it frees.
    await _device_left(db, meeting_id, "9000", minutes_ago=1)
    assert (await _decide(client, host, meeting_id, request_id, "admit")).status_code == 409
    await _device_left(db, meeting_id, "9000", minutes_ago=3)
    assert (await _decide(client, host, meeting_id, request_id, "admit")).status_code == 204


async def test_two_admits_never_take_the_last_place_twice(
    client, db, meeting_id, host, monkeypatch
):
    # Host + 23 connected = 24: one place left, two people waiting, two Admits at once.
    # A pause after counting makes both Admits count at the same moment, unless the
    # meeting lock makes the second wait for the first.
    count = admission.count_places

    async def slow_count(session, meeting_id):
        places = await count(session, meeting_id)
        await asyncio.sleep(0.2)
        return places

    monkeypatch.setattr(admission, "count_places", slow_count)
    for user in ("700", "701"):
        await _join(client, await _person(client, user), meeting_id)
    first = await _request_id(client, host, meeting_id, "700")
    second = await _request_id(client, host, meeting_id, "701")
    await _fill(db, meeting_id, 23)

    responses = await asyncio.gather(
        _decide(client, host, meeting_id, first, "admit"),
        _decide(client, host, meeting_id, second, "admit"),
    )

    assert sorted(response.status_code for response in responses) == [204, 409]


async def test_host_always_gets_back_into_a_full_meeting(client, db, meeting_id, host):
    # AC-06.10: Host away, 24 others connected: the Host's place is still theirs.
    await _device_left(db, meeting_id, HOST, minutes_ago=30)
    await _fill(db, meeting_id, 24)

    assert (await _join(client, await _person(client, "700"), meeting_id)).status_code == 409
    response = await _join(client, host, meeting_id, HOST_DEVICE)

    assert response.status_code == 200


async def test_held_place_covers_one_device(client, db, meeting_id, host):
    await _fill(db, meeting_id, 24)

    response = await _join(client, host, meeting_id, "host-phone-02")

    assert response.status_code == 409


async def test_one_person_on_two_devices_takes_two_places(client, db, meeting_id, host):
    # AC-06.6
    person = await _admitted(client, host, meeting_id, "700")
    await _fill(db, meeting_id, 22)  # Host 1 + Person 700 1 + 22 = 24

    assert (await _join(client, person, meeting_id, "phone-device-2")).status_code == 200
    assert (await _join(client, person, meeting_id, "tablet-device-3")).status_code == 409
    devices = await _all(db, MeetingDevice, MeetingDevice.user_id == "700")
    assert len(devices) == 2


async def test_dropped_participant_comes_back_within_2_minutes_without_waiting(
    client, db, meeting_id, host, fake_livekit
):
    # AC-06.9
    person = await _admitted(client, host, meeting_id, "700")
    await _device_left(db, meeting_id, "700", minutes_ago=1)

    response = await _join(client, person, meeting_id)

    assert response.status_code == 200
    assert response.json()["microphone_on"] is False  # any reconnect is muted


async def test_after_2_minutes_they_wait_again(client, db, meeting_id, host):
    person = await _admitted(client, host, meeting_id, "700")
    await _device_left(db, meeting_id, "700", minutes_ago=3)
    await db.execute(
        update(MeetingMember)
        .where(MeetingMember.user_id == "700")
        .values(admitted_at=func.now() - timedelta(minutes=10))
    )
    await db.commit()

    assert (await _join(client, person, meeting_id)).status_code == 202


async def test_admission_is_used_within_2_minutes_or_lost(client, db, meeting_id, host):
    colleague = await _person(client, "700")
    await _join(client, colleague, meeting_id)
    request_id = await _request_id(client, host, meeting_id, "700")
    await _decide(client, host, meeting_id, request_id, "admit")
    await db.execute(
        update(MeetingMember)
        .where(MeetingMember.user_id == "700")
        .values(admitted_at=func.now() - timedelta(minutes=3))
    )
    await db.commit()

    assert (await _join(client, colleague, meeting_id)).status_code == 202


async def test_host_coming_back_is_muted(client, db, meeting_id, host):
    await _webhook(client, "participant_joined", meeting_id, _identity(HOST, HOST_DEVICE))
    await _device_left(db, meeting_id, HOST, minutes_ago=5)

    response = await _join(client, host, meeting_id, HOST_DEVICE)

    assert response.json()["microphone_on"] is False


async def test_livekit_down_at_join_alerts_engineering(
    client, db, meeting_id, host, fake_livekit, caplog
):
    fake_livekit.healthy = False

    with caplog.at_level(logging.ERROR, logger="app.alerts"):
        response = await _join(client, host, meeting_id, HOST_DEVICE)

    assert response.status_code == 503
    assert any("ALERT livekit unavailable" in r.getMessage() for r in caplog.records)


# --- LiveKit webhook -------------------------------------------------------


async def test_webhook_records_connect_and_leave(client, db, meeting_id):
    identity = _identity(HOST, HOST_DEVICE)

    assert (await _webhook(client, "participant_joined", meeting_id, identity)).status_code == 200
    [device] = await _all(db, MeetingDevice)
    assert device.connected_at is not None and device.left_at is None

    await _webhook(client, "participant_left", meeting_id, identity)
    [device] = await _all(db, MeetingDevice)
    assert device.left_at is not None


async def test_webhook_ignores_an_old_connection_leaving_after_a_reconnect(client, db, meeting_id):
    identity = _identity(HOST, HOST_DEVICE)
    await _webhook(client, "participant_joined", meeting_id, identity, connection="old")
    await _webhook(client, "participant_joined", meeting_id, identity, connection="new")

    await _webhook(client, "participant_left", meeting_id, identity, connection="old")

    [device] = await _all(db, MeetingDevice)
    assert (device.connection_id, device.left_at) == ("new", None)


async def test_webhook_needs_livekits_signature(client, db, meeting_id):
    identity = _identity(HOST, HOST_DEVICE)

    response = await _webhook(client, "participant_left", meeting_id, identity, signature="forged")

    assert response.status_code == 401
    [device] = await _all(db, MeetingDevice)
    assert device.left_at is None


# --- Co-hosts (HEL-981) ----------------------------------------------------


async def _make_co_host(client, headers, meeting_id, user):
    return await client.post(f"/v1/meetings/{meeting_id}/co-hosts/{user}", headers=headers)


async def test_host_makes_an_admitted_participant_co_host(
    client, db, meeting_id, host, fake_livekit
):
    await _admitted(client, host, meeting_id, "700")
    await _webhook(client, "participant_joined", meeting_id, _identity("700", "tab-device-01"))

    response = await _make_co_host(client, host, meeting_id, "700")

    assert response.status_code == 204
    [member] = await _all(db, MeetingMember, MeetingMember.user_id == "700")
    assert member.role == MemberRole.CO_HOST
    # AC-09.4: the label, for everyone in the room.
    assert fake_livekit.attributes[(meeting_id, _identity("700", "tab-device-01"))] == {
        "role": "co_host"
    }
    [record] = await _all(db, AuditRecord, AuditRecord.action == "co_host_appointed")
    assert record.details == {"user_id": "700", "name": "Person 700"}


async def test_host_can_appoint_several_co_hosts(client, db, meeting_id, host):
    # AC-09.1
    for user in ("700", "701"):
        await _admitted(client, host, meeting_id, user)
        assert (await _make_co_host(client, host, meeting_id, user)).status_code == 204

    roles = {m.user_id: m.role for m in await _all(db, MeetingMember)}
    assert roles["700"] == roles["701"] == MemberRole.CO_HOST


async def test_co_host_can_admit_and_deny(client, db, meeting_id, host):
    # AC-09.2 (admit part)
    co_host = await _admitted(client, host, meeting_id, "700")
    await _make_co_host(client, host, meeting_id, "700")
    await _join(client, await _person(client, "701"), meeting_id)
    request_id = await _request_id(client, co_host, meeting_id, "701")

    assert (await _decide(client, co_host, meeting_id, request_id, "admit")).status_code == 204


async def test_co_host_cannot_make_co_hosts(client, db, meeting_id, host):
    # AC-09.6
    co_host = await _admitted(client, host, meeting_id, "700")
    await _make_co_host(client, host, meeting_id, "700")
    await _admitted(client, host, meeting_id, "701")

    response = await _make_co_host(client, co_host, meeting_id, "701")

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


async def test_someone_waiting_cannot_be_made_co_host(client, db, meeting_id, host):
    # AC-09.3
    await _join(client, await _person(client, "700"), meeting_id)

    response = await _make_co_host(client, host, meeting_id, "700")

    assert response.status_code == 409
    assert response.json()["error"]["code"] == "not_admitted"


async def test_co_host_keeps_the_role_and_rejoins_directly(client, db, meeting_id, host):
    # AC-09.5: away for a long time, yet straight back in as Co-host (place held).
    co_host = await _admitted(client, host, meeting_id, "700")
    await _make_co_host(client, host, meeting_id, "700")
    await _device_left(db, meeting_id, "700", minutes_ago=30)
    await db.execute(
        update(MeetingMember)
        .where(MeetingMember.user_id == "700")
        .values(admitted_at=func.now() - timedelta(minutes=31))
    )
    await db.commit()
    await _fill(db, meeting_id, 23)  # Host + Co-host (held) + 23 = 25

    # The Co-host's place is kept for them: a newcomer can't take it...
    assert (await _join(client, await _person(client, "799"), meeting_id)).status_code == 409
    # ...and they get straight back in.
    response = await _join(client, co_host, meeting_id)

    assert response.status_code == 200
    assert response.json()["role"] == "co_host"
    assert response.json()["microphone_on"] is False  # coming back is muted, Co-hosts too


async def test_new_co_host_gets_waiting_notifications(client, db, meeting_id, host, fake_livekit):
    await _admitted(client, host, meeting_id, "700")
    await _make_co_host(client, host, meeting_id, "700")

    await _join(client, await _person(client, "701"), meeting_id)

    assert sorted(fake_livekit.sent[-1].identities) == sorted(
        [_identity(HOST, HOST_DEVICE), _identity("700", "tab-device-01")]
    )


# --- Alerts ----------------------------------------------------------------


def test_alerts_are_not_repeated_within_5_minutes():
    now = [0.0]
    alerts = Alerts(clock=lambda: now[0])

    assert alerts.unavailable("livekit", "down")
    assert not alerts.unavailable("livekit", "still down")
    assert alerts.unavailable("database", "down")  # another service alerts at once
    now[0] = 301
    assert alerts.unavailable("livekit", "down again")