electric_client

defmodule Electric.Client.ShapeState do
  @moduledoc """
  State for polling a shape.

  This struct holds the state needed between polling requests, including:
  - The shape handle and offset for resuming
  - Schema and value mapper for parsing responses
  - Tag tracking data for generating synthetic deletes from move-out events

  ## Usage

      # Create initial state
      state = ShapeState.new()

      # Poll for changes
      {:ok, messages, new_state} = Client.poll(client, shape, state)

      # State can also be created from a ResumeMessage (interop with stream API)
      state = ShapeState.from_resume(resume_message)
  """

  alias Electric.Client
  alias Electric.Client.Offset
  alias Electric.Client.Message.ResumeMessage
  alias Electric.Client.Util

  require Logger

  # Fast-loop detection: if we see @fast_loop_threshold requests at the same
  # offset within @fast_loop_window_ms, that counts as one detection.
  # After @fast_loop_max_count consecutive detections we raise an error.
  @fast_loop_window_ms 500
  @fast_loop_threshold 5
  @fast_loop_max_count 5
  @fast_loop_backoff_base_ms 100
  @fast_loop_backoff_max_ms 5_000

  defstruct [
    :shape_handle,
    :schema,
    :value_mapper_fun,
    :next_cursor,
    :stale_cache_buster,
    offset: Offset.before_all(),
    up_to_date?: false,
    tag_to_keys: %{},
    key_data: %{},
    stale_cache_retry_count: 0,
    self_heal_attempted?: false,
    disjunct_positions: nil,
    recent_requests: [],
    fast_loop_consecutive_count: 0
  ]

  @type t :: %__MODULE__{
          shape_handle: Client.shape_handle() | nil,
          offset: Offset.t(),
          schema: Client.schema() | nil,
          value_mapper_fun: Client.ValueMapper.mapper_fun() | nil,
          next_cursor: Client.cursor() | nil,
          up_to_date?: boolean(),
          tag_to_keys: %{optional(term()) => MapSet.t()},
          key_data: %{optional(term()) => %{tags: MapSet.t(), msg: term()}},
          disjunct_positions: [[non_neg_integer()]] | nil,
          stale_cache_buster: String.t() | nil,
          stale_cache_retry_count: non_neg_integer(),
          self_heal_attempted?: boolean(),
          recent_requests: [{integer(), Offset.t()}],
          fast_loop_consecutive_count: non_neg_integer()
        }

  @doc """
  Create a new initial polling state.

  ## Options

    * `:shape_handle` - Optional shape handle to resume from
    * `:offset` - Optional offset to resume from (default: before_all)
    * `:schema` - Optional schema for value mapping

  """
  @spec new(keyword()) :: t()
  def new(opts \\ []) do
    struct(__MODULE__, opts)
  end

  @doc """
  Create polling state from a ResumeMessage.

  This allows interop between the streaming and polling APIs - you can
  use `live: false` to get a ResumeMessage from a stream, then continue
  polling from that point.
  """
  @spec from_resume(ResumeMessage.t()) :: t()
  def from_resume(%ResumeMessage{} = resume) do
    %__MODULE__{
      shape_handle: resume.shape_handle,
      offset: resume.offset,
      schema: resume.schema,
      up_to_date?: true,
      tag_to_keys: Map.get(resume, :tag_to_keys, %{}),
      key_data: Map.get(resume, :key_data, %{}),
      disjunct_positions: Map.get(resume, :disjunct_positions)
    }
  end

  @doc """
  Reset polling state for a new shape handle, preserving schema and value mapper.

  Used when a 409 (must-refetch) response is received — the shape handle changes
  but the schema remains the same.
  """
  @spec reset(t(), Client.shape_handle()) :: t()
  def reset(%__MODULE__{} = state, shape_handle) do
    %{
      state
      | offset: Offset.before_all(),
        shape_handle: shape_handle,
        up_to_date?: false,
        next_cursor: nil,
        tag_to_keys: %{},
        key_data: %{},
        recent_requests: [],
        fast_loop_consecutive_count: 0,
        disjunct_positions: nil
    }
  end

  @doc """
  Convert polling state to a ResumeMessage for use with the streaming API.
  """
  @spec to_resume(t()) :: ResumeMessage.t()
  def to_resume(%__MODULE__{} = state) do
    %ResumeMessage{
      shape_handle: state.shape_handle,
      offset: state.offset,
      schema: state.schema,
      tag_to_keys: state.tag_to_keys,
      key_data: state.key_data,
      disjunct_positions: state.disjunct_positions
    }
  end

  @doc """
  Enter stale retry mode by setting a cache buster and incrementing the retry count.

  Called when a stale CDN response is detected - the server returns an expired
  handle that matches our cached expired handle.
  """
  @spec enter_stale_retry(t()) :: t()
  def enter_stale_retry(%__MODULE__{} = state) do
    %{
      state
      | stale_cache_buster: generate_cache_buster(),
        stale_cache_retry_count: state.stale_cache_retry_count + 1
    }
  end

  @doc """
  Clear stale retry state after a successful response.

  Called when we receive a fresh (non-stale) response from the server.
  """
  @spec clear_stale_retry(t()) :: t()
  def clear_stale_retry(%__MODULE__{} = state) do
    %{
      state
      | stale_cache_buster: nil,
        stale_cache_retry_count: 0,
        self_heal_attempted?: false
    }
  end

  @doc """
  Generate a random cache buster string.

  Uses 8 random bytes encoded as hex (16 characters).
  """
  @spec generate_cache_buster() :: String.t()
  def generate_cache_buster do
    Util.generate_id(8)
  end

  @doc """
  Check for fast-loop condition on non-live requests.

  Tracks recent requests in a sliding window. If too many requests occur at
  the same offset within the window, the client is stuck in a retry loop.

  Returns:
    * `{:ok, state}` — no fast loop detected
    * `{:backoff, ms, state}` — fast loop detected, caller should sleep `ms`
    * `{:error, message}` — fast loop persisted beyond max retries
  """
  @spec check_fast_loop(t()) ::
          {:ok, t()} | {:backoff, non_neg_integer(), t()} | {:error, String.t()}
  def check_fast_loop(%__MODULE__{} = state) do
    now = System.monotonic_time(:millisecond)
    current_offset = state.offset

    # Prune entries outside the window
    recent =
      Enum.filter(state.recent_requests, fn {ts, _offset} ->
        now - ts < @fast_loop_window_ms
      end)

    recent = recent ++ [{now, current_offset}]

    same_offset_count =
      Enum.count(recent, fn {_ts, offset} -> offset == current_offset end)

    if same_offset_count < @fast_loop_threshold do
      {:ok, %{state | recent_requests: recent}}
    else
      consecutive = state.fast_loop_consecutive_count + 1

      if consecutive >= @fast_loop_max_count do
        {:error,
         "Client is stuck in a fast retry loop " <>
           "(#{@fast_loop_threshold} requests in #{@fast_loop_window_ms}ms at the same offset, " <>
           "repeated #{@fast_loop_max_count} times). " <>
           "Client-side caches were cleared automatically on first detection, but the loop persists. " <>
           "This usually indicates a proxy or CDN misconfiguration. " <>
           "Common causes:\n" <>
           "  - Proxy is not including query parameters (handle, offset) in its cache key\n" <>
           "  - CDN is serving stale 409 responses\n" <>
           "  - Proxy is stripping required Electric headers from responses\n" <>
           "For more information visit the troubleshooting guide: https://electric-sql.com/docs/guides/troubleshooting"}
      else
        # Clear the window so the next detection requires a fresh burst of rapid
        # requests. Without this, stale entries from before the backoff would
        # immediately re-trigger on the very next request, even if the underlying
        # issue (e.g. CDN cache) resolved during the backoff.
        state = %{state | recent_requests: [], fast_loop_consecutive_count: consecutive}

        if consecutive == 1 do
          Logger.warning(
            "[Electric] Detected fast retry loop " <>
              "(#{@fast_loop_threshold} requests in #{@fast_loop_window_ms}ms at the same offset). " <>
              "Clearing client-side caches and resetting stream to recover. " <>
              "If this persists, check that your proxy includes all query parameters " <>
              "(especially 'handle' and 'offset') in its cache key, " <>
              "and that required Electric headers are forwarded to the client. " <>
              "For more information visit the troubleshooting guide: https://electric-sql.com/docs/guides/troubleshooting"
          )

          # First detection: clear shape handle to force a fresh request that
          # bypasses the CDN cache slot, and reset offset to start from scratch.
          state = %{state | shape_handle: nil, offset: Offset.before_all()}
          {:backoff, 0, state}
        else
          # Subsequent detections: exponential backoff with jitter
          max_delay =
            min(
              @fast_loop_backoff_max_ms,
              @fast_loop_backoff_base_ms * Integer.pow(2, consecutive)
            )

          delay = :rand.uniform(max(max_delay, 1))
          {:backoff, delay, state}
        end
      end
    end
  end

  @doc """
  Clear fast-loop tracking state.

  Called when the client transitions to live mode (up-to-date), since
  rapid polling is expected behaviour in live mode.
  """
  @spec clear_fast_loop(t()) :: t()
  def clear_fast_loop(%__MODULE__{} = state) do
    %{state | recent_requests: [], fast_loop_consecutive_count: 0}
  end
end