"""
"Key changed" is said on a fresh roster, never on a stale one.

The roster is read once per connection. A member invited after it was read was
absent from it, and each of their messages raised "This account is using a key
you have not seen before" until the page was reloaded: found live, twice in one
conversation. These tests load the real `transport-roster.js` and
`transport-devices.js` into node, with the roster the node would give before and
after the member joined, and check what the chat would show.

What must not move: a key the fresh roster cannot evidence is still "changed",
and a node that makes keys up gets one extra read per key, not one per message.
"""

import json
import shutil
import subprocess
from pathlib import Path

import pytest

STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"

pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node is not available")

HARNESS = """
import vm from 'node:vm';
import fs from 'node:fs';

const store = new Map();
const ctx = {
  console,
  localStorage: {
    getItem: (k) => (store.has(k) ? store.get(k) : null),
    setItem: (k, v) => store.set(k, String(v)),
  },
  window: {},
  extendTransport(part) { ctx.Part = part; },
};
vm.createContext(ctx);
for (const f of %(files)s) vm.runInContext(fs.readFileSync(f, 'utf8'), ctx);

const entry = (all, verified = all, chain = []) =>
  ({ all, verified, chain: new Map(chain) });
const roster = (accounts) => ({ byAccount: new Map(Object.entries(accounts)) });

function transport(stale, fresh) {
  const t = Object.create(ctx.Part.prototype);
  t.nodePk = 'NODE';
  t.freshReads = 0;
  t.groupRoster = async ({ fresh: wantFresh = false } = {}) => {
    if (!wantFresh) return stale;
    t.freshReads++;
    await new Promise((r) => setTimeout(r, 5));
    if (fresh instanceof Error) throw fresh;
    return fresh;
  };
  return t;
}
const pins = (user) => JSON.parse(store.get(`meshbay_account_pins:NODE:${user}`) || 'null');
"""


def _run(tmp_path, scenario):
    files = [str(STATIC / "transport-roster.js"), str(STATIC / "transport-devices.js")]
    script = tmp_path / "recheck.mjs"
    script.write_text(HARNESS % {"files": json.dumps(files)} + scenario, encoding="utf-8")
    proc = subprocess.run(["node", str(script)], capture_output=True, text=True,
                          encoding="utf-8")
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


def test_a_member_who_joined_after_the_roster_was_read_is_a_first_sight(tmp_path):
    out = _run(tmp_path, """
const t = transport(roster({}), roster({ bob: entry(['B']) }));
const first = await t.accountDeviceStatus('bob', 'B');
const second = await t.accountDeviceStatus('bob', 'B');
console.log(JSON.stringify({ first, second, reads: t.freshReads, pins: pins('bob') }));
""")
    assert out == {"first": "first", "second": "pinned", "reads": 1, "pins": ["B"]}


def test_two_messages_at_once_share_the_one_read(tmp_path):
    out = _run(tmp_path, """
const t = transport(roster({}), roster({ bob: entry(['B']) }));
const both = await Promise.all([t.accountDeviceStatus('bob', 'B'),
                                t.accountDeviceStatus('bob', 'B')]);
console.log(JSON.stringify({ both, reads: t.freshReads }));
""")
    assert "changed" not in out["both"], out
    assert out["reads"] == 1


def test_a_device_added_after_the_roster_was_read_is_linked(tmp_path):
    out = _run(tmp_path, """
store.set('meshbay_account_pins:NODE:bob', JSON.stringify(['A']));
const t = transport(roster({ bob: entry(['A']) }),
                    roster({ bob: entry(['A', 'B'], ['A', 'B'], [['B', 'A']]) }));
const status = await t.accountDeviceStatus('bob', 'B');
console.log(JSON.stringify({ status, pins: pins('bob') }));
""")
    assert out == {"status": "linked", "pins": ["A", "B"]}


def test_a_key_the_fresh_roster_cannot_evidence_is_still_changed(tmp_path):
    out = _run(tmp_path, """
store.set('meshbay_account_pins:NODE:bob', JSON.stringify(['A']));
const t = transport(roster({ bob: entry(['A']) }), roster({ bob: entry(['A', 'X'], ['A']) }));
const first = await t.accountDeviceStatus('bob', 'X');
const second = await t.accountDeviceStatus('bob', 'X');
console.log(JSON.stringify({ first, second, reads: t.freshReads, pins: pins('bob') }));
""")
    assert out == {"first": "changed", "second": "changed", "reads": 1, "pins": ["A"]}


def test_a_failed_re_read_keeps_the_warning(tmp_path):
    out = _run(tmp_path, """
const t = transport(roster({}), new Error('node gone'));
const status = await t.accountDeviceStatus('bob', 'B');
console.log(JSON.stringify({ status }));
""")
    assert out == {"status": "changed"}


def test_a_roster_that_already_knows_the_key_is_not_read_again(tmp_path):
    out = _run(tmp_path, """
const t = transport(roster({ bob: entry(['B']) }), roster({}));
const status = await t.accountDeviceStatus('bob', 'B');
console.log(JSON.stringify({ status, reads: t.freshReads }));
""")
    assert out == {"status": "first", "reads": 0}