"""
The hub can send mail from its own domain, to anywhere, with no authentication
to its local Postfix. That is an open relay unless something bounds it, and
what bounds it has to be one place, or the next `send_*` helper added will not
have it.

Three API paths reach that ability and two of them mail an address the caller
types:

  POST  /v1/users/register           an address nobody has verified
  PATCH /v1/users/me                 an address nobody has verified, signed in
  POST  /v1/users/password/reset     only the address already on file
  POST  /v1/groups/{id}/invite-notify   only a registered member's address

The widest was the register *resend* branch: no token, no captcha, and the
username and address are the caller's own from a moment ago, so registering a
victim's address once bought the right to mail them at the endpoint's rate
limit indefinitely.

The bound that matters is per **recipient**, because that is what a person
being mail-bombed experiences, and no combination of accounts, addresses or
endpoints moves it. Per-account and per-IP limits sit above it and bound a
caller, which is a different and weaker thing.
"""

import base64

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_hub import mail as mail_mod
from meshbay_hub.api import users as users_mod

# Captured at import, before conftest's autouse fixture replaces them for every
# other test in the suite: one stubs `_send` out entirely and the other skips
# the registration mail. This file is about the code they stub.
_REAL_SEND = mail_mod._send
_REAL_CREATE_AND_SEND = users_mod._create_and_send_verification


@pytest.fixture(autouse=True)
def _real_mail_path(monkeypatch):
    """Undo conftest's stubs, for the tests that go through the API.

    The direct tests below call `_REAL_SEND` rather than `mail_mod._send`:
    depending on a fixture to *un*-patch a module global is a coupling that
    passes alone and fails in a full run, which is how this file first behaved.
    """
    monkeypatch.setattr(mail_mod, "_send", _REAL_SEND)
    monkeypatch.setattr(users_mod, "_create_and_send_verification",
                        _REAL_CREATE_AND_SEND)
    assert mail_mod._send is _REAL_SEND


@pytest.fixture(autouse=True)
def _configured_limits():
    """What `hub.toml` says, which is what a missing settings row falls back to.

    Set at startup by the lifespan; pinned here so a test that never builds an
    app still measures against the shipped defaults rather than against zero.
    """
    from meshbay_hub import hub_settings
    from meshbay_hub.config import MailConfig

    hub_settings.set_mail_defaults(MailConfig())


@pytest.fixture
def wire(monkeypatch):
    """Everything that actually reaches Postfix, captured at the socket."""
    sent = []

    class _FakeSMTP:
        def __init__(self, *a, **kw):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *a):
            return False

        def send_message(self, msg):
            sent.append((msg["To"], msg["Subject"]))

    monkeypatch.setattr(mail_mod.smtplib, "SMTP", _FakeSMTP)
    return sent


def _msg(to: str):
    from email.message import EmailMessage
    m = EmailMessage()
    m["From"] = "noreply@test"
    m["To"] = to
    m["Subject"] = "test"
    m.set_content("test")
    return m


async def _charge(db, purpose: str, address: str) -> bool:
    """One send against the allowance. True if it was granted."""
    try:
        await mail_mod.reserve(db, purpose, address)
    except mail_mod.MailRefused:
        await db.rollback()
        return False
    await db.commit()
    return True


# ── The gate itself ──────────────────────────────────────────────────────────

def test_a_send_must_name_a_purpose(wire):
    """Keyword-required, so a helper that forgets is a TypeError rather than
    an unrestricted send."""
    with pytest.raises(TypeError):
        _REAL_SEND(_msg("someone@example.test"))


def test_send_refuses_a_purpose_this_hub_does_not_have(wire):
    """The structural half of the gate, and the half that needs no state. It
    is in `_send` so that nothing which puts a message on the wire can name a
    reason this hub does not send mail for; the counting half needs a database
    session and lives in `reserve`."""
    assert not _REAL_SEND(_msg("b@example.test"), purpose="marketing")
    assert not _REAL_SEND(_msg("c@example.test"), purpose="")
    assert _REAL_SEND(_msg("a@example.test"), purpose="registration")
    assert [to for to, _ in wire] == ["a@example.test"]


