"""HEL-973: start an instant meeting and join it as Host."""

import asyncio
from datetime import UTC, datetime, timedelta

import jwt
import pytest
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import async_sessionmaker

from app.audit.models import AuditRecord
from app.livekit.adapter import LiveKitAdapter
from app.meetings.jobs import forget_unjoined_forever
from app.meetings.lifecycle import forget_unjoined
from app.meetings.models import Meeting, MeetingMember, MeetingStatus, MemberRole
from tests.conftest import make_app, make_settings, running, session_headers

ME = "674"  # the test token's user (sub), organisation 26
DEVICE = "tab-0001-abcd"
UNAVAILABLE = {"code": "service_unavailable", "message": "Meetings are temporarily unavailable"}


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


async def _start(client, headers):
    return await client.post("/v1/meetings", headers=headers)


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


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


async def _started_minutes_ago(db, meeting_id, minutes):
    await db.execute(
        update(Meeting)
        .where(Meeting.id == meeting_id)
        .values(created_at=func.now() - timedelta(minutes=minutes))
    )
    await db.commit()


# --- Start meeting ---------------------------------------------------------


async def test_start_meeting_creates_an_instant_meeting_with_the_caller_as_host(
    client, db, headers
):
    response = await _start(client, headers)

    assert response.status_code == 201
    body = response.json()
    assert body["title"] == "Instant meeting"
    assert body["host_name"] == "Test User"
    assert body["kind"] == "meeting"
    assert body["room_name"] is None

    [meeting] = await _all(db, Meeting)
    assert meeting.id == body["id"]
    assert (meeting.product, meeting.org_id) == ("helixtahr", "26")  # BR-02.3
    assert meeting.kind == "normal"
    assert meeting.status == MeetingStatus.ACTIVE
    assert meeting.owner_user_id == meeting.host_user_id == ME
    assert meeting.host_joined_at is None
    [member] = await _all(db, MeetingMember)
    assert (member.user_id, member.role) == (ME, MemberRole.HOST)


async def test_started_meeting_is_listed_for_the_host_at_once(client, db, headers):
    meeting_id = (await _start(client, headers)).json()["id"]

    listed = (await client.get("/v1/meetings", headers=headers)).json()["meetings"]

    assert [row["id"] for row in listed] == [meeting_id]


async def test_started_meeting_is_not_listed_for_anyone_else(client, db, headers):
    await _start(client, headers)
    colleague = await session_headers(client, sub="675", name="Colleague")

    listed = (await client.get("/v1/meetings", headers=colleague)).json()["meetings"]

    assert listed == []


async def test_start_meeting_writes_no_audit_record(client, db, headers):
    # BR-02.5 / AC-02.7: leaving pre-join before Join now leaves no audit record.
    await _start(client, headers)

    assert await _all(db, AuditRecord) == []


async def test_start_again_before_joining_returns_the_same_meeting(client, db, headers):
    first = await _start(client, headers)
    second = await _start(client, headers)

    assert second.status_code == 200
    assert second.json() == first.json()
    assert len(await _all(db, Meeting)) == 1


async def test_double_click_makes_only_one_meeting(client, db, headers):
    # AC-02.5: two Start meeting requests at the same moment.
    responses = await asyncio.gather(*(_start(client, headers) for _ in range(5)))

    assert {response.status_code for response in responses} <= {200, 201}
    assert len({response.json()["id"] for response in responses}) == 1
    assert len(await _all(db, Meeting)) == 1


async def test_start_after_joining_makes_a_new_meeting(client, db, headers):
    first = (await _start(client, headers)).json()["id"]
    await _join(client, headers, first)

    second = await _start(client, headers)

    assert second.status_code == 201
    assert second.json()["id"] != first


async def test_each_host_gets_their_own_meeting(client, db, headers):
    mine = (await _start(client, headers)).json()["id"]
    colleague = await session_headers(client, sub="675", name="Colleague")
    other_org = await session_headers(client, company_id=27)

    theirs = (await _start(client, colleague)).json()["id"]
    same_user_other_org = (await _start(client, other_org)).json()["id"]

    assert len({mine, theirs, same_user_other_org}) == 3


