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

  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
  ]

  @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: binary() | nil,
          up_to_date?: boolean(),
          tag_to_keys: %{optional(term()) => MapSet.t()},
          key_data: %{optional(term()) => %{tags: MapSet.t(), msg: term()}},
          stale_cache_buster: String.t() | nil,
          stale_cache_retry_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, %{})
    }
  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: %{}
    }
  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
    }
  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
    }
  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
end