xmtp_elixir_sdk

defmodule Xmtp.Sync do
  @moduledoc """
  Phoenix/Postgres-friendly sync helpers for room mirrors.
  """

  alias Xmtp.Log
  alias Xmtp.Room
  alias XmtpElixirSdk.Conversation
  alias XmtpElixirSdk.DecodedMessage
  alias XmtpElixirSdk.Messages
  alias XmtpElixirSdk.Types

  @spec backfill_room(Conversation.t(), keyword()) ::
          {:ok, [DecodedMessage.t()]} | {:error, term()}
  def backfill_room(%Conversation{} = conversation, opts \\ []) do
    since_ns = Keyword.get(opts, :since_ns, 0)
    limit = Keyword.get(opts, :limit, 100)

    Messages.list(conversation, %Types.ListMessagesOptions{
      sent_after_ns: since_ns,
      limit: limit,
      direction: :ascending
    })
  end

  @spec apply_stream_event(map(), keyword()) :: {:ok, term()} | {:error, term()}
  def apply_stream_event(%{message: %Types.Message{} = message}, opts) do
    append_message(message, opts)
  end

  def apply_stream_event(%Types.Message{} = message, opts), do: append_message(message, opts)

  def apply_stream_event(_event, _opts), do: {:error, :unsupported_stream_event}

  @spec idempotency_key(Types.Message.t() | DecodedMessage.t() | map()) :: String.t()
  def idempotency_key(%Types.Message{id: id, conversation_id: conversation_id}) do
    hash_key(conversation_id, id)
  end

  def idempotency_key(%DecodedMessage{id: id, conversation_id: conversation_id}) do
    hash_key(conversation_id, id)
  end

  def idempotency_key(%{id: id, conversation_id: conversation_id})
      when is_binary(id) and is_binary(conversation_id) do
    hash_key(conversation_id, id)
  end

  @spec message_order_key(Types.Message.t() | DecodedMessage.t() | map()) ::
          {non_neg_integer(), String.t()}
  def message_order_key(%Types.Message{id: id, sent_at_ns: sent_at_ns}), do: {sent_at_ns || 0, id}

  def message_order_key(%DecodedMessage{id: id, sent_at_ns: sent_at_ns}),
    do: {sent_at_ns || 0, id}

  def message_order_key(%{id: id, sent_at_ns: sent_at_ns}) when is_binary(id),
    do: {sent_at_ns || 0, id}

  @spec reconcile_membership(Room.t(), [String.t()]) :: %{
          required(:room_key) => String.t(),
          required(:joined_inbox_ids) => [String.t()]
        }
  def reconcile_membership(%Room{} = room, joined_inbox_ids) when is_list(joined_inbox_ids) do
    %{room_key: room.room_key, joined_inbox_ids: Enum.uniq(joined_inbox_ids)}
  end

  defp append_message(message, opts) do
    repo = Keyword.fetch!(opts, :repo)
    room = Keyword.fetch!(opts, :room)
    sender = Keyword.get(opts, :sender, %{})

    Log.append_message(repo, room, message, sender)
  end

  defp hash_key(conversation_id, message_id) do
    Base.encode16(:crypto.hash(:sha256, "#{conversation_id}:#{message_id}"), case: :lower)
  end
end