"""Tests for the load-bearing event-bus contracts:

- Bug 3 (now Candidate B): a `question` event MUST be persisted to the
  DB BEFORE its card is broadcast. The frontend renders the card on
  broadcast; a user Submit that races the save would find no question
  block in `chat.messages` and the answer would be silently dropped. C2
  enforces this via `_ChatEventSink.publish_question`, which submits a
  `QuestionCommit` to the writer actor, AWAITS its ack, and only then
  broadcasts — and `publish()` REJECTS question events so no runner can
  bypass the barrier.

- Bug 4: System events (theme_updated, app_updated, shell_*) MUST
  reach the `SystemBroadcast` so a Shell subscriber sees them
  regardless of whether a per-chat broadcast is currently active.

- Commit offload (C2): the streaming save's blocking `db.commit()` runs
  OFF the event loop on the writer-actor thread. `publish()` submits a
  fire-and-forget `PersistTranscript` (or `PersistError`). A slow commit
  on one chat can't stall the loop and starve other chats' SSE because it
  never runs on the loop.
"""

import asyncio

import pytest
from pydantic import ValidationError

from app import broadcast as bc_mod
from app import chat as chat_mod
from app import models
from app.broadcast import (
  ChatBroadcast,
  SystemBroadcast,
  clear_active_broadcast_if,
  get_active_broadcast,
  get_system_broadcast,
  set_active_broadcast,
)
from app.chat_writer import Barrier, get_writer
from app.deps import Principal
from app.routes.notify import NotifyBody


# --- Bug 3 / Candidate B: question save-before-broadcast --------------


class _OrderedBroadcast(ChatBroadcast):
  """ChatBroadcast that records each publish so a test can inspect what
  reached the wire (and in what order relative to the actor commit)."""

  def __init__(self, chat_id):
    super().__init__(chat_id)
    self.timeline = []
    # At the instant a `question` event is broadcast, was the question
    # block ALREADY durable? A fresh-session re-read taken inside publish()
    # is the only way to ENFORCE the ordering. Two independent post-hoc
    # checks — the block is persisted after publish_question returns AND
    # the event reached the timeline — both pass even for a regression
    # that broadcasts BEFORE awaiting the QuestionCommit ack, because by
    # the time the test re-reads the DB the (out-of-order) commit has
    # already landed. Capturing durability AT broadcast time is what
    # closes that gap.
    self.question_block_persisted_at_publish = None

  def publish(self, event):
    if event.get("type") == "question":
      from app.database import SessionLocal
      s = SessionLocal()
      try:
        row = (
          s.query(models.Chat)
          .filter(models.Chat.id == self.chat_id)
          .one_or_none()
        )
        blocks = (
          row.messages[-1].get("blocks") if row and row.messages else []
        ) or []
        self.question_block_persisted_at_publish = any(
          b.get("type") == "question" for b in blocks
        )
      finally:
        s.close()
    self.timeline.append(("publish", event.get("type")))
    super().publish(event)


def _drain_actor():
  """Block until the writer actor has processed everything queued so far.

  publish() submits fire-and-forget commands; a Barrier ack resolves only
  after every prior command has been dispatched, so awaiting it from a
  test gives a deterministic "the commit has landed" point."""
  get_writer().submit(Barrier()).result(timeout=5)


