"""The stream pushes the live facts a watching client used to poll for."""

import http.client
import json
import queue
import sqlite3
import threading
import time
import unittest
import urllib.error
import urllib.parse
import uuid

from harness import (
    ROOT,
    Albedo,
    Provider,
    exclusive,
    latest_user,
    python,
    text,
)
from inspect_support import Inspection

# Long enough for the daemon's sidebar pace plus a loaded test host.
PUSH_SECONDS = 25


class Stream:
    """One session subscription, read in the background so a test can wait."""

    def __init__(self, app, session):
        self.connection = http.client.HTTPConnection(
            "127.0.0.1", app.connection["port"]
        )
        self.connection.request(
            "GET",
            f"/sessions/{session}?tail=0",
            headers={
                "Authorization": "Bearer " + app.connection["token"],
                "Accept": "text/event-stream",
            },
        )
        self.response = self.connection.getresponse()
        self.batches = queue.Queue()
        self.reader = threading.Thread(target=self._read, daemon=True)
        self.reader.start()

    def _read(self):
        try:
            for line in self.response:
                if line.startswith(b"data: "):
                    self.batches.put(json.loads(line[6:]))
        except (OSError, ValueError, AttributeError):
            # The stream was closed under the reader at cleanup; http.client
            # reports that as a missing buffer rather than a socket error.
            return

    def batch(self, seconds=PUSH_SECONDS):
        try:
            return self.batches.get(timeout=seconds)
        except queue.Empty:
            raise AssertionError(f"no stream batch within {seconds}s") from None

    def pushed(self, matches, seconds=PUSH_SECONDS):
        deadline = time.monotonic() + seconds
        while time.monotonic() < deadline:
            for event in self.batch(deadline - time.monotonic())["events"]:
                if matches(event):
                    return event
        raise AssertionError(f"no pushed event within {seconds}s")

    def close(self):
        self.connection.close()


class SessionStatusPushTests(unittest.TestCase):
    def setUp(self):
        def script(request):
            if request["messages"][-1].get("role") == "tool":
                return text("job started")
            if "start job" in latest_user(request):
                # A job that outlives the cell: the kernel reports it live, and
                # its end is the change a watching client must hear about.
                return python("handle = run('sleep', '4')\nprint(handle.id)")
            return text("done")

        self.provider = Provider(script)
        self.addCleanup(self.provider.close)

    def subscribe(self, app, session):
        stream = Stream(app, session)
        self.addCleanup(stream.close)
        return stream

    def activate_work_item(self, app, title):
        """An item the work sidebar lists: it shows only what is active."""
        workspace = urllib.parse.quote(str(app.workspace), safe="")
        with app.api(
            f"/extensions/work/items?workspace={workspace}", {"title": title}
        ) as created:
            resource = json.load(created)["resource"]
        self.assertEqual(resource["value"]["title"], title)
        with app.api(
            f"/extensions/work/items/{resource['value']['id']}?workspace={workspace}",
            {"status": "active"},
            method="PATCH",
            headers={"If-Match": resource["etag"]},
        ) as changed:
            self.assertEqual(
                json.load(changed)["resource"]["value"]["status"], "active"
            )

    @exclusive
    def test_quiet_session_hibernates_and_handles_the_next_turn(self):
        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"

        with Albedo(self.provider, prepare=prepare) as app:
            session = app.session()
            app.prompt(session, "first turn").close()
            app.idle(session)
            probe = Inspection(app)
            expression = (
                "io:put_chars(Call(erlang, apply, [fun() -> "
                "{some, {subject, Pid, _}} = 'albedo@daemon@session':live("
                + "<<"
                + json.dumps(session)
                + ">>"
                + "), {current_function, {_, F, _}} = process_info(Pid, current_function), "
                "json:encode(atom_to_binary(F)) end, []]))"
            )
            deadline = time.monotonic() + 10
            while True:
                current = json.loads(probe.evaluate(expression))
                # hibernate/0 keeps the caller's frame, so the actor waits in quiet.
                if current == "quiet":
                    break
                if time.monotonic() >= deadline:
                    self.fail(f"quiet session did not hibernate: {current}")
                time.sleep(0.05)
            app.prompt(session, "second turn").close()
            app.idle(session)
            self.assertEqual(len(self.provider.requests), 2)
            self.assertIn(
                "second turn", latest_user(self.provider.requests[-1]["request"])
            )

    @exclusive
    def test_kernel_notifications_do_not_retain_session_state(self):
        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"

        with Albedo(self.provider, prepare=prepare) as app:
            session = app.session()
            # A job's cell opens a kernel, which the daemon then watches.
            app.prompt(session, "start job").close()
            app.idle(session)
            sizes = json.loads(
                Inspection(app).evaluate(
                    "io:put_chars(Call(erlang, apply, [fun() -> "
                    "json:encode([erts_debug:flat_size(F) || {_, F} <- "
                    "ets:tab2list(albedo_kernel_watch)]) end, []]))"
                )
            )
            self.assertTrue(sizes)
            self.assertLess(max(sizes), 256)

    def test_stream_pushes_job_lifecycle_and_sidebar_changes(self):
        with Albedo(self.provider) as app:
            session = app.session()
            stream = self.subscribe(app, session)
            # The initial batch is still a complete snapshot: a fresh client
            # never needs a read to fill its footer.
            first = stream.batch()
            self.assertEqual(first["snapshot"]["id"], session)
            self.assertIn("kernel", first["snapshot"])
            self.assertIn("glances", first["snapshot"])

            app.prompt(session, "start job").close()
            started = stream.pushed(
                lambda event: (
                    event["type"] == "kernel"
                    and (event["data"]["kernel"]["live_job_count"] or 0) >= 1
                )
            )
            self.assertEqual(started["data"]["kernel"]["state"], "attached")
            self.assertTrue(started["data"]["kernel"]["running_jobs"])

            app.idle(session)
            ended = stream.pushed(
                lambda event: (
                    event["type"] == "kernel"
                    and event["data"]["kernel"]["live_job_count"] == 0
                )
            )
            self.assertEqual(ended["data"]["kernel"]["running_jobs"], [])

            self.activate_work_item(app, "pushed glance")

            def shows_work(event):
                return event["type"] == "glances" and any(
                    row["text"] == "pushed glance"
                    for glance in event["data"]["glances"]
                    for row in glance["rows"]
                )

            event = stream.pushed(shows_work)
            glance = next(
                glance
                for glance in event["data"]["glances"]
                if glance["extension"] == "work"
            )
            self.assertEqual(glance["title"], "active work")