async def test_unjoined_meeting_older_than_10_minutes_is_replaced_at_start(client, db, headers):
    old = (await _start(client, headers)).json()["id"]
    await _started_minutes_ago(db, old, 11)

    response = await _start(client, headers)

    assert response.status_code == 201
    assert response.json()["id"] != old
    assert [meeting.id for meeting in await _all(db, Meeting)] == [response.json()["id"]]


async def test_start_with_livekit_down_says_unavailable_and_creates_nothing(
    client, db, headers, fake_livekit
):
    # AC-02.3
    fake_livekit.healthy = False

    response = await _start(client, headers)

    assert response.status_code == 503
    assert response.json() == {"error": UNAVAILABLE}
    assert await _all(db, Meeting) == []


async def test_start_with_database_down_says_unavailable(fake_livekit, fake_helixtahr):
    # AC-02.3: nothing listens on port 1.
    settings = make_settings(database_url="postgresql+asyncpg://x:y@localhost:1/none")
    async with running(make_app(settings, fake_livekit, fake_helixtahr)) as client:
        headers = await session_headers(client)

        response = await _start(client, headers)

    assert response.status_code == 503
    assert response.json() == {"error": UNAVAILABLE}


async def test_start_meeting_needs_a_session(client):
    response = await client.post("/v1/meetings")

    assert response.status_code == 401


# --- Join now (Host) -------------------------------------------------------


async def test_host_join_now_gives_a_meeting_pass_with_mic_on(client, db, headers, fake_livekit):
    meeting_id = (await _start(client, headers)).json()["id"]

    response = await _join(client, headers, meeting_id)

    assert response.status_code == 200
    assert response.json() == {
        "livekit_url": "ws://livekit.test",
        "meeting_pass": "fake-pass-1",
        "connect_within_seconds": 600,
        "role": "host",
        "microphone_on": True,
        "status": "joined",
        "original_host": True,
        "room_id": None,
    }
    [issued] = fake_livekit.passes
    assert issued.room == meeting_id
    assert issued.identity == f"helixtahr:26:{ME}:{DEVICE}"
    assert issued.name == "Test User"
    assert issued.sources == ["camera", "microphone"]
    assert issued.connect_within_seconds == 600


async def test_host_join_now_creates_the_livekit_room(client, db, headers, fake_livekit):
    meeting_id = (await _start(client, headers)).json()["id"]

    await _join(client, headers, meeting_id)

    assert fake_livekit.rooms == {
        meeting_id: {"max_participants": 25, "empty_timeout": 600, "departure_timeout": 120}
    }


async def test_first_join_now_writes_the_audit_record(client, db, headers):
    # AC-02.6: meeting, creator, organisation, time.
    meeting_id = (await _start(client, headers)).json()["id"]
    before = datetime.now(UTC)

    await _join(client, headers, meeting_id)

    [record] = await _all(db, AuditRecord)
    assert record.action == "meeting_created"
    assert record.meeting_id == meeting_id
    assert (record.actor_user_id, record.actor_name) == (ME, "Test User")
    assert (record.product, record.org_id) == ("helixtahr", "26")
    assert record.created_at >= before - timedelta(seconds=5)
    assert record.details == {"title": "Instant meeting"}
    [meeting] = await _all(db, Meeting)
    assert meeting.host_joined_at is not None


async def test_joining_from_a_second_device_writes_no_second_record(
    client, db, headers, fake_livekit
):
    meeting_id = (await _start(client, headers)).json()["id"]

    await _join(client, headers, meeting_id, device="tab-aaaa-1111")
    await _join(client, headers, meeting_id, device="tab-bbbb-2222")

    assert len(await _all(db, AuditRecord)) == 1
    identities = [issued.identity for issued in fake_livekit.passes]
    assert identities == [f"helixtahr:26:{ME}:tab-aaaa-1111", f"helixtahr:26:{ME}:tab-bbbb-2222"]