def test_question_event_is_saved_before_broadcast(db, chat):
  """User-Submit-races-save regression: an AskUserQuestion card must land
  in chat.messages BEFORE the broadcast reaches a frontend that might
  POST an answer. C2 enforces this in publish_question via QuestionCommit
  (commit-before-ack); the card is broadcast only after the ack."""
  chat.messages = [{"role": "user", "content": "hi", "ts": 1}]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-q")

  questions = [{
    "id": "q1",
    "question": "Color?",
    "options": [{"label": "Red"}, {"label": "Blue"}],
  }]

  async def go():
    # When publish_question returns, the QuestionCommit ack has resolved
    # (the block is durable) AND the card has been broadcast — in that
    # order, by construction.
    await sink.publish_question(
      {"type": "question", "question_id": "q1", "questions": questions}
    )
    # The block is already persisted at this point: re-read on a fresh
    # session proves it landed before the broadcast that just happened.
    from app.database import SessionLocal
    s = SessionLocal()
    try:
      row = s.query(models.Chat).filter(models.Chat.id == chat.id).one()
      blocks = row.messages[-1].get("blocks") or []
      assert any(b.get("type") == "question" for b in blocks), (
        "question block must be persisted before the broadcast"
      )
    finally:
      s.close()

  asyncio.run(go())
  # The card reached the wire.
  assert ("publish", "question") in bc.timeline
  # The ORDERING guarantee: the question block was already durable AT the
  # moment its card was broadcast. This is what makes the test catch a
  # regression that reorders publish() ahead of the QuestionCommit ack —
  # the two checks above would still pass for such a regression.
  assert bc.question_block_persisted_at_publish is True, (
    "save-before-broadcast: the question block MUST be durable at the moment "
    "its card is broadcast — broadcasting before the QuestionCommit ack would "
    "let a racing user Submit find no question block to attach the answer to"
  )


def test_publish_rejects_question_events(db, chat):
  """publish() must REJECT question events so a runner can't bypass the
  save-before-broadcast barrier — they MUST go through publish_question."""
  chat.messages = [{"role": "user", "content": "hi", "ts": 1}]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-q")
  with pytest.raises(AssertionError):
    sink.publish({"type": "question", "questions": [{"id": "q1"}]})


def test_non_question_save_routes_to_actor_off_loop(db, chat):
  """Non-question events broadcast immediately and route their DB commit
  to the writer actor (off the event loop). publish() submits a
  fire-and-forget PersistTranscript; the block is durable once the actor
  drains."""
  chat.messages = [{"role": "user", "content": "hi", "ts": 1}]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-t")

  sink._last_save = 0.0
  ok = sink.publish({"type": "tool_start", "tool": "Bash", "input": "ls"})
  assert ok is True
  # Broadcast happened synchronously on the calling thread.
  assert ("publish", "tool_start") in bc.timeline
  # The actor commits off-loop; drain it, then the tool block is durable.
  _drain_actor()
  db.expire_all()
  row = db.query(models.Chat).filter(models.Chat.id == chat.id).one()
  assert row.messages and row.messages[-1].get("role") == "assistant"
  assert any(b.get("type") == "tool" for b in row.messages[-1]["blocks"])


def test_streaming_commit_runs_off_the_event_loop_thread(db, chat):
  """The streaming commit must execute on the writer-actor thread, not
  the loop thread — that is the whole point of the offload (a slow SQLite
  commit can't stall the loop). The actor owns a dedicated thread named
  'chat-writer'; the publish call on the loop only enqueues."""
  import threading

  chat.messages = [{"role": "user", "content": "hi", "ts": 1}]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-t")

  seen: dict = {}
  from app import chat_writer as chat_writer_mod
  original = chat_writer_mod.update_last_assistant_message

  def spy(db_, chat_id, message):
    seen["commit_thread"] = threading.get_ident()
    return original(db_, chat_id, message)

  chat_writer_mod.update_last_assistant_message = spy
  try:
    seen["loop_thread"] = threading.get_ident()
    sink._last_save = 0.0
    sink.publish({"type": "tool_start", "tool": "Bash", "input": "ls"})
    _drain_actor()
  finally:
    chat_writer_mod.update_last_assistant_message = original

  assert "commit_thread" in seen, "the actor never ran the commit"
  assert seen["commit_thread"] != seen["loop_thread"], (
    "the streaming commit ran on the event loop thread — it must run on "
    "the writer-actor thread so a SQLite lock can't stall the loop"
  )


def test_finalize_awaits_actor_and_persists(db, chat):
  """finalize() submits a Finalize and AWAITS its ack: the terminal
  message is durable the instant finalize() returns (commit-before-ack)."""
  chat.messages = [{"role": "user", "content": "hi", "ts": 1}]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-f")
  sink.assistant_blocks = [{"type": "text", "content": "done"}]

  asyncio.run(sink.finalize())

  db.expire_all()
  row = db.query(models.Chat).filter(models.Chat.id == chat.id).one()
  assert row.messages[-1].get("role") == "assistant"
  assert row.messages[-1]["blocks"][-1]["content"] == "done"


