"""
Invitation links: a code bound to no account until somebody redeems it.

A link is sent to someone who may not have an account yet, so its code cannot
name one. That makes it a bearer code at the node — the hub's ticket, bound to
a verified address, is what decides who can reach the node at all
(docs/MESHBAY_DESIGN.md §3.4). Everything here is a way a bearer code could be made
to mean more than "one new member of this group, once", or a way an unbound row
could leak into the code paths written for bound ones. The second family is the
one to watch: `user_id = ''` must never read as "anyone" (AV1).
"""

import sqlite3
from datetime import UTC, datetime, timedelta

import pytest
from meshbay_common.crypto import generate_gek, unwrap_gek_aes
from meshbay_common.join import ROLE_MEMBER, ROLE_OPERATOR
from meshbay_node.roster import (
    KIND_ACCOUNT,
    KIND_LINK,
    MAX_LINK_INVITES_PER_GROUP,
    LinkInviteLimit,
    Roster,
    hash_code,
)
from test_roster_pairing import _join_msg, _keypair, _keypair_full, _last, _session, _x_raw

GROUP_A = "a" * 32
GROUP_B = "b" * 32


@pytest.fixture
async def roster(tmp_path):
    r = Roster(db_path=tmp_path / "roster.db")
    await r.open()
    yield r
    await r.close()


async def _link(roster, group_id=GROUP_B):
    code, invite_id, _expires = await roster.create_link_invite(group_id, "cbesson")
    return code, invite_id


# ── The roster ───────────────────────────────────────────────────────────────

async def test_a_link_is_redeemed_once_and_then_names_its_redeemer(roster):
    code, _ = await _link(roster)
    invite = await roster.consume_invite(code, "alice", group_id=GROUP_B)
    assert invite and invite["role"] == ROLE_MEMBER and invite["group_id"] == GROUP_B
    assert invite["user_id"] == "alice"

    assert await roster.consume_invite(code, "mallory", group_id=GROUP_B) is None
    used = [i for i in await roster.list_invites(include_used=True)
            if i["code_hash"] == hash_code(code)]
    assert used[0]["user_id"] == "alice" and used[0]["used_at"]


async def test_a_link_is_good_for_its_own_group_only(roster):
    code, _ = await _link(roster, GROUP_B)
    for other in (GROUP_A, ""):
        assert await roster.consume_invite(code, "alice", group_id=other) is None
    # Not spent by the refusals.
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B)


async def test_an_active_member_cannot_spend_somebody_elses_link(roster):
    await roster.set_member(GROUP_B, "bob", ROLE_MEMBER, "active", "cbesson")
    code, _ = await _link(roster)
    assert await roster.consume_invite(code, "bob", group_id=GROUP_B) is None
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B)


async def test_a_link_never_carries_operator_authority(roster, tmp_path):
    code, _ = await _link(roster)
    rows = [i for i in await roster.list_invites() if i["kind"] == KIND_LINK]
    assert rows and all(r["role"] == ROLE_MEMBER for r in rows)

    # Even a row edited on disk to say otherwise is refused, not honoured.
    con = sqlite3.connect(tmp_path / "roster.db")
    con.execute("UPDATE invites SET role = ? WHERE code_hash = ?",
                (ROLE_OPERATOR, hash_code(code)))
    con.commit()
    con.close()
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B) is None


async def test_an_empty_account_on_a_bound_invite_is_nobody(roster, tmp_path):
    """
    AV1's shape: an account invitation whose `user_id` is empty must match no
    one — not whoever turns up. Only `kind = 'link'` makes a row a bearer code.
    """
    code = await roster.create_invite(GROUP_B, "", ROLE_MEMBER, "cbesson")
    for who in ("alice", ""):
        assert await roster.consume_invite(code, who, group_id=GROUP_B) is None


async def test_bound_invitations_and_links_do_not_cancel_each_other(roster):
    code, invite_id = await _link(roster)
    # Re-inviting an account supersedes that account's earlier code — and must
    # not take the group's unredeemed links with it.
    await roster.create_invite(GROUP_B, "carol", ROLE_MEMBER, "cbesson")
    await roster.create_invite(GROUP_B, "carol", ROLE_MEMBER, "cbesson")
    # Cancelling "nobody's" invitations is not cancelling the links.
    assert await roster.drop_invites(GROUP_B, "") == 0
    assert invite_id in {i["invite_id"] for i in await roster.list_invites()}
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B)


async def test_links_are_capped_per_group(roster):
    codes = [(await _link(roster, GROUP_B))[0] for _ in range(MAX_LINK_INVITES_PER_GROUP)]
    with pytest.raises(LinkInviteLimit):
        await _link(roster, GROUP_B)
    # Another group has its own allowance.
    await _link(roster, GROUP_A)
    # A redeemed link is no longer outstanding, so it frees a place.
    assert await roster.consume_invite(codes[0], "alice", group_id=GROUP_B)
    await _link(roster, GROUP_B)


async def test_a_link_needs_a_group(roster):
    with pytest.raises(ValueError):
        await roster.create_link_invite("", "cbesson")


async def test_cancel_takes_back_an_unredeemed_link_of_its_own_group(roster):
    code, invite_id = await _link(roster, GROUP_B)
    assert not await roster.cancel_invite(GROUP_A, invite_id)
    assert await roster.cancel_invite(GROUP_B, invite_id)
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B) is None

    code, invite_id = await _link(roster, GROUP_B)
    await roster.consume_invite(code, "alice", group_id=GROUP_B)
    assert not await roster.cancel_invite(GROUP_B, invite_id), (
        "a redeemed link is the record of the join, not something to cancel")


