"""
What one member's chat costs the others.

This is the node-side entry of the register `docs/MESHBAY_DESIGN.md` §13.5b
keeps — the question none of the three security reviews asked: *a participant
supplies input; who else bears the cost?* Its sibling
`meshbay-hub/tests/test_availability_between_members.py` holds the hub's cases
and cannot hold this one, because the defect lives in the node's chat handler
and needs the node's harness.

A chat message is the plainest member-supplied write there is. The node stores
it in `chat.db` on the operator's disk, where nothing expires it — retention is
a manual CLI command (§6.6) — relays it to every other connected member, and
has the hub write a notification for every member of the group. Uploads, the
other member-supplied write, have carried a filename allowlist, strict chunk
ordering, a no-overwrite rule and a per-file size cap since C5a. Chat carried nothing:
the only ceiling was the DataChannel frame, 64 MB once the handshake is done.
One member in a loop filled the operator's disk and saturated everyone else's
connection, and the node's own answer to each message was `ack`.

Two bounds close it, and they answer different halves: **size** bounds what one
message costs, **rate** bounds how often one member may impose it. There is
deliberately no node-wide ceiling — a chat message spends the sender's own
group, and a node-wide one would let a busy group silence a quiet one, which is
this same defect one level up.

The last test is the one that says the bound is the right shape: a member who
has spent their budget has not spent anybody else's.
"""

import base64
import os

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.chatbox import NONCE_LEN, SIG_LEN
from meshbay_common.crypto import pk_to_b64
from meshbay_node.chat import FORMAT_SEALED_V1, ChatStore
from meshbay_node.transport.webrtc import chat as ws
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

# Read with a default rather than imported. Against the source these were
# written for there is no such bound and therefore no such constant, and a test
# that dies of ImportError there proves only that a name is missing — the
# assertions below are what say the node does the limiting.
MAX_CHAT_CIPHERTEXT = getattr(ws, "MAX_CHAT_CIPHERTEXT", 64 * 1024)
_CHAT_RATE_PER_ACCOUNT = getattr(ws, "_CHAT_RATE_PER_ACCOUNT", 60)

pytestmark = pytest.mark.asyncio

GROUP = "g" * 32


@pytest.fixture
async def store(tmp_path):
    s = ChatStore(db_path=tmp_path / "chat.db")
    await s.open()
    yield s
    await s.close()


def _session(ctx, store, *, user: str, conn: str) -> WebRTCPeerSession:
    """One connection, with a device already identified.

    The node checks the envelope's *shape* and that the device named is the one
    this connection proved; it verifies no signature, because the reader does
    that (§4.5). So these tests need no real sealing to reach the bounds, which
    is also a fair description of what the node itself knows.
    """
    device = pk_to_b64(Ed25519PrivateKey.generate().public_key())
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    ctx.setdefault("groups", {})[GROUP] = ctx["group_ctx"]
    session._group_id = GROUP
    session._user_id = user
    session._username = user
    session._pinned_pk = device
    session._device_confirmed = True
    session._registry_key = conn
    session.sent = []
    session._send = session.sent.append
    session.audited = []
    session._audit = lambda event, detail="": session.audited.append(event)
    session._spawn = lambda coro: ctx["pending"].append(coro)
    ctx["group_ctx"]["_peers"][conn] = session
    return session


@pytest.fixture
def ctx(store):
    return {
        "group_ctx": {"chat_store": store, "gek": b"\x01" * 32, "_peers": {}},
        "pending": [],
    }


async def _drain(ctx):
    """Run the writes the handler spawned, as the loop would."""
    for coro in ctx["pending"]:
        await coro
    ctx["pending"].clear()


def _message(session, *, size: int):
    return {
        "format": FORMAT_SEALED_V1,
        "epoch": 1,
        "device": base64.b64decode(session._pinned_pk),
        # A fresh nonce per message, as the real thing has (96 random bits,
        # never a counter — §4.5). A fixed one makes the store's unique
        # `(device, nonce)` refuse every message after the first as a replay,
        # so a flood would look bounded here while being unbounded in
        # production: the fixture, not the node, would be doing the limiting.
        "nonce": os.urandom(NONCE_LEN),
        "sig": b"\x00" * SIG_LEN,
        "ct": b"x" * size,
    }


def _errors(session):
    return [m for m in session.sent if m.get("type") == "error"]


def _acks(session):
    return [m for m in session.sent if m.get("type") == "ack"]


# ── size ─────────────────────────────────────────────────────────────────────

async def test_a_message_larger_than_any_text_is_refused(ctx, store):
    alice = _session(ctx, store, user="alice", conn="c1")
    alice._do_chat_message(_message(alice, size=1024 * 1024))
    await _drain(ctx)

    assert _errors(alice)[-1]["code"] == "chat_too_large"
    assert not _acks(alice)
    assert await store.message_count() == 0, (
        "a megabyte of somebody's choosing reached the operator's disk")


async def test_a_message_at_the_ceiling_is_still_sent(ctx, store):
    """The bound has to admit what it claims to admit: a ceiling that refuses
    an ordinary message is a broken feature, not a strict one."""
    alice = _session(ctx, store, user="alice", conn="c1")
    alice._do_chat_message(_message(alice, size=MAX_CHAT_CIPHERTEXT))
    await _drain(ctx)

    assert not _errors(alice), _errors(alice)
    assert await store.message_count() == 1


async def test_the_refusal_is_audited(ctx, store):
    """An operator asking "why is my disk full" gets an answer."""
    alice = _session(ctx, store, user="alice", conn="c1")
    alice._do_chat_message(_message(alice, size=1024 * 1024))
    await _drain(ctx)
    assert "chat_refused" in alice.audited


# ── rate ─────────────────────────────────────────────────────────────────────

async def _flood(session, ctx, n):
    for _ in range(n):
        session._do_chat_message(_message(session, size=16))
    await _drain(ctx)


async def test_a_flood_stops_at_the_window(ctx, store):
    alice = _session(ctx, store, user="alice", conn="c1")
    await _flood(alice, ctx, _CHAT_RATE_PER_ACCOUNT + 25)

    assert len(_acks(alice)) == _CHAT_RATE_PER_ACCOUNT
    assert _errors(alice)[-1]["code"] == "chat_rate_limited"
    assert await store.message_count() == _CHAT_RATE_PER_ACCOUNT, (
        "the store kept growing after the bound was reached")


async def test_a_second_tab_does_not_double_the_budget(ctx, store):
    """Keyed by account, not by connection. A second tab does not make a person
    type faster, and keying on the session would hand a script one budget per
    socket it opens."""
    first = _session(ctx, store, user="alice", conn="c1")
    second = _session(ctx, store, user="alice", conn="c2")

    await _flood(first, ctx, _CHAT_RATE_PER_ACCOUNT)
    assert not _errors(first)

    await _flood(second, ctx, 1)
    assert _errors(second)[-1]["code"] == "chat_rate_limited"


# ── and the reason it is per account ─────────────────────────────────────────

async def test_one_member_at_their_limit_has_not_spent_anyone_elses(ctx, store):
    """The property the whole bound exists for, and the reason there is no
    node-wide ceiling beside it: a member who floods costs themselves their own
    budget, and costs the group nothing it can notice."""
    alice = _session(ctx, store, user="alice", conn="c1")
    bob = _session(ctx, store, user="bob", conn="c2")

    await _flood(alice, ctx, _CHAT_RATE_PER_ACCOUNT + 5)
    assert _errors(alice)

    await _flood(bob, ctx, 1)
    assert not _errors(bob), "one member's flood silenced another"
    assert _acks(bob)