@pytest.mark.asyncio
async def test_reserve_refuses_a_purpose_this_hub_does_not_have(db_session):
    with pytest.raises(mail_mod.MailRefused):
        await mail_mod.reserve(db_session, "marketing", "a@example.test")


@pytest.mark.asyncio
async def test_one_recipient_has_a_cooldown_and_a_daily_allowance(db_session):
    """Across every purpose — the recipient is what is protected, not the
    sender."""
    from meshbay_hub.config import MailConfig
    from meshbay_hub.db.models import MailQuota

    victim = "victim@example.test"
    assert await _charge(db_session, "registration", victim)
    assert not await _charge(db_session, "password_reset", victim), (
        "a second purpose walked around the cooldown")

    cap = MailConfig().destination_daily_cap
    cool = "cool:invitation:" + mail_mod.destination_key(victim)[len("dest:"):]

    async def past_the_cooldown():
        # Step past it without waiting it out. The daily count is deliberately
        # left alone.
        row = await db_session.get(MailQuota, cool)
        if row:
            row.last_sent = None
            await db_session.commit()

    for _ in range(cap - 1):
        await past_the_cooldown()
        assert await _charge(db_session, "invite", victim)

    await past_the_cooldown()
    assert not await _charge(db_session, "invite", victim), (
        f"the {cap}-a-day allowance for one recipient was not enforced")


@pytest.mark.asyncio
async def test_an_invitation_does_not_stop_the_code_its_invitee_asks_for(db_session):
    """
    The regression: an invitation link arrives, the invitee registers a minute
    later, and the verification code was refused — silently — by the cooldown
    the invitation had just armed. Nobody invited by link could sign up without
    waiting two minutes and then asking again, with nothing on screen saying so.

    The two are one conversation, so they do not share a cooldown; each family
    still keeps its own, and the daily cap still counts both.
    """
    async def charge(purpose, addr):
        try:
            await mail_mod.reserve(db_session, purpose, addr, sender="owner-1")
        except mail_mod.MailRefused:
            await db_session.rollback()
            return False
        await db_session.commit()
        return True

    for invitation in ("invite", "invite_link"):
        addr = f"newcomer-{invitation}@example.test"
        assert await charge(invitation, addr)
        assert await charge("registration", addr), (
            f"a verification code was refused after an {invitation} mail")
        assert not await charge("password_reset", addr), (
            "within the account family the cooldown must still hold")
        assert not await charge("invite", addr), (
            "a second invitation in the same burst must still wait")


@pytest.mark.asyncio
async def test_the_same_recipient_is_one_recipient_however_it_is_written(db_session):
    """Case and surrounding space are not a new person."""
    assert await _charge(db_session, "registration", "Victim@Example.Test")
    assert not await _charge(db_session, "registration", " victim@example.test ")


@pytest.mark.asyncio
async def test_the_instance_has_a_ceiling_no_account_can_buy_past(db_session):
    """Registration is open, so "per account" is a bound an attacker buys more
    of. This one is not."""
    from meshbay_hub.config import MailConfig
    cfg = MailConfig()
    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery

    for i in range(general):
        assert await _charge(db_session, "registration", f"u{i}@example.test")
    assert not await _charge(db_session, "registration", "one-more@example.test")


@pytest.mark.asyncio
async def test_a_flood_of_sign_ups_cannot_lock_out_a_passphrase_reset(db_session):
    """The reserved share: someone waiting on a reset code does not wait
    because strangers were registering all hour."""
    from meshbay_hub.config import MailConfig
    cfg = MailConfig()
    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery

    for i in range(general):
        assert await _charge(db_session, "registration", f"flood{i}@example.test")
    assert not await _charge(db_session, "registration", "another@example.test")

    assert await _charge(db_session, "password_reset", "waiting@example.test"), (
        "a sign-up flood spent the share kept for recovery")
    assert await _charge(db_session, "invite", "invited@example.test")


