tribes-plugin-aether/lib/aether/chat.ex

defmodule TribeOne.TribesPlugin.Aether.Chat do
  @moduledoc """
  Chat domain and provider facade for TribeOne.TribesPlugin.Aether.
  """

  use Ash.Domain,
    otp_app: :tribe_one_aether

  require Ash.Query

  alias TribeOne.TribesPlugin.Aether.Chat.{Channel, Message, Participant}

  @default_channel_slug "general"
  @default_channel_title "General Chat"
  @pubsub Tribes.PubSub

  resources do
    resource Channel do
      define(:create_channel, action: :create)
      define(:get_channel, action: :by_id, args: [:id])
      define(:get_channel_by_slug, action: :by_slug, args: [:slug])
      define(:list_channels, action: :read)
    end

    resource Message do
      define(:create_message, action: :create)
      define(:get_message, action: :by_id, args: [:id])
      define(:list_all_messages, action: :read)
    end

    resource Participant do
      define(:create_participant, action: :create)
      define(:get_participant, action: :by_channel_pubkey, args: [:channel_id, :pubkey])
      define(:list_participants, action: :read)
      define(:list_participants_by_pubkey, action: :by_pubkey, args: [:pubkey])
    end
  end

  def backend_module(:public_sync), do: TribeOne.TribesPlugin.Aether.Chat.Backends.PublicSync
  def backend_module(:nostr_nip17), do: TribeOne.TribesPlugin.Aether.Chat.Backends.NostrNip17

  def backend_module(:nostr_nip04),
    do: TribeOne.TribesPlugin.Aether.Chat.Backends.NostrNip04ReadOnly

  def backend_module(:marmot), do: TribeOne.TribesPlugin.Aether.Chat.Backends.Marmot
  def backend_module(_backend), do: TribeOne.TribesPlugin.Aether.Chat.Backends.PublicSync

  def backend_capabilities(backend) do
    module = backend_module(backend)
    module.capabilities()
  end

  def supported_conversation_kinds, do: [:group, :context_group, :dm, :legacy_dm, :marmot_group]

  def supported_backends, do: [:public_sync, :nostr_nip17, :nostr_nip04, :marmot]

  def chat_panel_component, do: TribeOne.TribesPlugin.AetherWeb.ChatPanelComponent

  def recipient_picker_component, do: TribeOne.TribesPlugin.AetherWeb.ChatRecipientPickerComponent

  def default_channel_slug, do: @default_channel_slug

  def standalone_path(%Channel{slug: slug}), do: standalone_path(slug)
  def standalone_path(slug) when is_binary(slug), do: "/aether/chat/" <> slug

  def embed_path(%Channel{slug: slug}), do: embed_path(slug)
  def embed_path(slug) when is_binary(slug), do: "/aether/chat/embed/" <> slug

  def marmot_path(%Channel{slug: slug}), do: marmot_path(slug)
  def marmot_path(slug) when is_binary(slug), do: "/aether/chat/marmot/" <> slug

  def ensure_context_channel(provider, type, id, attrs \\ %{}, opts \\ [])
      when is_binary(provider) and is_binary(type) and is_binary(id) and is_map(attrs) do
    attrs
    |> Map.merge(%{
      context_provider: provider,
      context_type: type,
      context_id: id
    })
    |> ensure_conversation(opts)
  end

  def ensure_direct_conversation(sender, recipient, attrs \\ %{}, opts \\ [])

  def ensure_direct_conversation(sender_pubkey, recipient_pubkey, attrs, opts)
      when is_binary(sender_pubkey) and is_binary(recipient_pubkey) and is_map(attrs) do
    participants = Enum.sort([sender_pubkey, recipient_pubkey])

    attrs =
      attrs
      |> Map.update(:metadata, %{"participants" => participants}, fn metadata ->
        Map.put(metadata || %{}, "participants", participants)
      end)
      |> Map.put_new(:slug, direct_slug(sender_pubkey, recipient_pubkey))
      |> Map.put_new(:title, "Direct Message")
      |> Map.put_new(:conversation_kind, :dm)
      |> Map.put_new(:backend, :nostr_nip17)

    with {:ok, %Channel{} = channel} <- ensure_conversation(attrs, opts),
         {:ok, _sender} <- ensure_participant(channel, sender_pubkey, %{role: :member}, opts),
         {:ok, _recipient} <-
           ensure_participant(channel, recipient_pubkey, %{role: :member}, opts) do
      {:ok, channel}
    end
  end

  def direct_slug(pubkey_a, pubkey_b) when is_binary(pubkey_a) and is_binary(pubkey_b) do
    digest =
      [pubkey_a, pubkey_b]
      |> Enum.sort()
      |> Enum.join(":")
      |> then(&:crypto.hash(:sha256, &1))
      |> Base.encode16(case: :lower)

    "dm-" <> digest
  end

  def ash_opts(context \\ nil) do
    [
      authorize?: false,
      context: %{
        private: %{
          system?: true,
          system_purpose: :aether_chat,
          plugin_management_context: context
        }
      }
    ]
  end

  def ensure_conversation(attrs \\ %{}, opts \\ []) when is_map(attrs) do
    backend = attrs |> atomize_known_keys() |> Map.get(:backend, :public_sync)
    backend_module(backend).ensure_conversation(attrs, opts)
  end

  def ensure_channel(attrs \\ %{}, opts \\ []) when is_map(attrs) do
    attrs = normalize_channel_attrs(attrs)
    ash_opts = Keyword.get(opts, :ash_opts, ash_opts())

    case get_existing_channel(attrs.slug, ash_opts) do
      {:ok, nil} -> create_channel(attrs, ash_opts)
      {:ok, %Channel{} = channel} -> {:ok, channel}
      {:error, _reason} = error -> error
    end
  end

  defp get_existing_channel(slug, ash_opts) do
    Channel
    |> Ash.Query.filter(slug == ^slug)
    |> Ash.Query.limit(1)
    |> Ash.read_one(ash_opts)
  end

  defp get_existing_participant(channel_id, pubkey, ash_opts) do
    Participant
    |> Ash.Query.filter(channel_id == ^channel_id and pubkey == ^pubkey)
    |> Ash.Query.limit(1)
    |> Ash.read_one(ash_opts)
  end

  def ensure_participant(%Channel{} = channel, pubkey, attrs \\ %{}, opts \\ [])
      when is_binary(pubkey) and is_map(attrs) do
    attrs = atomize_known_keys(attrs)
    ash_opts = Keyword.get(opts, :ash_opts, ash_opts())

    case get_existing_participant(channel.id, pubkey, ash_opts) do
      {:ok, nil} ->
        attrs
        |> Map.merge(%{channel_id: channel.id, pubkey: pubkey})
        |> Map.put_new(:role, :member)
        |> Map.put_new(:metadata, %{})
        |> create_participant(ash_opts)

      {:ok, %Participant{} = participant} ->
        {:ok, participant}

      {:error, _reason} = error ->
        error
    end
  end

  def list_conversation_messages(%Channel{backend: backend} = channel, opts \\ []) do
    backend_module(backend).list_messages(channel, opts)
  end

  def list_messages(channel_or_id, opts \\ []) do
    channel_id = channel_id(channel_or_id)
    limit = Keyword.get(opts, :limit, 100)
    ash_opts = Keyword.get(opts, :ash_opts, ash_opts())

    result =
      Message
      |> Ash.Query.filter(channel_id == ^channel_id)
      |> Ash.Query.sort(inserted_at: :desc, id: :desc)
      |> Ash.Query.limit(limit)
      |> Ash.read(ash_opts)

    case result do
      {:ok, messages} -> {:ok, Enum.reverse(messages)}
      {:error, _reason} = error -> error
    end
  end

  def send_message(%Channel{backend: backend} = channel, attrs, opts \\ []) when is_map(attrs) do
    backend_module(backend).send_message(channel, attrs, opts)
  end

  def post_message(%Channel{} = channel, attrs, opts \\ []) when is_map(attrs) do
    ash_opts = Keyword.get(opts, :ash_opts, ash_opts())

    attrs
    |> normalize_message_attrs(channel)
    |> create_message(ash_opts)
    |> maybe_broadcast_message()
  end

  def subscribe_conversation(%Channel{backend: backend} = channel, pid \\ self(), opts \\ []) do
    backend_module(backend).subscribe(channel, pid, opts)
  end

  def subscribe_channel(%Channel{} = channel), do: subscribe_channel(channel.id)

  def subscribe_channel(channel_id) when is_binary(channel_id) do
    if pubsub_started?() do
      Phoenix.PubSub.subscribe(@pubsub, topic(channel_id))
    else
      {:error, :pubsub_not_started}
    end
  end

  def topic(channel_id) when is_binary(channel_id), do: "aether:chat:" <> channel_id

  defp maybe_broadcast_message({:ok, %Message{} = message}) do
    broadcast_message(message)
    {:ok, message}
  end

  defp maybe_broadcast_message({:error, _reason} = error), do: error

  defp broadcast_message(%Message{} = message) do
    if pubsub_started?() do
      Phoenix.PubSub.broadcast_from(
        @pubsub,
        self(),
        topic(message.channel_id),
        {:aether_chat, :message, message}
      )
    end
  end

  defp pubsub_started?, do: Process.whereis(@pubsub) != nil

  defp normalize_channel_attrs(attrs) do
    attrs = atomize_known_keys(attrs)
    title = attrs[:title] || @default_channel_title
    slug = attrs[:slug] || context_slug(attrs) || slugify(title) || @default_channel_slug

    %{
      slug: slug,
      title: title,
      description: attrs[:description],
      backend: attrs[:backend] || :public_sync,
      conversation_kind: attrs[:conversation_kind] || default_conversation_kind(attrs),
      context_provider: attrs[:context_provider],
      context_type: attrs[:context_type],
      context_id: attrs[:context_id],
      metadata: attrs[:metadata] || %{}
    }
  end

  defp normalize_message_attrs(attrs, %Channel{} = channel) do
    attrs = atomize_known_keys(attrs)

    %{
      channel_id: channel.id,
      author_id: attrs[:author_id],
      author_pubkey: attrs[:author_pubkey],
      body: String.trim(attrs[:body] || ""),
      client_message_id: attrs[:client_message_id],
      metadata: attrs[:metadata] || %{}
    }
  end

  defp atomize_known_keys(attrs) do
    Enum.reduce(attrs, %{}, fn
      {key, value}, acc when is_atom(key) -> Map.put(acc, key, value)
      {key, value}, acc when is_binary(key) -> put_known_key(acc, key, value)
    end)
  end

  defp put_known_key(acc, key, value) do
    case key do
      "slug" -> Map.put(acc, :slug, value)
      "title" -> Map.put(acc, :title, value)
      "description" -> Map.put(acc, :description, value)
      "backend" -> Map.put(acc, :backend, parse_atom(value))
      "conversation_kind" -> Map.put(acc, :conversation_kind, parse_atom(value))
      "context_provider" -> Map.put(acc, :context_provider, value)
      "context_type" -> Map.put(acc, :context_type, value)
      "context_id" -> Map.put(acc, :context_id, value)
      "metadata" -> Map.put(acc, :metadata, value)
      "pubkey" -> Map.put(acc, :pubkey, value)
      "user_id" -> Map.put(acc, :user_id, value)
      "display_name" -> Map.put(acc, :display_name, value)
      "role" -> Map.put(acc, :role, parse_atom(value))
      "author_id" -> Map.put(acc, :author_id, value)
      "author_pubkey" -> Map.put(acc, :author_pubkey, value)
      "body" -> Map.put(acc, :body, value)
      "client_message_id" -> Map.put(acc, :client_message_id, value)
      _other -> acc
    end
  end

  defp parse_atom(value) when is_atom(value), do: value
  defp parse_atom("public_sync"), do: :public_sync
  defp parse_atom("nostr_nip17"), do: :nostr_nip17
  defp parse_atom("nostr_nip04"), do: :nostr_nip04
  defp parse_atom("marmot"), do: :marmot
  defp parse_atom("group"), do: :group
  defp parse_atom("context_group"), do: :context_group
  defp parse_atom("dm"), do: :dm
  defp parse_atom("legacy_dm"), do: :legacy_dm
  defp parse_atom("marmot_group"), do: :marmot_group
  defp parse_atom("owner"), do: :owner
  defp parse_atom("member"), do: :member
  defp parse_atom(_value), do: nil

  defp context_slug(%{context_provider: provider, context_type: type, context_id: id})
       when is_binary(provider) and is_binary(type) and is_binary(id) do
    Enum.map_join([provider, type, id], "-", &slugify/1)
  end

  defp context_slug(_attrs), do: nil

  defp default_conversation_kind(%{context_provider: provider}) when is_binary(provider),
    do: :context_group

  defp default_conversation_kind(_attrs), do: :group

  defp channel_id(%Channel{id: id}), do: id
  defp channel_id(id) when is_binary(id), do: id

  defp slugify(value) when is_binary(value) do
    slug =
      value
      |> String.downcase()
      |> String.replace(~r/[^a-z0-9]+/u, "-")
      |> String.trim("-")

    if slug == "", do: nil, else: slug
  end
end