def test_error_event_routes_to_persist_error(db, chat):
  """An `error` event routes to a PersistError (non-coalescing) so it
  can't be collapsed away by a later text snapshot; it lands durably."""
  chat.messages = [
    {"role": "user", "content": "hi", "ts": 1},
    {"role": "assistant", "content": "", "ts": 2,
     "blocks": [{"type": "text", "content": "partial"}]},
  ]
  db.commit()
  bc = _OrderedBroadcast(chat.id)
  sink = chat_mod._ChatEventSink(bc, chat.id, run_token="rt-e")
  sink.assistant_blocks = [{"type": "text", "content": "partial"}]

  sink._last_save = 0.0
  sink.publish({"type": "error", "message": "boom"})
  _drain_actor()

  db.expire_all()
  row = db.query(models.Chat).filter(models.Chat.id == chat.id).one()
  blocks = row.messages[-1]["blocks"]
  assert any(
    b.get("type") == "error" and b.get("message") == "boom" for b in blocks
  )


# --- Bug 4: SystemBroadcast end-to-end --------------------------------


@pytest.mark.asyncio
async def test_system_broadcast_delivers_to_subscriber():
  """A subscriber to SystemBroadcast receives events published
  while it's listening — independent of any per-chat broadcast."""
  sb = SystemBroadcast()
  q = sb.subscribe()
  try:
    sb.publish({"type": "app_updated", "appId": "36"})
    event = await asyncio.wait_for(q.get(), timeout=1.0)
    assert event == {"type": "app_updated", "appId": "36"}
  finally:
    sb.unsubscribe(q)


@pytest.mark.asyncio
async def test_notify_endpoint_reaches_system_broadcast(client, auth):
  """POST /api/notify must publish to the SystemBroadcast — that is
  the channel Shell.jsx subscribes to so app_updated reaches the
  iframe-version-bumper even when no chat is active.

  Regression for the "agent updates app → spinner stuck forever"
  bug: previously the publish loop only hit per-chat broadcasts,
  which close 30s after each turn, so any later notify had nowhere
  to land.
  """
  # Subscribe to the system broadcast BEFORE the POST so we can
  # observe the event landing in real time.
  sb = get_system_broadcast()
  q = sb.subscribe()
  try:
    r = client.post(
      "/api/notify",
      headers=auth,
      json={"type": "app_updated", "appId": "36"},
    )
    assert r.status_code == 204, r.text

    event = await asyncio.wait_for(q.get(), timeout=1.0)
    assert event["type"] == "app_updated"
    assert event["appId"] == "36"
  finally:
    sb.unsubscribe(q)


def test_notify_rejects_cross_site_request(client, auth):
  cross = client.post(
    "/api/notify",
    headers={**auth, "Sec-Fetch-Site": "cross-site"},
    json={"type": "app_updated", "appId": "36"},
  )
  assert cross.status_code == 403


def test_notify_body_type_validator_rejects_unknown():
  """NotifyBody rejects unknown system-event types."""
  try:
    NotifyBody(type="bogus")
  except ValidationError:
    pass
  else:
    raise AssertionError("Expected ValidationError for bogus event type")


# --- Chat-scoped app_built CTA (feature 094) --------------------------