@pytest.mark.asyncio
async def test_the_allowance_is_written_down_and_survives_a_restart(db_session):
    """It was two module dicts, so every deploy handed out a fresh budget —
    and the hub is deployed often."""
    from meshbay_hub.db.models import MailQuota

    assert await _charge(db_session, "registration", "kept@example.test")

    # A restart loses everything that was not written down. These were.
    assert await db_session.get(MailQuota, "hour") is not None
    assert await db_session.get(
        MailQuota, mail_mod.destination_key("kept@example.test")) is not None
    assert not await _charge(db_session, "registration", "kept@example.test")


@pytest.mark.asyncio
async def test_a_bound_an_admin_changed_is_the_one_that_applies(db_session):
    """These are settings, not constants: the hour a budget runs out is not
    when anyone wants to edit a file and restart."""
    from meshbay_hub import hub_settings

    await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
    await hub_settings.set_raw(db_session, "mail.destination_daily_cap", "2")
    await db_session.commit()

    assert await _charge(db_session, "registration", "tuned@example.test")
    assert await _charge(db_session, "registration", "tuned@example.test")
    assert not await _charge(db_session, "registration", "tuned@example.test")


def test_a_refusal_never_names_the_address():
    """This line goes to the journal, which is not where addresses live.

    Listens on the module's own logger rather than through `caplog`: the hub
    configures logging when an app is built, so whether pytest's capture sees
    anything depends on which tests ran first. In the full suite it saw
    nothing, and the assertion that had to hold — no address in the line —
    was never reached.
    """
    import logging

    records: list[str] = []

    class _Capture(logging.Handler):
        def emit(self, record):
            records.append(record.getMessage())

    # And the line as emitted. Level and `logging.disable` are both forced:
    # attaching a handler is not enough when an earlier test has raised the
    # logger's level, which is why the first version of this saw nothing in
    # the full suite and everything on its own.
    handler = _Capture()
    previous_level, previously_disabled = mail_mod.log.level, mail_mod.log.disabled
    mail_mod.log.addHandler(handler)
    mail_mod.log.setLevel(logging.WARNING)
    # `.disabled` as well as the level and the manager-wide switch. Something
    # earlier in a full run leaves this logger disabled — pytest's own capture
    # plugin toggles loggers between tests — and a disabled logger drops the
    # record before any handler sees it. Alone, none of this is needed; the
    # first two versions of this test passed alone and failed in the suite for
    # two different reasons, which is the whole argument for not reading a
    # global's state in a test without pinning it first.
    mail_mod.log.disabled = False
    logging.disable(logging.NOTSET)
    try:
        _REAL_SEND(_msg("private@example.test"), purpose="marketing")
    finally:
        mail_mod.log.removeHandler(handler)
        mail_mod.log.setLevel(previous_level)
        mail_mod.log.disabled = previously_disabled

    assert records, "a refusal must leave a trace"
    assert not any("private@example.test" in line for line in records)
    assert any("marketing" in line for line in records), (
        "the trace has to say what was refused to be worth writing")


# ── The doors, driven through the API ────────────────────────────────────────

async def _register(client, username: str, email: str, captcha=None):
    sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate()
    return await client.post("/v1/users/register", json={
        "username": username,
        "email": email,
        "auth_key": base64.b64encode(b"k" * 32).decode(),
        "pk_user_ed25519": pk_to_b64(sk_ed.public_key()),
        "pk_user_x25519": pk_to_b64(sk_x.public_key()),
    })


@pytest.mark.asyncio
async def test_the_register_resend_branch_cannot_be_hammered(client, wire):
    """
    No token and no captcha reach this branch. Registering a victim's address
    once used to buy the right to mail them at the endpoint's rate limit for
    as long as the account stayed pending.
    """
    victim = "bombing-target@example.test"
    r = await _register(client, "relay_a_test", victim)
    assert r.status_code == 201, r.text
    assert len(wire) == 1

    for _ in range(5):
        r = await _register(client, "relay_a_test", victim)
        assert r.status_code == 201, r.text
    assert len(wire) == 1, f"{len(wire)} mails to one address from one sign-up"


