Step 2 — Define proactive triggers

"""Proactive guidance state and trigger detection for the Inkeep onboarding bridge.

ProactiveState is kept separate from the Autoplay SDK pipeline (AsyncContextStore)
so the two state models don't interfere: AsyncContextStore owns the rolling event
window for chat grounding; ProactiveState owns the explicit demo-action counts
and SSE fan-out for trigger delivery.
"""

from __future__ import annotations

import asyncio
import time
from collections import defaultdict, deque
from dataclasses import asdict, dataclass
from typing import Any
from uuid import uuid4

TRIGGER_WINDOW_SECONDS = 120
DATA_SOURCE_OPEN_THRESHOLD = 3


@dataclass
class GuidanceEvent:
    type: str          # "guidance_offer" or "guidance_start"
    session_id: str
    message: str
    target: str
    tour_id: str
    reason: str
    created_at: float


@dataclass
class ProactiveOffer:
    id: str
    session_id: str
    user_id: str | None
    email: str | None
    message: str
    target: str
    tour_id: str
    reason: str
    created_at: float


class ProactiveState:
    """Lightweight in-memory store for the proactive guidance layer."""

    def __init__(self) -> None:
        self.recent_actions: dict[str, deque[dict[str, Any]]] = defaultdict(
            lambda: deque(maxlen=30)
        )
        self.pending_offer_by_session: dict[str, ProactiveOffer] = {}
        self.guidance_subscribers: dict[str, set[asyncio.Queue[GuidanceEvent]]] = defaultdict(set)

    def store_actions(self, session_id: str, actions: list[dict[str, Any]]) -> None:
        bucket = self.recent_actions[session_id]
        now = time.time()
        for action in actions:
            item = dict(action)
            item.setdefault("timestamp_start", now)
            bucket.append(item)

    async def publish_guidance(self, event: GuidanceEvent) -> None:
        for queue in list(self.guidance_subscribers[event.session_id]):
            await queue.put(event)


def _is_data_source_open(action: dict[str, Any]) -> bool:
    return action.get("type", "") == "data_source_open"


def _detect_stuck_onboarding(
    *, session_id: str, user_id: str | None, email: str | None, state: ProactiveState
) -> ProactiveOffer | None:
    # Guard: fire at most once per session
    if session_id in state.pending_offer_by_session:
        return None

    cutoff = time.time() - TRIGGER_WINDOW_SECONDS
    recent = [
        a for a in state.recent_actions[session_id]
        if float(a.get("timestamp_start") or 0) >= cutoff
    ]

    # If the user already completed the step in this window, don't nudge them
    completed = any(
        a.get("type") == "data_source_complete" for a in recent
    )
    if completed:
        return None

    open_clicks = [a for a in recent if _is_data_source_open(a)]
    if len(open_clicks) < DATA_SOURCE_OPEN_THRESHOLD:
        return None

    return ProactiveOffer(
        id=f"offer_{uuid4().hex}",
        session_id=session_id,
        user_id=user_id,
        email=email,
        message=(
            "Having trouble connecting your data source? "
            "The most common issue is IP whitelisting — "
            "want me to walk you through it?"
        ),
        target="[data-guide='connect-data-source']",
        tour_id="data-source-connection",
        reason=f"{len(open_clicks)} data_source_open actions in {TRIGGER_WINDOW_SECONDS}s without completion",
        created_at=time.time(),
    )

@app.get("/guidance/stream/{session_id}")
async def guidance_stream(session_id: str, request: Request) -> StreamingResponse:
    queue: asyncio.Queue[GuidanceEvent] = asyncio.Queue()
    proactive_state.guidance_subscribers[session_id].add(queue)

    async def events():
        try:
            # Send a ready event immediately so the client knows it's connected
            yield "event: ready\ndata: {}\n\n"
            while not await request.is_disconnected():
                try:
                    event = await asyncio.wait_for(queue.get(), timeout=15)
                    yield f"event: {event.type}\ndata: {json.dumps(asdict(event))}\n\n"
                except asyncio.TimeoutError:
                    yield "event: heartbeat\ndata: {}\n\n"
        finally:
            proactive_state.guidance_subscribers[session_id].discard(queue)

    return StreamingResponse(events(), media_type="text/event-stream")