@pytest.mark.asyncio
async def test_notify_app_updated_fires_app_built_on_owning_chat(
  client, auth, db
):
  """POST /api/notify {app_updated, appId} ALSO publishes a chat-scoped
  `app_built` onto the broadcast of the chat that built the app — and ONLY
  that chat's broadcast. This is what scopes the "Open app" CTA to the
  building chat instead of leaking it globally via app_updated."""
  app = models.App(
    name="builttoy", description="t", jsx_source="export default () => null",
    chat_id="builder-chat",
  )
  db.add(app)
  db.commit()
  db.refresh(app)

  building = bc_mod.create_broadcast("builder-chat")
  other = bc_mod.create_broadcast("bystander-chat")
  q_build = building.subscribe()[1]
  try:
    r = client.post(
      "/api/notify",
      headers=auth,
      json={"type": "app_updated", "appId": str(app.id)},
    )
    assert r.status_code == 204, r.text

    # The building chat sees BOTH the (catch-up) app_updated and the
    # chat-scoped app_built. Drain until we observe app_built.
    seen = []
    for _ in range(4):
      ev = await asyncio.wait_for(q_build.get(), timeout=1.0)
      seen.append(ev["type"])
      if ev["type"] == "app_built":
        assert ev["appId"] == str(app.id)
        break
    assert "app_built" in seen, seen
    # The bystander chat must NOT have received app_built.
    assert all(
      e.get("type") != "app_built" for e in other.event_log
    ), other.event_log
  finally:
    bc_mod.remove_broadcast("builder-chat")
    bc_mod.remove_broadcast("bystander-chat")


@pytest.mark.asyncio
async def test_notify_app_updated_no_app_built_without_owning_chat(
  client, auth, db
):
  """An app with no chat_id (App Store install, manual create) produces NO
  app_built — there is no building chat to scope a CTA to. The global
  app_updated still fires on the SystemBroadcast as before."""
  app = models.App(
    name="orphantoy", description="t", jsx_source="export default () => null",
    chat_id=None,
  )
  db.add(app)
  db.commit()
  db.refresh(app)

  sb = get_system_broadcast()
  sq = sb.subscribe()
  try:
    r = client.post(
      "/api/notify",
      headers=auth,
      json={"type": "app_updated", "appId": str(app.id)},
    )
    assert r.status_code == 204, r.text
    ev = await asyncio.wait_for(sq.get(), timeout=1.0)
    assert ev["type"] == "app_updated"
    assert ev["appId"] == str(app.id)
  finally:
    sb.unsubscribe(sq)


@pytest.mark.asyncio
async def test_notify_app_built_only_when_owning_chat_streaming(
  client, auth, db
):
  """If the building chat's turn already ended (no live broadcast), no
  app_built is emitted — the CTA only makes sense while the chat is the
  active streaming turn."""
  app = models.App(
    name="finishedtoy", description="t",
    jsx_source="export default () => null",
    chat_id="finished-chat",
  )
  db.add(app)
  db.commit()
  db.refresh(app)

  # No broadcast registered for "finished-chat" → the turn ended.
  sb = get_system_broadcast()
  sq = sb.subscribe()
  try:
    r = client.post(
      "/api/notify",
      headers=auth,
      json={"type": "app_updated", "appId": str(app.id)},
    )
    assert r.status_code == 204, r.text
    # Only the global app_updated lands; nothing raises trying to publish
    # app_built to a missing broadcast.
    ev = await asyncio.wait_for(sq.get(), timeout=1.0)
    assert ev["type"] == "app_updated"
  finally:
    sb.unsubscribe(sq)


# --- clear_active_broadcast_if: identity-keyed compare-and-clear -------


def test_clear_active_broadcast_if_clears_when_pointer_matches():
  """When the active-broadcast pointer still points at `bc`, the
  identity-keyed clear releases it and returns True — the no-successor case
  where the dying run must release the pointer rather than leak it."""
  bc = ChatBroadcast("c-match")
  set_active_broadcast(bc)
  try:
    assert clear_active_broadcast_if(bc) is True
    assert get_active_broadcast() is None, "the matching pointer must be cleared"
  finally:
    set_active_broadcast(None)


def test_clear_active_broadcast_if_leaves_when_different_active():
  """When a DIFFERENT broadcast is active (a fresh owner replaced the pointer),
  the clear must NOT touch it: it returns False and leaves the live owner's
  pointer intact — never clobbering a turn that isn't `bc`."""
  fresh = ChatBroadcast("c-fresh")
  superseded = ChatBroadcast("c-superseded")
  set_active_broadcast(fresh)
  try:
    assert clear_active_broadcast_if(superseded) is False
    assert get_active_broadcast() is fresh, (
      "a non-matching clear must leave the fresh owner's pointer untouched"
    )
  finally:
    set_active_broadcast(None)


