lightspeed

//// Transport profile matrix with progressive enhancement semantics for M24.

import gleam/int
import gleam/string
import lightspeed/agent/session
import lightspeed/protocol
import lightspeed/transport
import lightspeed/transport/contract
import lightspeed/transport/wisp_websocket

/// Supported transport profiles for compatibility matrix testing.
pub type Profile {
  WebSocketOnly
  WebSocketWithServerSentEventsFallback
  WebSocketWithLongPollingFallback
}

/// Active runtime mode.
pub type Mode {
  PrimaryRealtime
  FallbackPolling
}

/// Security defaults enforced by matrix adapters.
pub type SecurityPolicy {
  SecurityPolicy(
    require_https_origin: Bool,
    require_csrf: Bool,
    max_payload_bytes: Int,
  )
}

/// Timeout defaults for heartbeat and reconnect behavior.
pub type TimeoutPolicy {
  TimeoutPolicy(heartbeat_timeout_ms: Int, reconnect_grace_ms: Int)
}

/// Capability flags used by compatibility fixtures.
pub type Capability {
  OrderedClientEvents
  ReconnectSemantics
  ServerPushFrames
  BidirectionalChannel
  ProgressiveFallback
}

/// Connect/reconnect request shape.
pub type ConnectRequest {
  ConnectRequest(
    route: String,
    csrf_token: String,
    origin: String,
    now_ms: Int,
    prefer_fallback: Bool,
  )
}

/// Matrix adapter state.
pub type AdapterState {
  AdapterState(
    websocket: wisp_websocket.AdapterState,
    profile: Profile,
    mode: Mode,
    queued_outbound: List(String),
    last_client_sequence: Int,
    last_client_seen_at_ms: Int,
    security_policy: SecurityPolicy,
    timeout_policy: TimeoutPolicy,
  )
}

/// Connect/reconnect result.
pub type ConnectResult {
  Connected(state: AdapterState, outbound_frames: List(String))
  Rejected(error: contract.AdapterError)
}

/// Receive/poll result.
pub type ReceiveResult {
  ReceiveResult(state: AdapterState, outbound_frames: List(String))
}

/// Default transport security policy.
pub fn default_security_policy() -> SecurityPolicy {
  SecurityPolicy(
    require_https_origin: True,
    require_csrf: True,
    max_payload_bytes: 4096,
  )
}

/// Default timeout policy.
pub fn default_timeout_policy() -> TimeoutPolicy {
  TimeoutPolicy(heartbeat_timeout_ms: 30_000, reconnect_grace_ms: 120_000)
}

/// Connect with default protection, security, and timeout policies.
pub fn connect(
  session_state: session.Session,
  request: ConnectRequest,
  profile: Profile,
  auth_hook: contract.AuthHook,
) -> ConnectResult {
  connect_with_policies(
    session_state,
    request,
    profile,
    auth_hook,
    contract.allow_protection(),
    default_security_policy(),
    default_timeout_policy(),
  )
}

/// Connect with explicit protection, security, and timeout policies.
pub fn connect_with_policies(
  session_state: session.Session,
  request: ConnectRequest,
  profile: Profile,
  auth_hook: contract.AuthHook,
  protection_hook: contract.ProtectionHook,
  security_policy: SecurityPolicy,
  timeout_policy: TimeoutPolicy,
) -> ConnectResult {
  case ensure_connection_security(request, security_policy) {
    Error(error) -> Rejected(error: error)
    Ok(Nil) ->
      case
        wisp_websocket.connect_with_hooks(
          session_state,
          to_ws_request(request),
          auth_hook,
          protection_hook,
        )
      {
        wisp_websocket.Rejected(error) -> Rejected(error: error)
        wisp_websocket.Connected(ws_state, frames) -> {
          let mode = select_mode(profile, request.prefer_fallback)
          let state =
            AdapterState(
              websocket: ws_state,
              profile: profile,
              mode: mode,
              queued_outbound: case mode {
                PrimaryRealtime -> []
                FallbackPolling -> frames
              },
              last_client_sequence: 0,
              last_client_seen_at_ms: request.now_ms,
              security_policy: security_policy,
              timeout_policy: timeout_policy,
            )
          Connected(state: state, outbound_frames: case mode {
            PrimaryRealtime -> frames
            FallbackPolling -> []
          })
        }
      }
  }
}

/// Reconnect with the state's existing policy defaults.
pub fn reconnect(
  state: AdapterState,
  request: ConnectRequest,
  auth_hook: contract.AuthHook,
) -> ConnectResult {
  reconnect_with_policies(
    state,
    request,
    auth_hook,
    contract.allow_protection(),
    state.security_policy,
    state.timeout_policy,
  )
}