async def _signed_in(client, db_session, username: str, email: str) -> dict:
    """Register, activate, sign in.

    Activated by hand because this file restores the real verification mail,
    which conftest stubs out for the rest of the suite — so the account is
    genuinely `pending` here, exactly as it would be in production, and a
    pending account cannot log in.
    """
    from meshbay_hub.db.models import User
    from sqlalchemy import select, update

    r = await _register(client, username, email)
    assert r.status_code == 201, r.text
    await db_session.execute(
        update(User).where(User.username == username).values(status="active"))
    await db_session.commit()
    assert (await db_session.execute(
        select(User.status).where(User.username == username))).scalar() == "active"

    r = await client.post("/v1/users/login", json={
        "username": username,
        "auth_key": base64.b64encode(b"k" * 32).decode()})
    assert r.status_code == 200, r.text
    return {"Authorization": f"Bearer {r.json()['access_token']}"}


@pytest.mark.asyncio
async def test_an_account_may_point_the_hub_at_one_stranger_then_wait(
        client, db_session, wire):
    """
    `PATCH /v1/users/me` is the only path where a signed-in account chooses the
    recipient. A short delay alone still allows one stranger per interval,
    indefinitely, so the delay is long — and a second, different address
    inside it is refused.
    """
    headers = await _signed_in(client, db_session, "relay_b_test",
                               "relay_b@example.test")
    before = len(wire)

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "a-stranger@example.test"})
    assert r.status_code == 200, r.text

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "another-stranger@example.test"})
    assert r.status_code == 429, r.text
    assert len(wire) - before == 1, "the hub mailed a second stranger on demand"


@pytest.mark.asyncio
async def test_the_address_already_pending_may_be_asked_for_again(
        client, db_session, wire):
    """
    The exemption, and why it is not a hole: re-asking for a code for the
    address already pending reaches no new recipient, and that recipient is
    bounded by its own daily allowance regardless. Without it, a typo would
    lock the account out of correcting it for the whole window.
    """
    from meshbay_hub import hub_settings
    from meshbay_hub.db.models import MailQuota

    headers = await _signed_in(client, db_session, "relay_d_test",
                               "relay_d@example.test")
    typo = "jean@gmial.test"

    r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
    assert r.status_code == 200, r.text

    # The recipient's own cooldown is not what is under test here.
    await hub_settings.set_raw(db_session, "mail.destination_cooldown_seconds", "0")
    await db_session.commit()
    row = await db_session.get(MailQuota, mail_mod.destination_key(typo))
    row.last_sent = None
    await db_session.commit()

    before = len(wire)
    r = await client.patch("/v1/users/me", headers=headers, json={"email": typo})
    assert r.status_code == 200, (
        "a typo locked the account out of correcting it: " + r.text)
    assert len(wire) - before == 1


@pytest.mark.asyncio
async def test_the_delay_survives_the_verification_row_being_deleted(
        client, db_session, wire):
    """
    The handler deletes this account's unverified rows before writing a new
    one, so a window counted from that table would count one, always. Which is
    what the first version of this counted. It comes off the IP log.
    """
    headers = await _signed_in(client, db_session, "relay_c_test",
                               "relay_c@example.test")

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "c-first@example.test"})
    assert r.status_code == 200, r.text

    from meshbay_hub.db.models import EmailVerification
    from sqlalchemy import delete
    await db_session.execute(delete(EmailVerification))
    await db_session.commit()

    r = await client.patch("/v1/users/me", headers=headers,
                           json={"email": "c-second@example.test"})
    assert r.status_code == 429, (
        "the delay was counted from a table the handler empties")


# ── The operator's controls ──────────────────────────────────────────────────