async def test_an_expired_link_is_refused(roster):
    code, _, _ = await roster.create_link_invite(GROUP_B, "cbesson", ttl=-1)
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B) is None


async def test_a_roster_from_before_links_opens_and_keeps_its_codes(tmp_path):
    """The two columns arrive by ALTER TABLE; an existing code stays a bound one."""
    path = tmp_path / "old.db"
    con = sqlite3.connect(path)
    con.execute("""CREATE TABLE invites (
        code_hash TEXT PRIMARY KEY, group_id TEXT NOT NULL, user_id TEXT NOT NULL,
        username TEXT NOT NULL DEFAULT '', role TEXT NOT NULL, created_by TEXT NOT NULL,
        created_at TEXT NOT NULL, expires_at TEXT NOT NULL, used_at TEXT)""")
    expires = (datetime.now(UTC) + timedelta(days=1)).isoformat(timespec="seconds")
    con.execute("INSERT INTO invites VALUES (?, ?, ?, '', ?, 'op', ?, ?, NULL)",
                (hash_code("K7P2-9WQX"), GROUP_B, "alice", ROLE_MEMBER,
                 datetime.now(UTC).isoformat(), expires))
    con.commit()
    con.close()

    r = Roster(db_path=path)
    await r.open()
    try:
        [row] = await r.list_invites()
        assert row["kind"] == KIND_ACCOUNT
        assert await r.consume_invite("K7P2-9WQX", "mallory", group_id=GROUP_B) is None
        assert await r.consume_invite("K7P2-9WQX", "alice", group_id=GROUP_B)
    finally:
        await r.close()


# ── The join ─────────────────────────────────────────────────────────────────

async def test_a_newcomer_joins_with_a_link_code(tmp_path, roster):
    gek = generate_gek()
    code, _ = await _link(roster)
    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()
    session = _session(tmp_path, roster, user_id="alice", group_id=GROUP_B, gek=gek)

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="alice", group_id=GROUP_B))

    reply = _last(session)
    assert reply["ok"] is True and reply["gek"] is True
    assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek
    member = await roster.get_member(GROUP_B, "alice")
    assert member and member["role"] == ROLE_MEMBER


async def test_a_link_code_opens_no_other_group(tmp_path, roster):
    code, _ = await _link(roster, GROUP_A)
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    session = _session(tmp_path, roster, user_id="alice", group_id=GROUP_B,
                       gek=generate_gek())
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="alice", group_id=GROUP_B))
    assert _last(session).get("reason") == "code_invalid"
    assert await roster.get_member(GROUP_A, "alice") is None


async def test_someone_already_pinned_elsewhere_joins_with_a_link(tmp_path, roster):
    """The common case: known to this node through another group."""
    gek_b = generate_gek()
    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP_A, "grenet", ROLE_MEMBER, "active", "cbesson")
    code, _ = await _link(roster, GROUP_B)

    session = _session(tmp_path, roster, user_id="grenet", group_id=GROUP_B, gek=gek_b)
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="grenet", group_id=GROUP_B))

    reply = _last(session)
    assert reply["ok"] is True and reply["gek"] is True
    assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek_b
    assert await roster.get_member(GROUP_B, "grenet")


async def test_a_member_opening_the_group_with_a_link_leaves_it_unspent(tmp_path, roster):
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP_B, "grenet", ROLE_MEMBER, "active", "cbesson")
    code, _ = await _link(roster, GROUP_B)

    session = _session(tmp_path, roster, user_id="grenet", group_id=GROUP_B,
                       gek=generate_gek())
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="grenet", group_id=GROUP_B))
    assert _last(session)["ok"] is True
    assert await roster.consume_invite(code, "alice", group_id=GROUP_B)


async def test_a_known_device_with_a_wrong_code_is_told_so(tmp_path, roster):
    """Not the flat `not_authorized_for_group`: a code was offered and refused,
    and the refusal counts against the attempt budget like any other."""
    sk_ed, pk_ed_b64, pk_x_b64 = _keypair()
    await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP_A, "eve", ROLE_MEMBER, "active", "cbesson")
    session = _session(tmp_path, roster, user_id="eve", group_id=GROUP_B,
                       gek=generate_gek())
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code="AAAA-AAAA",
                  user_id="eve", group_id=GROUP_B))
    assert _last(session).get("reason") == "code_invalid"
    assert session._join_attempts == 1


async def test_someone_removed_can_come_back_with_a_link(tmp_path, roster):
    """A revoked member still has a member row here. A link sent to bring them
    back must work — it is what the operator asked for."""
    gek = generate_gek()
    sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full()
    await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code")
    await roster.set_member(GROUP_B, "grenet", ROLE_MEMBER, "revoked", "cbesson")
    session = _session(tmp_path, roster, user_id="grenet", group_id=GROUP_B, gek=gek)

    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, user_id="grenet", group_id=GROUP_B))
    assert _last(session).get("reason") == "not_authorized_for_group"

    code, _ = await _link(roster, GROUP_B)
    await session._do_join_request(
        _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, code=code,
                  user_id="grenet", group_id=GROUP_B))
    reply = _last(session)
    assert reply["ok"] is True and reply["gek"] is True
    assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek
    assert (await roster.get_member(GROUP_B, "grenet"))["status"] == "active"