/// Reconnect with explicit protection, security, and timeout policies.
pub fn reconnect_with_policies(
  state: AdapterState,
  request: ConnectRequest,
  auth_hook: contract.AuthHook,
  protection_hook: contract.ProtectionHook,
  security_policy: SecurityPolicy,
  timeout_policy: TimeoutPolicy,
) -> ConnectResult {
  case ensure_connection_security(request, security_policy) {
    Error(error) -> Rejected(error: error)
    Ok(Nil) ->
      case
        request.now_ms - state.last_client_seen_at_ms
        > timeout_policy.reconnect_grace_ms
      {
        True ->
          Rejected(error: contract.InvalidAdapterState(
            "reconnect_grace_exceeded",
          ))
        False ->
          case
            wisp_websocket.reconnect_with_hooks(
              state.websocket,
              to_ws_request(request),
              auth_hook,
              protection_hook,
            )
          {
            wisp_websocket.Rejected(error) -> Rejected(error: error)
            wisp_websocket.Connected(ws_state, frames) -> {
              let next_mode =
                select_mode(state.profile, request.prefer_fallback)
              let merged_queue = case next_mode {
                PrimaryRealtime -> []
                FallbackPolling -> append(state.queued_outbound, frames)
              }
              let next =
                AdapterState(
                  websocket: ws_state,
                  profile: state.profile,
                  mode: next_mode,
                  queued_outbound: merged_queue,
                  last_client_sequence: state.last_client_sequence,
                  last_client_seen_at_ms: request.now_ms,
                  security_policy: security_policy,
                  timeout_policy: timeout_policy,
                )
              Connected(state: next, outbound_frames: case next_mode {
                PrimaryRealtime ->
                  case state.mode {
                    PrimaryRealtime -> frames
                    FallbackPolling -> append(state.queued_outbound, frames)
                  }
                FallbackPolling -> []
              })
            }
          }
      }
  }
}

/// Receive one client frame.
///
/// `client_sequence` must be monotonically increasing per connection.
pub fn receive(
  state: AdapterState,
  payload: String,
  now_ms: Int,
  client_sequence: Int,
) -> ReceiveResult {
  receive_with_hooks(
    state,
    payload,
    now_ms,
    client_sequence,
    contract.allow_rate_limit(),
  )
}

/// Receive one client frame with explicit rate-limit hook.
pub fn receive_with_hooks(
  state: AdapterState,
  payload: String,
  now_ms: Int,
  client_sequence: Int,
  rate_limit_hook: contract.RateLimitHook,
) -> ReceiveResult {
  case
    now_ms - state.last_client_seen_at_ms
    > state.timeout_policy.heartbeat_timeout_ms
  {
    True ->
      emit_frames(state, [
        protocol.encode(protocol.Failure(
          ref: "",
          reason: contract.error_to_string(contract.InvalidAdapterState(
            "heartbeat_timeout",
          )),
        )),
      ])

    False ->
      case string.length(payload) > state.security_policy.max_payload_bytes {
        True ->
          emit_frames(AdapterState(..state, last_client_seen_at_ms: now_ms), [
            protocol.encode(protocol.Failure(
              ref: "",
              reason: contract.error_to_string(contract.UnsupportedClientFrame(
                "payload_too_large",
              )),
            )),
          ])

        False ->
          case client_sequence <= state.last_client_sequence {
            True ->
              emit_frames(
                AdapterState(..state, last_client_seen_at_ms: now_ms),
                [
                  protocol.encode(protocol.Failure(
                    ref: "",
                    reason: contract.error_to_string(
                      contract.InvalidAdapterState(event_order_violation_reason(
                        state.last_client_sequence,
                        client_sequence,
                      )),
                    ),
                  )),
                ],
              )

            False -> {
              let received =
                wisp_websocket.receive_with_hooks(
                  state.websocket,
                  payload,
                  now_ms,
                  rate_limit_hook,
                )
              let next =
                AdapterState(
                  ..state,
                  websocket: received.state,
                  last_client_sequence: client_sequence,
                  last_client_seen_at_ms: now_ms,
                )
              emit_frames(next, received.outbound_frames)
            }
          }
      }
  }
}

/// Poll queued server frames.
pub fn poll(state: AdapterState) -> ReceiveResult {
  case state.mode {
    PrimaryRealtime -> ReceiveResult(state: state, outbound_frames: [])
    FallbackPolling ->
      ReceiveResult(
        state: AdapterState(..state, queued_outbound: []),
        outbound_frames: state.queued_outbound,
      )
  }
}