async def _admin(client, db_session, username: str) -> dict:
    from meshbay_hub.db.models import User
    from sqlalchemy import update

    headers = await _signed_in(client, db_session, username,
                               f"{username}@example.test")
    await db_session.execute(
        update(User).where(User.username == username).values(role="admin"))
    await db_session.commit()
    return headers


@pytest.mark.asyncio
async def test_the_panel_shows_the_bounds_their_defaults_and_the_range(
        client, db_session):
    """The panel draws the fields from this, so a value it cannot store is one
    the operator is told about rather than one that is silently clamped."""
    headers = await _admin(client, db_session, "mailadmin1")

    r = await client.get("/v1/admin/settings", headers=headers)
    assert r.status_code == 200, r.text
    body = r.json()
    from meshbay_hub import hub_settings
    assert set(body["mail"]) == set(hub_settings.MAIL_KEYS)
    assert set(body["mail_defaults"]) == set(hub_settings.MAIL_KEYS)
    assert body["mail_bounds"]["hourly_budget"][0] >= 1, (
        "a budget of zero would stop the hub sending anything at all")


@pytest.mark.asyncio
async def test_an_admin_changes_a_bound_and_it_takes_effect(client, db_session):
    headers = await _admin(client, db_session, "mailadmin2")

    r = await client.patch("/v1/admin/settings", headers=headers,
                           json={"mail": {"destination_daily_cap": 3}})
    assert r.status_code == 200, r.text
    assert r.json()["mail"]["destination_daily_cap"] == 3

    from meshbay_hub import hub_settings
    assert (await hub_settings.mail_limits(db_session))[
        "destination_daily_cap"] == 3


@pytest.mark.asyncio
async def test_a_value_outside_the_range_is_clamped_not_stored(client, db_session):
    """A number typed into a web form is not a reason to send without a bound."""
    headers = await _admin(client, db_session, "mailadmin3")

    r = await client.patch("/v1/admin/settings", headers=headers,
                           json={"mail": {"hourly_budget": 0,
                                          "destination_daily_cap": 10 ** 9}})
    assert r.status_code == 200, r.text
    from meshbay_hub import hub_settings
    low, _ = hub_settings.MAIL_BOUNDS["hourly_budget"]
    _, high = hub_settings.MAIL_BOUNDS["destination_daily_cap"]
    assert r.json()["mail"]["hourly_budget"] == low
    assert r.json()["mail"]["destination_daily_cap"] == high


@pytest.mark.asyncio
async def test_an_unknown_mail_setting_is_refused(client, db_session):
    headers = await _admin(client, db_session, "mailadmin4")
    r = await client.patch("/v1/admin/settings", headers=headers,
                           json={"mail": {"send_to_everyone": 1}})
    assert r.status_code == 422, r.text


@pytest.mark.asyncio
async def test_a_moderator_may_read_the_bounds_but_not_change_them(
        client, db_session):
    from meshbay_hub.db.models import User
    from sqlalchemy import update

    headers = await _signed_in(client, db_session, "mailmod_test",
                               "mailmod@example.test")
    await db_session.execute(
        update(User).where(User.username == "mailmod_test").values(role="moderator"))
    await db_session.commit()

    assert (await client.get("/v1/admin/settings", headers=headers)).status_code == 200
    r = await client.patch("/v1/admin/settings", headers=headers,
                           json={"mail": {"hourly_budget": 10}})
    assert r.status_code == 403, r.text


@pytest.mark.asyncio
async def test_the_operator_can_see_whether_the_hub_is_still_sending(
        client, db_session, wire):
    """There was no way to see this at all: a refusal was a line in the
    journal, so an instance that had stopped sending sign-up codes looked,
    from the panel, exactly like one with no sign-ups."""
    headers = await _admin(client, db_session, "mailadmin5")

    r = await client.get("/v1/admin/mail", headers=headers)
    assert r.status_code == 200, r.text
    before = r.json()
    # A delta, not an absolute: creating the admin above signed somebody up,
    # which sent a verification code and spent one of the hour's allowance.
    assert before["general_remaining"] < before["recovery_remaining"], (
        "the reserved share must be visible as the difference between the two")

    assert await _charge(db_session, "registration", "seen@example.test")

    after = (await client.get("/v1/admin/mail", headers=headers)).json()
    assert after["hourly_used"] == before["hourly_used"] + 1
    assert after["general_remaining"] == before["general_remaining"] - 1
    assert after["recipients_tracked"] > before["recipients_tracked"]