async def test_join_now_asks_the_product_fresh(client, db, headers, fake_helixtahr):
    meeting_id = (await _start(client, headers)).json()["id"]
    before = fake_helixtahr.allowed_calls

    await _join(client, headers, meeting_id)
    await _join(client, headers, meeting_id)

    assert fake_helixtahr.allowed_calls == before + 2


async def test_join_refused_once_the_organisation_loses_meetings(
    client, db, headers, fake_helixtahr
):
    meeting_id = (await _start(client, headers)).json()["id"]
    fake_helixtahr.meetings_allowed = False

    response = await _join(client, headers, meeting_id)

    assert response.status_code == 403
    assert response.json()["error"]["code"] == "no_access"
    assert await _all(db, AuditRecord) == []


async def test_join_with_livekit_down_saves_nothing(client, db, headers, fake_livekit):
    meeting_id = (await _start(client, headers)).json()["id"]
    fake_livekit.healthy = False

    response = await _join(client, headers, meeting_id)

    assert response.status_code == 503
    assert response.json() == {"error": UNAVAILABLE}
    assert await _all(db, AuditRecord) == []
    [meeting] = await _all(db, Meeting)
    assert meeting.host_joined_at is None

    fake_livekit.healthy = True
    assert (await _join(client, headers, meeting_id)).status_code == 200
    assert len(await _all(db, AuditRecord)) == 1


async def test_join_after_10_minutes_unjoined_says_link_invalid(client, db, headers, fake_livekit):
    meeting_id = (await _start(client, headers)).json()["id"]
    await _started_minutes_ago(db, meeting_id, 11)

    response = await _join(client, headers, meeting_id)

    assert response.status_code == 404
    assert response.json()["error"]["code"] == "link_invalid"
    assert await _all(db, AuditRecord) == []
    assert fake_livekit.rooms == {}


async def test_host_can_rejoin_after_10_minutes_once_joined(client, db, headers):
    meeting_id = (await _start(client, headers)).json()["id"]
    await _join(client, headers, meeting_id)
    await _started_minutes_ago(db, meeting_id, 30)

    assert (await _join(client, headers, meeting_id)).status_code == 200


# --- Join now (everyone else) ----------------------------------------------


@pytest.mark.parametrize("opened_link_first", [True, False])
async def test_others_wait_for_admission(client, db, headers, fake_livekit, opened_link_first):
    # BR-02.4 / AC-02.8: a same-organisation employee never joins directly; the
    # waiting room itself is tested in test_waiting_room.py.
    meeting_id = (await _start(client, headers)).json()["id"]
    colleague = await session_headers(client, sub="675", name="Colleague")
    if opened_link_first:
        await client.get(f"/v1/meetings/{meeting_id}", headers=colleague)

    response = await _join(client, colleague, meeting_id)

    assert response.status_code == 202
    assert response.json() == {"status": "waiting"}
    assert fake_livekit.passes == []
    assert fake_livekit.rooms == {}


async def test_join_another_organisations_meeting_is_refused(client, db, headers):
    meeting_id = (await _start(client, headers)).json()["id"]
    outsider = await session_headers(client, company_id=27)

    response = await _join(client, outsider, meeting_id)

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


@pytest.mark.parametrize("meeting_id", ["zzzz-zzzz-zzz", "NOT_A_LINK"])
async def test_join_unknown_link_says_link_invalid(client, db, headers, meeting_id):
    response = await _join(client, headers, meeting_id)

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


@pytest.mark.parametrize("device_id", ["short", "has spaces in it", "x" * 65, ""])
async def test_join_needs_a_valid_device_id(client, db, headers, device_id):
    meeting_id = (await _start(client, headers)).json()["id"]

    response = await _join(client, headers, meeting_id, device=device_id)

    assert response.status_code == 422