/// Session state accessor.
pub fn session_state(state: AdapterState) -> session.Session {
  wisp_websocket.session_state(state.websocket)
}

/// Active profile label.
pub fn profile_label(profile: Profile) -> String {
  case profile {
    WebSocketOnly -> "websocket_only"
    WebSocketWithServerSentEventsFallback -> "websocket_plus_sse"
    WebSocketWithLongPollingFallback -> "websocket_plus_long_polling"
  }
}

/// Active mode label.
pub fn mode_label(state: AdapterState) -> String {
  case state.mode {
    PrimaryRealtime -> "primary_realtime"
    FallbackPolling -> "fallback_polling"
  }
}

/// Whether fallback mode is active.
pub fn progressive_enhancement_active(state: AdapterState) -> Bool {
  case state.mode {
    PrimaryRealtime -> False
    FallbackPolling -> True
  }
}

/// Whether the profile has a fallback mode available.
pub fn fallback_available(profile: Profile) -> Bool {
  case profile {
    WebSocketOnly -> False
    WebSocketWithServerSentEventsFallback -> True
    WebSocketWithLongPollingFallback -> True
  }
}

/// Active transport for current profile+mode.
pub fn active_transport(state: AdapterState) -> transport.Transport {
  case state.mode, state.profile {
    PrimaryRealtime, _ -> transport.WebSocket
    FallbackPolling, WebSocketWithServerSentEventsFallback ->
      transport.ServerSentEvents
    FallbackPolling, WebSocketWithLongPollingFallback -> transport.LongPolling
    FallbackPolling, WebSocketOnly -> transport.WebSocket
  }
}

/// Active transport label.
pub fn active_transport_label(state: AdapterState) -> String {
  transport.label(active_transport(state))
}

/// Capability predicate for fixtures and docs.
pub fn capability_enabled(state: AdapterState, capability: Capability) -> Bool {
  case capability {
    OrderedClientEvents -> True
    ReconnectSemantics -> True
    ServerPushFrames ->
      case state.mode {
        PrimaryRealtime -> True
        FallbackPolling -> False
      }
    BidirectionalChannel ->
      case state.mode {
        PrimaryRealtime -> True
        FallbackPolling -> False
      }
    ProgressiveFallback -> progressive_enhancement_active(state)
  }
}

/// Number of queued frames pending `poll`.
pub fn queued_outbound_count(state: AdapterState) -> Int {
  list_length(state.queued_outbound)
}

fn select_mode(profile: Profile, prefer_fallback: Bool) -> Mode {
  case prefer_fallback && fallback_available(profile) {
    True -> FallbackPolling
    False -> PrimaryRealtime
  }
}

fn ensure_connection_security(
  request: ConnectRequest,
  policy: SecurityPolicy,
) -> Result(Nil, contract.AdapterError) {
  case policy.require_csrf && request.csrf_token == "" {
    True -> Error(contract.ProtectionRejected("csrf_token_missing"))
    False ->
      case policy.require_https_origin && is_insecure_origin(request.origin) {
        True -> Error(contract.ProtectionRejected("insecure_origin"))
        False -> Ok(Nil)
      }
  }
}

fn is_insecure_origin(origin: String) -> Bool {
  case
    string.starts_with(origin, "https://")
    || string.starts_with(origin, "http://localhost")
    || string.starts_with(origin, "http://127.0.0.1")
  {
    True -> False
    False -> True
  }
}

fn emit_frames(state: AdapterState, frames: List(String)) -> ReceiveResult {
  case state.mode {
    PrimaryRealtime -> ReceiveResult(state: state, outbound_frames: frames)
    FallbackPolling ->
      ReceiveResult(
        state: AdapterState(
          ..state,
          queued_outbound: append(state.queued_outbound, frames),
        ),
        outbound_frames: [],
      )
  }
}

fn to_ws_request(request: ConnectRequest) -> wisp_websocket.WebSocketRequest {
  wisp_websocket.WebSocketRequest(
    route: request.route,
    csrf_token: request.csrf_token,
    origin: request.origin,
    now_ms: request.now_ms,
  )
}

fn event_order_violation_reason(last: Int, incoming: Int) -> String {
  "event_order_violation:last="
  <> int.to_string(last)
  <> ":incoming="
  <> int.to_string(incoming)
}

fn append(left: List(String), right: List(String)) -> List(String) {
  case left {
    [] -> right
    [entry, ..rest] -> [entry, ..append(rest, right)]
  }
}

fn list_length(values: List(a)) -> Int {
  case values {
    [] -> 0
    [_, ..rest] -> 1 + list_length(rest)
  }
}