"""
Groups nobody hosts, and the admission policy a public group is allowed.

A group created before its node exists has no files, no key and nothing to
connect to. Showing it in the directory produces a dead end for whoever clicks
it, so it stays with its owner until a node announces it — and if none ever
does, `prune-groups` collects it.

The distinction that matters is *ever hosted* against *online now*. `hosted_at`
is stamped once and never cleared, so a node being offline today cannot make a
live group look abandoned. Checking the live socket registry instead would have
deleted every group during a hub restart.
"""

import base64
import hashlib
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from meshbay_hub.db.models import Group, GroupMember
from meshbay_hub.tasks.cleanup import find_unhosted_groups, prune_unhosted_groups


def _auth_key(password: str, username: str) -> str:
    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
    return base64.b64encode(
        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()


async def _user(client, username, password="a-long-enough-passphrase"):
    await client.post("/v1/users/register", json={
        "username": username, "email": f"{username}@example.com",
        "auth_key": _auth_key(password, username)})
    r = await client.post("/v1/users/login", json={
        "username": username, "auth_key": _auth_key(password, username)})
    return {"Authorization": f"Bearer {r.json()['access_token']}"}


async def _group(client, owner, name, visibility="private", join_policy=None):
    body = {"name": name, "visibility": visibility}
    if join_policy is not None:
        body["join_policy"] = join_policy
    elif visibility == "public":
        body["join_policy"] = "open"
    return await client.post("/v1/groups", json=body, headers=owner)


async def _mark_hosted(db_session, group_id, when=None):
    g = await db_session.get(Group, group_id)
    g.hosted_at = when or datetime.now(timezone.utc)
    await db_session.commit()


# ── Visibility before a node exists ───────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_owner_sees_their_unhosted_group(client):
    """They have to: it is the page they set the node up from."""
    owner = await _user(client, "setup1_test")
    r = await _group(client, owner, "not-yet")
    assert r.status_code == 201

    mine = await client.get("/v1/groups/mine", headers=owner)
    entry = next(g for g in mine.json()["groups"] if g["name"] == "not-yet")
    assert entry["hosted"] is False


@pytest.mark.asyncio
async def test_a_member_does_not_see_an_unhosted_group(client):
    """A name they cannot open, with no way to say why, is worse than nothing."""
    owner = await _user(client, "setup2_test")
    member = await _user(client, "early_bird")
    gid = (await _group(client, owner, "premature")).json()["group_id"]
    await client.post(f"/v1/groups/{gid}/members/early_bird", json={}, headers=owner)

    mine = await client.get("/v1/groups/mine", headers=member)
    assert [g["name"] for g in mine.json()["groups"]] == []


@pytest.mark.asyncio
async def test_a_member_sees_it_once_a_node_has_announced_it(client, db_session):
    owner = await _user(client, "setup3_test")
    member = await _user(client, "patient_test")
    gid = (await _group(client, owner, "ready")).json()["group_id"]
    await client.post(f"/v1/groups/{gid}/members/patient_test", json={}, headers=owner)

    await _mark_hosted(db_session, gid)

    mine = await client.get("/v1/groups/mine", headers=member)
    entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
    assert entry["hosted"] is True


@pytest.mark.asyncio
async def test_the_public_directory_hides_unhosted_groups(client, db_session):
    owner = await _user(client, "setup4_test")
    hidden = (await _group(client, owner, "pub-unhosted", "public")).json()["group_id"]
    shown = (await _group(client, owner, "pub-hosted", "public")).json()["group_id"]
    await _mark_hosted(db_session, shown)

    listed = (await client.get("/v1/groups")).json()["groups"]
    ids = [g["id"] for g in listed]
    assert shown in ids
    assert hidden not in ids


@pytest.mark.asyncio
async def test_a_group_stays_visible_when_its_node_goes_offline(client, db_session):
    """`hosted_at` records that a node existed, not that one is answering now."""
    owner = await _user(client, "setup5_test")
    gid = (await _group(client, owner, "quiet-node", "public")).json()["group_id"]
    await _mark_hosted(db_session, gid)

    listed = (await client.get("/v1/groups")).json()["groups"]
    assert gid in [g["id"] for g in listed], "no node is online, and that is not the question"


# ── Collecting the abandoned ones ─────────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_fresh_unhosted_group_is_left_alone(client, db_session):
    owner = await _user(client, "reaper1_test")
    gid = (await _group(client, owner, "brand-new")).json()["group_id"]

    assert await find_unhosted_groups(db_session) == []
    assert await db_session.get(Group, gid) is not None


@pytest.mark.asyncio
async def test_an_unhosted_group_past_the_grace_period_is_collected(client, db_session):
    owner = await _user(client, "reaper2_test")
    gid = (await _group(client, owner, "abandoned")).json()["group_id"]

    g = await db_session.get(Group, gid)
    g.created_at = datetime.now(timezone.utc) - timedelta(days=8)
    await db_session.commit()

    gone = await prune_unhosted_groups(db_session)
    assert [name for _, name in gone] == ["abandoned"]
    assert await db_session.get(Group, gid) is None


@pytest.mark.asyncio
async def test_an_old_group_that_was_hosted_is_never_collected(client, db_session):
    """The whole point of the column: age alone must not condemn a group."""
    owner = await _user(client, "reaper3_test")
    gid = (await _group(client, owner, "long-lived")).json()["group_id"]

    g = await db_session.get(Group, gid)
    g.created_at = datetime.now(timezone.utc) - timedelta(days=400)
    g.hosted_at = datetime.now(timezone.utc) - timedelta(days=399)
    await db_session.commit()

    assert await find_unhosted_groups(db_session) == []


@pytest.mark.asyncio
async def test_dry_run_reports_without_deleting(client, db_session):
    owner = await _user(client, "reaper4_test")
    gid = (await _group(client, owner, "still-here")).json()["group_id"]
    g = await db_session.get(Group, gid)
    g.created_at = datetime.now(timezone.utc) - timedelta(days=30)
    await db_session.commit()

    gone = await prune_unhosted_groups(db_session, dry_run=True)
    assert [name for _, name in gone] == ["still-here"]
    assert await db_session.get(Group, gid) is not None, "dry run deleted a group"


@pytest.mark.asyncio
async def test_collecting_a_group_takes_its_memberships_with_it(client, db_session):
    """Nothing cascades in the schema, and an orphan row keeps the group in /mine."""
    owner = await _user(client, "reaper5_test")
    await _user(client, "tagalong")
    gid = (await _group(client, owner, "doomed")).json()["group_id"]
    await client.post(f"/v1/groups/{gid}/members/tagalong", json={}, headers=owner)

    g = await db_session.get(Group, gid)
    g.created_at = datetime.now(timezone.utc) - timedelta(days=9)
    await db_session.commit()

    await prune_unhosted_groups(db_session)

    rows = (await db_session.execute(
        select(GroupMember).where(GroupMember.group_id == gid))).scalars().all()
    assert rows == []


# ── What a public group may be ────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_a_public_group_cannot_be_invite_only(client):
    """It would be listed to everyone and admit nobody.

    Admission by request was the alternative and was dropped: between strangers
    the only channel is the hub, so a one-time code would travel through the
    party it exists to exclude.
    """
    owner = await _user(client, "policy1_test")
    r = await _group(client, owner, "contradiction", "public", join_policy="invite")
    assert r.status_code == 422, r.text
    assert "private" in r.json()["detail"]


@pytest.mark.asyncio
async def test_a_public_group_is_open(client):
    owner = await _user(client, "policy2_test")
    r = await _group(client, owner, "welcoming", "public", join_policy="open")
    assert r.status_code == 201, r.text


@pytest.mark.asyncio
async def test_a_private_group_is_invite_only_by_default(client, db_session):
    owner = await _user(client, "policy3_test")
    gid = (await _group(client, owner, "closed")).json()["group_id"]
    assert (await db_session.get(Group, gid)).join_policy == "invite"


# ── The stamp itself ──────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_node_registration_stamps_hosted_at(client, db_session):
    """The tests above set `hosted_at` directly; this one exercises the writer.

    `_mark_hosted` is called from the node WebSocket handler with the group list
    the hub derived from its own tables — a node can narrow that set but never
    widen it (finding C2), so being announced here is evidence, not a claim.
    """
    from meshbay_hub.api.revocation import _mark_hosted

    owner = await _user(client, "stamped_test")
    gid = (await _group(client, owner, "about-to-be-hosted")).json()["group_id"]
    assert (await db_session.get(Group, gid)).hosted_at is None

    await _mark_hosted([gid])

    await db_session.refresh(await db_session.get(Group, gid))
    assert (await db_session.get(Group, gid)).hosted_at is not None


@pytest.mark.asyncio
async def test_the_stamp_is_not_moved_by_a_later_reconnection(client, db_session):
    """It records that a node once existed, not when one last showed up."""
    from meshbay_hub.api.revocation import _mark_hosted

    owner = await _user(client, "stamped2")
    gid = (await _group(client, owner, "steady")).json()["group_id"]
    first = datetime.now(timezone.utc) - timedelta(days=30)
    await _mark_hosted([gid])
    g = await db_session.get(Group, gid)
    g.hosted_at = first
    await db_session.commit()

    await _mark_hosted([gid])

    g = await db_session.get(Group, gid)
    await db_session.refresh(g)
    # SQLite hands back a naive datetime where PostgreSQL keeps the offset, so
    # the comparison is made on common ground rather than on the driver.
    stored = g.hosted_at.replace(tzinfo=timezone.utc) if g.hosted_at.tzinfo is None \
        else g.hosted_at
    assert abs((stored - first).total_seconds()) < 1, "the stamp moved"


@pytest.mark.asyncio
async def test_marking_hosted_survives_an_empty_list(client):
    """A node that hosts nothing must not take the socket down."""
    from meshbay_hub.api.revocation import _mark_hosted
    await _mark_hosted([])