async def test_join_needs_a_session(client):
    response = await client.post("/v1/meetings/abcd-efgh-ijk/join", json={"device_id": DEVICE})

    assert response.status_code == 401


# --- Forgetting unjoined meetings (AC-02.7) --------------------------------


def _seed(db, meeting_id, *, minutes_ago, joined=False, host="900", **fields):
    started = datetime.now(UTC) - timedelta(minutes=minutes_ago)
    db.add(
        Meeting(
            id=meeting_id,
            product="helixtahr",
            org_id="26",
            title="Instant meeting",
            owner_user_id=host,
            host_user_id=host,
            host_name="Someone",
            created_at=started,
            host_joined_at=started if joined else None,
            **fields,
        )
    )


async def test_forget_unjoined_removes_only_meetings_nobody_joined_in_10_minutes(db):
    _seed(db, "old-unjoined", minutes_ago=11, host="901")
    _seed(db, "new-unjoined", minutes_ago=9, host="902")
    _seed(db, "old-joined", minutes_ago=60, joined=True, host="903")
    _seed(
        db,
        "old-ended",
        minutes_ago=60,
        joined=True,
        host="904",
        status=MeetingStatus.ENDED,
        ended_at=datetime.now(UTC),
    )
    await db.flush()
    db.add(
        MeetingMember(
            meeting_id="old-unjoined",
            product="helixtahr",
            org_id="26",
            user_id="905",
            user_name="Opened the link",
            link_opened_at=datetime.now(UTC),
        )
    )
    await db.commit()

    assert await forget_unjoined(db) == 1

    remaining = {meeting.id for meeting in await _all(db, Meeting)}
    assert remaining == {"new-unjoined", "old-joined", "old-ended"}
    assert await _all(db, MeetingMember) == []


async def test_forgotten_meetings_are_not_in_history(client, db, headers):
    meeting_id = (await _start(client, headers)).json()["id"]
    await _started_minutes_ago(db, meeting_id, 11)
    await forget_unjoined(db)

    assert (await client.get("/v1/meetings", headers=headers)).json()["meetings"] == []
    history = (await client.get("/v1/meetings/history", headers=headers)).json()
    assert history["meetings"] == []
    link = await client.get(f"/v1/meetings/{meeting_id}", headers=headers)
    assert link.json()["error"]["code"] == "link_invalid"


async def test_clean_up_job_runs_in_the_background(db):
    _seed(db, "old-unjoined", minutes_ago=11)
    await db.commit()
    sessions = async_sessionmaker(db.bind, expire_on_commit=False)

    job = asyncio.create_task(forget_unjoined_forever(sessions, interval=0.01))
    try:
        for _ in range(200):
            await asyncio.sleep(0.01)
            if not await _all(db, Meeting):
                break
    finally:
        job.cancel()

    assert await _all(db, Meeting) == []


# --- The real LiveKit adapter's meeting pass -------------------------------


async def test_meeting_pass_lets_one_device_into_one_room():
    secret = "test-livekit-secret-0123456789abcdef"
    adapter = LiveKitAdapter("http://livekit.test", "test-key", secret)
    try:
        token = adapter.make_pass(
            room="k3x9-p2mq-7ad",
            identity="helixtahr:26:674:tab-0001",
            name="Test User",
            sources=["camera", "microphone"],
            connect_within_seconds=600,
        )
    finally:
        await adapter.aclose()

    claims = jwt.decode(token, secret, algorithms=["HS256"])
    assert claims["sub"] == "helixtahr:26:674:tab-0001"
    assert claims["name"] == "Test User"
    assert claims["exp"] - claims["nbf"] == 600
    assert claims["video"] == {
        "roomJoin": True,
        "room": "k3x9-p2mq-7ad",
        "canPublish": True,
        "canSubscribe": True,
        "canPublishData": True,  # chat (HEL-980)
        "canPublishSources": ["camera", "microphone"],
        "canUpdateOwnMetadata": False,
    }