def test_clear_active_broadcast_if_returns_false_when_already_none():
  """When the pointer is already None, the clear is a no-op: returns False and
  leaves it None."""
  set_active_broadcast(None)
  bc = ChatBroadcast("c-none")
  assert clear_active_broadcast_if(bc) is False
  assert get_active_broadcast() is None


# --- SSE subscription-leak: subscribe pairs with unsubscribe inside ----
# --- the GET /stream generator's lifecycle -----------------------------


class _NeverDisconnectedRequest:
  """Minimal stand-in for the Starlette Request the stream handler reads.

  The generator only calls `request.is_disconnected()`; a completed
  broadcast returns before that loop, so this is never invoked there,
  but the abandon-early case may, so it answers honestly (connected)."""

  async def is_disconnected(self):
    return False


@pytest.mark.asyncio
async def test_stream_subscribe_happens_inside_generator_not_at_endpoint(
  db, chat,
):
  """The leak fix: bc.subscribe() must run INSIDE the GET /stream
  generator (when iteration starts), NOT when the endpoint builds the
  StreamingResponse. Otherwise a client that disconnects before the
  generator's body ever runs leaves a queue in bc.subscribers whose
  pairing `finally: unsubscribe` never fires — a leaked subscriber that
  lingers until the broadcast completes."""
  from app.routes.chats_stream import stream_chat

  bc = ChatBroadcast(chat.id)
  bc.publish({"type": "text", "content": "hello"})
  bc_mod._broadcasts[bc.chat_id] = bc
  try:
    baseline = len(bc.subscribers)

    resp = await stream_chat(
      request=_NeverDisconnectedRequest(), chat_id=bc.chat_id,
      principal=Principal(owner=db.query(models.Owner).one(), app_id=None),
      db=db,
    )

    # Building the response must NOT have subscribed yet — the generator
    # body has not run. A subscriber here is exactly the leak.
    assert len(bc.subscribers) == baseline, (
      "subscribe ran at the endpoint, before the generator's try/finally — "
      "a pre-iteration client disconnect would leak this queue"
    )

    # Abandon the stream WITHOUT iterating (the client disconnected before
    # the first iteration), then close the generator. No subscriber must
    # have leaked.
    await resp.body_iterator.aclose()
    assert len(bc.subscribers) == baseline, (
      "abandoning the stream before iterating leaked a subscriber"
    )
  finally:
    bc_mod._broadcasts.pop(bc.chat_id, None)


@pytest.mark.asyncio
async def test_stream_generator_unsubscribes_after_finally_runs(db, chat):
  """Driving the GET /stream generator to completion must leave
  bc.subscribers back at baseline: subscribe (inside the generator) and
  the unsubscribe in its finally are paired across the generator's whole
  lifecycle, so a finished/abandoned stream never leaks a subscriber."""
  from app.routes.chats_stream import stream_chat

  bc = ChatBroadcast(chat.id)
  bc.publish({"type": "text", "content": "hi"})
  # A completed broadcast WITHOUT a done event in the catch-up: the
  # generator synthesises a done and returns, exercising the early-return
  # path that still must run its finally.
  bc.running = False
  bc_mod._broadcasts[bc.chat_id] = bc
  try:
    baseline = len(bc.subscribers)

    resp = await stream_chat(
      request=_NeverDisconnectedRequest(), chat_id=bc.chat_id,
      principal=Principal(owner=db.query(models.Owner).one(), app_id=None),
      db=db,
    )

    saw_subscriber_mid_stream = False
    async for _chunk in resp.body_iterator:
      # While the generator is live (mid-iteration), the subscriber IS
      # registered — proof subscribe happened inside generate().
      if len(bc.subscribers) > baseline:
        saw_subscriber_mid_stream = True

    assert saw_subscriber_mid_stream, (
      "subscribe never ran inside the generator — the catch-up burst was "
      "served without ever registering a live subscriber"
    )
    # The finally ran on generator exhaustion → back to baseline.
    assert len(bc.subscribers) == baseline, (
      "the generator finished but its finally never unsubscribed — leak"
    )
  finally:
    bc_mod._broadcasts.pop(bc.chat_id, None)