def test_the_hub_refuses_to_start_with_more_than_one_worker():
    """The node registry and the signaling futures are per-process dicts, so a
    second worker makes a node intermittently unreachable for half its
    members — a symptom that describes something else entirely."""
    from meshbay_hub.daemon import single_worker_or_exit

    single_worker_or_exit(1)
    for workers in (0, 2, 8):
        with pytest.raises(SystemExit) as exit_:
            single_worker_or_exit(workers)
        assert exit_.value.code == 2


# ── The operator is told, not left to notice ─────────────────────────────────

@pytest.mark.asyncio
async def test_spending_the_sign_up_share_notifies_the_administrators(
        client, db_session):
    """
    A refusal is otherwise a line in the journal. A hub that has stopped
    sending sign-up codes looks, from every screen anyone opens, exactly like
    one nobody is signing up to.
    """
    from meshbay_hub.config import MailConfig

    headers = await _admin(client, db_session, "mailwatcher")
    await client.delete("/v1/notifications", headers=headers)

    cfg = MailConfig()
    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
    for i in range(general):
        await _charge(db_session, "registration", f"ceiling{i}@example.test")
    assert not await _charge(db_session, "registration", "over@example.test")

    r = await client.get("/v1/notifications", headers=headers)
    assert r.status_code == 200, r.text
    kinds = [n["kind"] for n in r.json()["notifications"]]
    assert "mail_budget_general" in kinds, (
        "the sign-up ceiling fell and nobody was told")
    assert "mail_budget_all" not in kinds, (
        "recovery still has its share; saying otherwise would be alarming and "
        "wrong")


@pytest.mark.asyncio
async def test_the_administrators_are_told_once_an_hour_not_once_a_refusal(
        client, db_session):
    """A flood is what spends the budget, so a notification per refusal would
    bury the one that matters under the thing that caused it."""
    from meshbay_hub.config import MailConfig

    headers = await _admin(client, db_session, "mailwatcher2")
    await client.delete("/v1/notifications", headers=headers)

    cfg = MailConfig()
    general = cfg.hourly_budget - cfg.hourly_reserved_for_recovery
    for i in range(general):
        await _charge(db_session, "registration", f"burst{i}@example.test")
    for i in range(5):
        assert not await _charge(db_session, "registration", f"over{i}@example.test")

    r = await client.get("/v1/notifications", headers=headers)
    general_alerts = [n for n in r.json()["notifications"]
                      if n["kind"] == "mail_budget_general"]
    assert len(general_alerts) == 1, f"{len(general_alerts)} alerts for one hour"


@pytest.mark.asyncio
async def test_the_panel_says_which_ceiling_has_fallen(client, db_session):
    """Two states, and the difference matters to whoever is reading: one means
    newcomers are turned away, the other means somebody locked out of their
    account cannot get back in."""
    from meshbay_hub.config import MailConfig

    headers = await _admin(client, db_session, "mailwatcher3")
    cfg = MailConfig()

    for i in range(cfg.hourly_budget - cfg.hourly_reserved_for_recovery):
        await _charge(db_session, "registration", f"state{i}@example.test")
    body = (await client.get("/v1/admin/mail", headers=headers)).json()
    assert body["general_exhausted"] and not body["all_exhausted"]

    for i in range(cfg.hourly_reserved_for_recovery):
        await _charge(db_session, "password_reset", f"rec{i}@example.test")
    body = (await client.get("/v1/admin/mail", headers=headers)).json()
    assert body["all_exhausted"]