# exclusive: holds the daemon's one runtime actor still
@exclusive
class RuntimeStallTests(unittest.TestCase):
    def test_status_reads_wait_out_a_busy_runtime(self):
        # A runtime whose mailbox is stalled can take seconds to answer.
        # A session without a kernel asks it what is loaded;
        # asking from the session's own actor failed every read of that session
        # ("session capture unavailable") once the runtime took 5 s.
        def prepare(app):
            app.daemon.env["ALBEDO_INSPECT"] = "1"

        with Albedo(prepare=prepare) as app:
            probe = Inspection(app, ROOT / "test/e2e/albedo_runtime_stall_probe.erl")
            session = app.session()
            reads = {}

            def read(view):
                try:
                    with app.api(f"/sessions/{session}{view}") as response:
                        reads[view] = (response.status, json.load(response))
                except urllib.error.HTTPError as error:
                    reads[view] = (error.code, json.load(error))

            self.assertEqual(probe.call_json("suspend_json"), {"suspended": True})
            try:
                readers = [
                    threading.Thread(target=read, args=(view,))
                    for view in ("", "?view=status")
                ]
                for reader in readers:
                    reader.start()
                time.sleep(7)
            finally:
                self.assertEqual(probe.call_json("resume_json"), {"resumed": True})
            for reader in readers:
                reader.join(30)
            self.assertEqual(reads[""][0], 200, reads[""])
            self.assertEqual(reads["?view=status"][0], 200, reads["?view=status"])
            self.assertEqual(reads["?view=status"][1]["kernel"]["state"], "none")


if __name__ == "__main__":
    unittest.main()


# exclusive: fills the store with thousands of sessions
@exclusive
class CollectionStreamTests(unittest.TestCase):
    def test_the_sessions_stream_starts_however_many_sessions_it_lists(self):
        # Following the collection reads every session it covers. That once ran
        # where mist gives a chunked response one second to start, and a big
        # store (or a busy host) answered every watcher with an empty stream.
        with Albedo() as app:
            seed = app.session()

            def fill(app):
                with sqlite3.connect(app.home / "albedo.sqlite") as db:
                    columns = [
                        row[1] for row in db.execute("PRAGMA table_info(sessions)")
                    ]
                    row = db.execute(
                        f"SELECT {','.join(columns)} FROM sessions WHERE id=?", (seed,)
                    ).fetchone()
                    at = columns.index("id")
                    db.executemany(
                        f"INSERT INTO sessions({','.join(columns)}) "
                        f"VALUES({','.join('?' * len(columns))})",
                        (
                            [*row[:at], str(uuid.uuid4()), *row[at + 1 :]]
                            for _ in range(8000)
                        ),
                    )

            app.restart(prepare=fill)
            connection = http.client.HTTPConnection(
                "127.0.0.1", app.connection["port"], timeout=60
            )
            self.addCleanup(connection.close)
            connection.request(
                "GET",
                "/sessions?scope=all",
                headers={
                    "Authorization": "Bearer " + app.connection["token"],
                    "Accept": "text/event-stream",
                },
            )
            first = next(
                (
                    line
                    for line in connection.getresponse()
                    if line.startswith(b"data: ")
                ),
                None,
            )
            if first is None:
                self.fail("the sessions stream ended without a batch")
            events = [event["type"] for event in json.loads(first[6:])["events"]]
            self.assertEqual(events[:2], ["ready", "reset"])