electric

defmodule Electric.Shapes do
  alias Electric.Replication.LogOffset
  alias Electric.ShapeCache.Storage
  alias Electric.ShapeCache
  alias Electric.ShapeCache.ShapeStatus
  alias Electric.Shapes.Api
  alias Electric.Shapes.Shape
  alias Electric.Telemetry.OpenTelemetry

  import Electric, only: [is_stack_id: 1, is_shape_handle: 1]

  @type shape_handle :: Electric.shape_handle()
  @type stack_id :: Electric.stack_id()

  @doc """
  Get the snapshot followed by the log.

  Accepts a `:snapshot_status` opt carrying the outcome of a prior
  `await_snapshot_start` call (typically performed by the admission
  plug, which holds an `:initial` permit across the wait so the snapshot
  pool is protected by admission-gate backpressure). When set, the
  redundant wait is skipped; this is required for correctness on the
  snapshot-failure path, where re-issuing the GenServer call would race
  against the consumer's `stop_and_clean`.
  """
  @spec get_merged_log_stream(stack_id(), shape_handle(),
          since: LogOffset.t(),
          up_to: LogOffset.t(),
          read_only?: boolean(),
          snapshot_status: Api.Request.snapshot_status()
        ) :: {:ok, Storage.log()} | {:error, term()}
  def get_merged_log_stream(stack_id, shape_handle, opts)
      when is_shape_handle(shape_handle) and is_stack_id(stack_id) do
    offset = Access.get(opts, :since, LogOffset.before_all())
    max_offset = Access.get(opts, :up_to, LogOffset.last())

    case Access.get(opts, :snapshot_status) do
      {:error, _} = error ->
        error

      :started ->
        {:ok, build_log_stream(stack_id, shape_handle, offset, max_offset, opts)}

      nil ->
        if ShapeCache.has_shape?(shape_handle, stack_id) do
          with :started <- ShapeCache.await_snapshot_start(shape_handle, stack_id) do
            {:ok, build_log_stream(stack_id, shape_handle, offset, max_offset, opts)}
          end
        else
          # If we have a shape handle, but no shape, it means the shape was deleted. Send a 409
          # and expect the client to retry - if the state of the world allows, it'll get a new handle.
          {:error, Electric.Shapes.Api.Error.must_refetch()}
        end
    end
  end

  defp build_log_stream(stack_id, shape_handle, offset, max_offset, opts) do
    storage =
      Storage.for_shape(
        shape_handle,
        Storage.for_stack(stack_id, read_only?: opts[:read_only?])
      )

    Storage.get_log_stream(offset, max_offset, storage)
  end

  @doc """
  Get the shape handle that corresponds to this shape definition and return it
  """
  @spec fetch_handle_by_shape(stack_id(), Shape.t()) :: {:ok, shape_handle()} | :error
  def fetch_handle_by_shape(stack_id, %Shape{} = shape_def) when is_stack_id(stack_id) do
    ShapeCache.fetch_handle_by_shape(shape_def, stack_id)
  end

  @spec fetch_shape_by_handle(stack_id(), shape_handle()) :: Shape.t() | :error
  def fetch_shape_by_handle(stack_id, shape_handle)
      when is_shape_handle(shape_handle) and is_stack_id(stack_id) do
    ShapeCache.fetch_shape_by_handle(shape_handle, stack_id)
  end

  @doc """
  Cheaply validate that a shape handle matches the shape definition.
  """
  def resolve_shape_handle(stack_id, shape_handle, %Shape{} = shape, opts \\ []) do
    ShapeCache.resolve_shape_handle(shape_handle, shape, stack_id, opts)
  end

  @doc """
  Get or create a shape handle and return it along with the latest offset of the shape
  """
  @spec get_or_create_shape_handle(stack_id(), Shape.t()) ::
          {shape_handle(), LogOffset.t()} | {:error, term()}
  def get_or_create_shape_handle(stack_id, shape_def) when is_stack_id(stack_id) do
    ShapeCache.get_or_create_shape_handle(
      shape_def,
      stack_id,
      otel_ctx: OpenTelemetry.get_current_context()
    )
  end

  @doc """
  Get the last exclusive offset of the chunk starting from the given offset

  If `nil` is returned, chunk is not complete and the shape's latest offset should be used
  """
  @spec get_chunk_end_log_offset(stack_id(), shape_handle(), LogOffset.t(), keyword()) ::
          LogOffset.t() | nil
  def get_chunk_end_log_offset(stack_id, shape_handle, offset, opts \\ []) do
    storage =
      Storage.for_shape(
        shape_handle,
        Storage.for_stack(stack_id, read_only?: opts[:read_only?])
      )

    Storage.get_chunk_end_log_offset(offset, storage)
  end

  @doc """
  Check whether the log has an entry for a given shape handle
  """
  @spec has_shape?(stack_id(), shape_handle()) :: boolean()
  def has_shape?(stack_id, shape_handle) do
    ShapeCache.has_shape?(shape_handle, stack_id)
  end

  @doc """
  Remove and clean up all data (meta data and shape log + snapshot) associated with
  the given shape handle
  """
  @spec clean_shape(stack_id(), shape_handle()) :: :ok
  def clean_shape(stack_id, shape_handle) do
    ShapeCache.clean_shape(shape_handle, stack_id)
    :ok
  end

  @spec clean_shapes(stack_id(), [shape_handle()]) :: :ok
  def clean_shapes(stack_id, shape_handles) do
    for shape_handle <- shape_handles do
      ShapeCache.clean_shape(shape_handle, stack_id)
    end

    :ok
  end

  @doc """
  Wrap the writing of a snapshot to some Storage backend with the required
  ShapeStatus update calls.
  """
  @spec make_new_snapshot!(
          Electric.Shapes.Querying.json_result_stream(),
          Storage.shape_storage(),
          stack_id(),
          shape_handle()
        ) :: :ok | {:error, term()}
  def make_new_snapshot!(stream, storage, stack_id, shape_handle) do
    with :ok <- Storage.make_new_snapshot!(stream, storage),
         :ok <- ShapeStatus.mark_snapshot_complete(stack_id, shape_handle) do
      :ok
    end
  end

  @spec mark_snapshot_started(Storage.shape_storage(), stack_id(), shape_handle()) ::
          :ok | {:error, term()}
  def mark_snapshot_started(storage, stack_id, shape_handle) do
    with :ok <- Storage.mark_snapshot_as_started(storage) do
      ShapeStatus.mark_snapshot_started(stack_id, shape_handle)
    end
  end

  defdelegate query_subset(handle, shape, subset, opts),
    to: Electric.Postgres.SnapshotQuery,
    as: :execute_for_subset
end