@app.get("/context/{session_id}")
async def get_context(session_id: str) -> dict[str, Any]:
    """Return assembled context for injection into InkeepEmbeddedChat.

    Merges two sources:
    1. SDK context from AsyncContextStore (PostHog autocapture events via Step 1).
    2. Demo-specific stuck-connection narrative (explicit data_source_open events).

    Replaces the Step 1 version of this endpoint with one that is also
    aware of the proactive guidance state.
    """
    # Source 1 — Autoplay SDK activity context (assembled rolling summary)
    product_id = _session_product.get(session_id)
    sdk_context = context_store.get(session_id, product_id=product_id)

    # Source 2 — explicit demo actions tracked by ProactiveState
    demo_actions = list(proactive_state.recent_actions.get(session_id, []))

    if not sdk_context and not demo_actions:
        return {"context": "", "has_activity": False, "session_id": session_id}

    lines = []
    for a in demo_actions[-10:]:  # last 10 explicit events
        atype = a.get("type", "")
        title = a.get("title", "")
        ts_raw = a.get("timestamp_start", 0)
        try:
            ts = float(ts_raw)
        except (TypeError, ValueError):
            ts = 0.0
        lines.append(f"- [{atype}] {title} (t={ts:.0f})")

    parts = []
    if sdk_context:
        parts.append(sdk_context)
    if lines:
        parts.append(
            "The user is working through the Nexus Cloud onboarding. "
            "Here is what they have been doing recently:\n"
            + "\n".join(lines)
            + "\n\nThey appear to be struggling with the Connect Data Source step. "
            "Start by acknowledging this and offer specific guidance."
        )

    context = "\n\n".join(parts)
    return {
        "context": context,
        "has_activity": bool(context.strip()),
        "session_id": session_id,
    }


@app.post("/inkeep/reply")
async def inkeep_reply(reply: InkeepReply) -> dict[str, Any]:
    offer = proactive_state.pending_offer_by_session.pop(reply.session_id, None)
    if offer is None:
        raise HTTPException(
            status_code=404, detail="No pending offer for this session"
        )
    if reply.offer_id and offer.id != reply.offer_id:
        raise HTTPException(status_code=409, detail="Offer ID does not match")

    accepted = GuidanceEvent(
        type="guidance_start",
        session_id=offer.session_id,
        message="Start the data source connection guide.",
        target=offer.target,
        tour_id=offer.tour_id,
        reason=offer.reason,
        created_at=time.time(),
    )
    await proactive_state.publish_guidance(accepted)
    return {"accepted": True, "guidance": asdict(accepted)}


@app.post("/demo/actions")
async def demo_actions(payload: DemoActionsPayload) -> dict[str, Any]:
    await handle_demo_actions(payload)
    offer = proactive_state.pending_offer_by_session.get(payload.session_id)
    return {
        "stored": len(payload.actions),
        "pending_offer": offer is not None,
        "offer": asdict(offer) if offer else None,
    }

const API_BASE = process.env.NEXT_PUBLIC_BRIDGE_URL ?? "http://localhost:8787";

export async function sendDataSourceOpen(sessionId: string): Promise<void> {
  await fetch(`${API_BASE}/demo/actions`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      session_id: sessionId,
      actions: [
        {
          type: "data_source_open",
          title: "Connect Data Source — opened",
          canonical_url: "https://app.example.com/onboarding",
        },
      ],
    }),
  });
}

export function guidanceStreamUrl(sessionId: string): string {
  return `${API_BASE}/guidance/stream/${encodeURIComponent(sessionId)}`;
}

export async function fetchContext(sessionId: string): Promise<string> {
  const response = await fetch(
    `${API_BASE}/context/${encodeURIComponent(sessionId)}`
  );
  if (!response.ok) return "";
  const data = await response.json();
  return (data as { context?: string }).context ?? "";
}

export async function acceptInkeepOffer(
  sessionId: string,
  offerId: string | undefined,
): Promise<void> {
  await fetch(`${API_BASE}/inkeep/reply`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ session_id: sessionId, offer_id: offerId, text: "yes" }),
  });
}

.proactivePopup {
  position: fixed;
  bottom: 84px;
  right: 24px;
  width: 320px;
  background: var(--panel);
  border: 1px solid var(--border);
  border-radius: 12px;
  box-shadow: 0 8px 32px rgba(0,0,0,.14);
  padding: 14px 16px;
  z-index: 300;
  animation: popIn 180ms ease;
}

@keyframes popIn {
  from { opacity: 0; transform: translateY(10px); }
  to   { opacity: 1; transform: translateY(0); }
}

.proactivePopup__x {
  position: absolute;
  top: 10px;
  right: 12px;
  background: none;
  border: none;
  cursor: pointer;
  font-size: 18px;
  line-height: 1;
  color: var(--muted);
}
.proactivePopup__x:hover { color: var(--ink); }

.proactivePopup__header {
  display: flex;
  align-items: center;
  gap: 7px;
  font-size: 11px;
  font-weight: 700;
  letter-spacing: 0.06em;
  color: var(--primary);
  text-transform: uppercase;
  margin-bottom: 8px;
}
.proactivePopup__icon {
  width: 22px;
  height: 22px;
  border-radius: 50%;
  background: var(--primary);
  color: #fff;
  display: flex;
  align-items: center;
  justify-content: center;
  flex-shrink: 0;
}
.proactivePopup__msg {
  font-size: 13.5px;
  line-height: 1.5;
  margin-bottom: 12px;
  padding-right: 20px;
}
.proactivePopup__actions {
  display: flex;
  flex-direction: column;
  gap: 8px;
}