xmtp_elixir_sdk

defmodule Xmtp.RoomServer do
  @moduledoc false

  use GenServer

  import Ecto.Query, warn: false

  alias Xmtp.Log
  alias Xmtp.Manager
  alias Xmtp.MessageLog
  alias Xmtp.Principal
  alias Xmtp.Room
  alias Xmtp.RoomMembership
  alias Xmtp.RoomPanel
  alias Xmtp.Wallet
  alias XmtpElixirSdk.Client
  alias XmtpElixirSdk.Clients
  alias XmtpElixirSdk.Conversation
  alias XmtpElixirSdk.Conversations
  alias XmtpElixirSdk.Events
  alias XmtpElixirSdk.Groups
  alias XmtpElixirSdk.Internal.ConversationServer
  alias XmtpElixirSdk.Messages
  alias XmtpElixirSdk.Signer
  alias XmtpElixirSdk.Types

  @message_limit 24

  def start_link(opts) do
    definition = Keyword.fetch!(opts, :definition)
    registry = Keyword.fetch!(opts, :registry)
    GenServer.start_link(__MODULE__, opts, name: {:via, Registry, {registry, definition.key}})
  end

  def child_spec(opts) do
    definition = Keyword.fetch!(opts, :definition)

    %{
      id: {__MODULE__, definition.key},
      start: {__MODULE__, :start_link, [opts]}
    }
  end

  @impl true
  def init(opts) do
    state =
      %{
        manager: Keyword.fetch!(opts, :manager),
        repo: Keyword.fetch!(opts, :repo),
        pubsub: Keyword.fetch!(opts, :pubsub),
        runtime_name: Keyword.fetch!(opts, :runtime_name),
        definition: Keyword.fetch!(opts, :definition),
        definition_loader: Keyword.get(opts, :definition_loader),
        mode: :unavailable,
        unavailable_reason: :room_unavailable,
        relay_client: nil,
        public_room: nil,
        room: nil,
        clients_by_wallet: %{},
        pending_signatures: %{}
      }
      |> restore_state!()

    schedule_presence_tick(state)
    {:ok, state}
  end

  @impl true
  def handle_call({:public_room_panel, principal, _claims}, _from, state) do
    state = refresh_definition(state)
    {:reply, {:ok, build_panel(state, principal)}, state}
  end

  def handle_call({:request_join, principal, claims}, _from, state) do
    state = refresh_definition(state)

    with {:ok, ready_state} <- require_ready(state),
         {:ok, principal} <- require_principal(principal),
         {:ok, wallet_address} <- fetch_wallet_address(principal),
         :ok <- authorize_join(ready_state, principal, claims),
         false <- room_full?(ready_state, principal),
         {:ok, client, next_state} <-
           ensure_join_candidate(ready_state, principal, wallet_address) do
      cond do
        joined?(next_state, wallet_address) ->
          touched_state =
            touch_membership_presence(next_state, principal, wallet_address, client.inbox_id)

          {:reply, {:ok, build_panel(touched_state, principal)}, touched_state}

        client.ready? ->
          {:ok, panel, updated_state} = invite_joined_member(next_state, principal, client)
          {:reply, {:ok, panel}, updated_state}

        true ->
          {:ok, %{signature_request_id: request_id, signature_text: signature_text}} =
            Clients.unsafe_create_inbox_signature_text(client)

          updated_state =
            put_in(next_state.pending_signatures[request_id], %{
              wallet_address: wallet_address,
              action: :join,
              principal: principal
            })

          panel =
            build_panel(
              updated_state,
              principal,
              "Check your wallet to finish joining.",
              request_id
            )

          {:reply,
           {:needs_signature,
            %{
              request_id: request_id,
              signature_text: signature_text,
              wallet_address: wallet_address,
              panel: panel
            }}, updated_state}
      end
    else
      true -> {:reply, {:error, :room_full}, state}
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call(
        {:complete_join_signature, principal, request_id, signature, claims},
        _from,
        state
      ) do
    state = refresh_definition(state)

    with {:ok, ready_state} <- require_ready(state),
         {:ok, principal} <- require_principal(principal),
         {:ok, wallet_address} <- fetch_wallet_address(principal),
         :ok <- authorize_join(ready_state, principal, claims),
         :ok <- validate_pending_signature(ready_state, request_id, wallet_address, :join),
         {:ok, client} <- fetch_cached_client(ready_state, wallet_address),
         identifier = wallet_identifier(wallet_address),
         {:ok, signer} <- Signer.eoa(identifier, signature),
         :ok <- Clients.unsafe_apply_signature_request(client, request_id, signer),
         {:ok, registered_client} <- Clients.register(client),
         false <- room_full?(ready_state, principal) do
      next_state =
        ready_state
        |> put_in([:clients_by_wallet, wallet_address], registered_client)
        |> update_in([:pending_signatures], &Map.delete(&1, request_id))

      {:ok, panel, updated_state} =
        invite_joined_member(next_state, principal, registered_client)

      {:reply, {:ok, panel}, updated_state}
    else
      true -> {:reply, {:error, :room_full}, state}
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:send_public_message, principal, body}, _from, state) do
    state = refresh_definition(state)
    body = normalize_body(body)

    with {:ok, ready_state} <- require_ready(state),
         {:ok, principal} <- require_principal(principal),
         {:ok, wallet_address} <- fetch_wallet_address(principal),
         :ok <- validate_body(body),
         :ok <- require_joined(ready_state, principal, wallet_address),
         {:ok, client, next_state} <-
           ensure_registered_client(ready_state, principal, wallet_address),
         {:ok, room} <- Conversations.get_by_id(client, ready_state.public_room.id),
         {:ok, message_id} <- Messages.send_text(room, body),
         {:ok, message} <- fetch_message(client, message_id) do
      updated_state =
        next_state
        |> persist_streamed_message(message)
        |> touch_membership_presence(principal, wallet_address, client.inbox_id)

      {:reply, {:ok, build_panel(updated_state, principal)}, updated_state}
    else
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:invite_user, actor, target, claims}, _from, state) do
    state = refresh_definition(state)

    with {:ok, ready_state} <- require_ready(state),
         :ok <- authorize_invite(actor),
         {:ok, principal} <- resolve_target_principal(target),
         {:ok, wallet_address} <- fetch_wallet_address(principal),
         :ok <- authorize_join(ready_state, principal, claims),
         false <- room_full?(ready_state, principal),
         {:ok, client, next_state} <-
           ensure_registered_client(ready_state, principal, wallet_address),
         {:ok, panel, updated_state} <- invite_joined_member(next_state, principal, client) do
      {:reply, {:ok, build_panel(updated_state, actor_or_target(actor, principal)) || panel},
       updated_state}
    else
      true -> {:reply, {:error, :room_full}, state}
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:kick_user, actor, target}, _from, state) do
    state = refresh_definition(state)

    with {:ok, ready_state} <- require_ready(state),
         :ok <- authorize_kick(ready_state, actor),
         {:ok, target_member} <- resolve_target_member(ready_state, target),
         {:ok, updated_state} <-
           remove_member(ready_state, target_member.wallet_address, target_member.inbox_id) do
      broadcast_refresh!(updated_state)
      {:reply, {:ok, build_panel(updated_state, actor_or_target(actor, nil))}, updated_state}
    else
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:moderator_delete_message, actor, message_id}, _from, state) do
    state = refresh_definition(state)

    with {:ok, ready_state} <- require_ready(state),
         {:ok, moderator_wallet} <- fetch_moderator_wallet(ready_state, actor),
         {:ok, _entry} <- tombstone_room_message(ready_state, message_id, moderator_wallet) do
      broadcast_refresh!(ready_state)
      {:reply, {:ok, build_panel(ready_state, actor_or_target(actor, nil))}, ready_state}
    else
      {:error, reason} -> {:reply, {:error, reason}, state}
    end
  end

  def handle_call({:bootstrap_room, opts}, _from, state) do
    state = refresh_definition(state)
    reuse? = Keyword.get(opts, :reuse, false)

    case bootstrap_room(state, reuse?) do
      {:ok, room_info} ->
        :ok = XmtpElixirSdk.Runtime.reset!(state.runtime_name)
        next_state = restore_state!(state)
        {:reply, {:ok, room_info}, next_state}

      {:error, reason} ->
        {:reply, {:error, reason}, state}
    end
  end

  def handle_call(:reset_for_test, _from, state) do
    state = refresh_definition(state)
    delete_room_data(state)
    :ok = XmtpElixirSdk.Runtime.reset!(state.runtime_name)
    {:reply, :ok, restore_state!(state)}
  end

  @impl true
  def handle_cast({:heartbeat, principal}, state) do
    state = refresh_definition(state)

    next_state =
      with {:ok, ready_state} <- require_ready(state),
           {:ok, principal} <- require_principal(principal),
           {:ok, wallet_address} <- fetch_wallet_address(principal),
           {:ok, client, updated_state} <-
             ensure_registered_client(ready_state, principal, wallet_address),
           :ok <- require_joined(updated_state, principal, wallet_address) do
        touch_membership_presence(updated_state, principal, wallet_address, client.inbox_id)
      else
        _ -> state
      end

    {:noreply, next_state}
  end

  @impl true
  def handle_info(:presence_tick, state) do
    state = refresh_definition(state)

    next_state =
      case require_ready(state) do
        {:ok, ready_state} -> expire_stale_memberships(ready_state)
        {:error, _} -> state
      end

    schedule_presence_tick(next_state)
    {:noreply, next_state}
  end

  def handle_info({:xmtp, _topic, %Events.MessageCreated{message: message}}, state) do
    state = refresh_definition(state)

    next_state =
      case require_ready(state) do
        {:ok, ready_state} ->
          if message.conversation_id == ready_state.public_room.id do
            persist_streamed_message(ready_state, message)
          else
            ready_state
          end

        {:error, _} ->
          state
      end

    broadcast_refresh!(next_state)
    {:noreply, next_state}
  end

  def handle_info({:xmtp, _topic, %Events.ConversationUpdated{conversation: conversation}}, state) do
    state = refresh_definition(state)

    next_state =
      case require_ready(state) do
        {:ok, ready_state} ->
          if conversation.id == ready_state.public_room.id do
            public_room =
              XmtpElixirSdk.Conversation.from_record(ready_state.relay_client, conversation)

            persist_room_snapshot(%{ready_state | public_room: public_room}, public_room)
          else
            ready_state
          end

        {:error, _} ->
          state
      end

    broadcast_refresh!(next_state)
    {:noreply, next_state}
  end

  def handle_info({:xmtp, _topic, _event}, state) do
    state = refresh_definition(state)
    broadcast_refresh!(state)
    {:noreply, state}
  end

  defp restore_state!(state) do
    case restore_state(state) do
      {:ok, next_state} ->
        next_state

      {:error, :agent_private_key_missing} ->
        unavailable_state(state, :room_unavailable)

      {:error, :room_not_bootstrapped} ->
        unavailable_state(state, :room_unavailable)

      {:error, reason} ->
        raise "XMTP room agent failed to start for #{state.definition.key}: #{inspect(reason)}"
    end
  end

  defp restore_state(state) do
    repo = state.repo

    with {:ok, private_key} <- configured_private_key(state),
         %Room{} = room <- load_room(repo, state.definition.key),
         {:ok, configured_wallet} <- Wallet.wallet_address(private_key),
         true <-
           configured_wallet == Principal.normalize_wallet(room.agent_wallet_address) or
             {:error, :agent_wallet_mismatch},
         :ok <- XmtpElixirSdk.Runtime.reset!(state.runtime_name),
         {:ok, relay_client} <- build_registered_client(state.runtime_name, private_key),
         true <- relay_client.inbox_id == room.agent_inbox_id or {:error, :agent_inbox_mismatch},
         {:ok, runtime_room} <- import_room_snapshot(state, relay_client, room),
         :ok <- subscribe_room(state.runtime_name, runtime_room.id, self()) do
      {:ok,
       %{
         state
         | mode: :ready,
           unavailable_reason: nil,
           relay_client: relay_client,
           public_room: runtime_room,
           room: room,
           clients_by_wallet: %{},
           pending_signatures: %{}
       }}
    else
      nil -> {:error, :room_not_bootstrapped}
      false -> {:error, :agent_wallet_mismatch}
      {:error, reason} -> {:error, reason}
    end
  end

  defp bootstrap_room(state, reuse?) do
    with {:ok, private_key} <- configured_private_key(state),
         {:ok, agent_wallet} <- Wallet.wallet_address(private_key) do
      case load_room(state.repo, state.definition.key) do
        %Room{} = room when reuse? ->
          {:ok, encode_room_info(room)}

        %Room{} ->
          {:error, :room_already_bootstrapped}

        nil ->
          :ok = XmtpElixirSdk.Runtime.reset!(state.runtime_name)

          with {:ok, relay_client} <- build_registered_client(state.runtime_name, private_key),
               {:ok, room} <-
                 Conversations.create_group_optimistic(
                   relay_client,
                   %Types.CreateGroupOptions{
                     name: state.definition.name,
                     description: state.definition.description,
                     app_data: state.definition.app_data
                   }
                 ),
               {:ok, room_record} <-
                 persist_bootstrapped_room(state, room, agent_wallet, relay_client.inbox_id) do
            {:ok, encode_room_info(room_record)}
          end
      end
    end
  end

  defp encode_room_info(%Room{} = room) do
    %{
      room_key: room.room_key,
      conversation_id: room.conversation_id,
      agent_wallet_address: room.agent_wallet_address,
      agent_inbox_id: room.agent_inbox_id
    }
  end

  defp persist_bootstrapped_room(state, %Conversation{} = room, agent_wallet, agent_inbox_id) do
    attrs = %{
      room_key: state.definition.key,
      conversation_id: room.id,
      agent_wallet_address: agent_wallet,
      agent_inbox_id: agent_inbox_id,
      status: "active",
      capacity: state.definition.capacity,
      room_name: room.name || state.definition.name,
      description: room.description || state.definition.description,
      app_data: room.app_data || state.definition.app_data,
      created_at_ns: room.created_at_ns,
      last_activity_ns: room.last_activity_ns,
      snapshot: room_snapshot(room)
    }

    %Room{}
    |> Room.changeset(attrs)
    |> state.repo.insert()
  end

  defp import_room_snapshot(state, relay_client, %Room{} = room) do
    conversation =
      build_runtime_conversation(relay_client, room, list_joined_memberships(state.repo, room))

    :ok = ConversationServer.import_conversations(state.runtime_name, [conversation])
    Conversations.get_by_id(relay_client, room.conversation_id)
  end

  defp build_runtime_conversation(relay_client, %Room{} = room, memberships) do
    members = [agent_member(relay_client) | Enum.map(memberships, &membership_to_group_member/1)]
    snapshot = room.snapshot || %{}

    metadata = %Types.ConversationMetadata{
      creator_inbox_id: room.agent_inbox_id,
      conversation_type: :group
    }

    %Types.Conversation{
      id: room.conversation_id,
      conversation_type: :group,
      created_at_ns: room.created_at_ns,
      metadata: metadata,
      added_by_inbox_id: room.agent_inbox_id,
      name: room.room_name,
      image_url: Map.get(snapshot, "image_url"),
      description: room.description || "",
      app_data: room.app_data || "",
      permissions: Types.default_permissions(),
      consent_state: :allowed,
      disappearing_settings: nil,
      paused_for_version: nil,
      pending_removal: false,
      last_activity_ns: room.last_activity_ns,
      members: Enum.uniq_by(members, & &1.inbox_id),
      admins: [room.agent_inbox_id],
      super_admins: [room.agent_inbox_id],
      hmac_keys: [],
      last_read_times: [],
      messages: []
    }
  end

  defp agent_member(relay_client) do
    %Types.GroupMember{
      inbox_id: relay_client.inbox_id,
      account_identifiers: [relay_client.identifier.identifier],
      installation_ids: [relay_client.installation_id],
      permission_level: :admin,
      consent_state: :allowed
    }
  end

  defp membership_to_group_member(%RoomMembership{} = membership) do
    %Types.GroupMember{
      inbox_id: membership.inbox_id,
      account_identifiers: [membership.wallet_address],
      installation_ids: [],
      permission_level: :member,
      consent_state: :allowed
    }
  end

  defp build_registered_client(runtime_name, private_key) do
    with {:ok, wallet_address} <- Wallet.wallet_address(private_key),
         identifier = wallet_identifier(wallet_address),
         {:ok, client} <- Clients.build(runtime_name, identifier, env: :dev),
         {:ok, %{signature_request_id: request_id, signature_text: signature_text}} <-
           Clients.unsafe_create_inbox_signature_text(client),
         {:ok, signature} <- Wallet.sign_personal_message(private_key, signature_text),
         {:ok, signer} <- Signer.eoa(identifier, signature),
         :ok <- Clients.unsafe_apply_signature_request(client, request_id, signer),
         {:ok, registered_client} <- Clients.register(client) do
      {:ok, registered_client}
    end
  end

  defp subscribe_room(runtime_name, room_id, pid) do
    :ok = Events.subscribe(runtime_name, {:messages, room_id}, pid)
    :ok = Events.subscribe(runtime_name, {:conversation, room_id}, pid)
  end

  defp build_panel(state, principal, copy_override \\ nil, pending_request_id \\ nil)

  defp build_panel(
         %{mode: :unavailable, definition: definition},
         principal,
         copy_override,
         pending_request_id
       ) do
    connected_wallet = principal && Principal.wallet(principal)

    RoomPanel.new!(%{
      room_key: definition.key,
      xmtp_group_id: nil,
      name: definition.name,
      status: :disabled,
      membership: if(connected_wallet, do: :not_joined, else: :not_connected),
      connected_wallet: connected_wallet,
      can_join: false,
      can_send: false,
      can_moderate: moderator_wallet?(definition, connected_wallet),
      pending_signature_request_id: pending_request_id,
      member_count: 0,
      active_member_count: 0,
      capacity: definition.capacity,
      seats_remaining: definition.capacity,
      presence_ttl_seconds: div(definition.presence_timeout_ms, 1_000),
      last_synced_at: nil,
      messages: [],
      user_copy: RoomPanel.copy(copy_override || "This room is unavailable right now.")
    })
  end

  defp build_panel(
         %{mode: :ready, room: room, definition: definition} = state,
         principal,
         copy_override,
         pending_signature_request_id
       ) do
    connected_wallet = principal && Principal.wallet(principal)

    membership =
      membership(state, principal, connected_wallet, pending_signature_request_id)

    joined? = membership == :joined
    moderator? = moderator_wallet?(definition, connected_wallet)
    capacity = room.capacity
    member_count = human_member_count(state.repo, room)
    active_member_count = active_human_member_count(state.repo, room, definition)
    seats_remaining = max(capacity - member_count, 0)

    pending_request_id =
      pending_signature_request_id || pending_request_id_for_wallet(state, connected_wallet)

    RoomPanel.new!(%{
      room_key: definition.key,
      xmtp_group_id: room.conversation_id,
      name: room.room_name,
      status: :ready,
      membership: membership,
      connected_wallet: connected_wallet,
      can_join: can_join?(membership, principal),
      can_send: joined?,
      can_moderate: moderator?,
      pending_signature_request_id: pending_request_id,
      member_count: member_count,
      active_member_count: active_member_count,
      capacity: capacity,
      seats_remaining: seats_remaining,
      presence_ttl_seconds: div(definition.presence_timeout_ms, 1_000),
      last_synced_at: room.updated_at,
      user_copy:
        RoomPanel.copy(copy_override || default_copy(membership, principal, seats_remaining)),
      messages: list_panel_messages(state, connected_wallet, moderator?)
    })
  end

  defp list_panel_messages(%{repo: repo, room: room}, connected_wallet, moderator?) do
    repo
    |> Log.list_messages(room)
    |> Enum.reject(&membership_change_message?/1)
    |> Enum.take(-@message_limit)
    |> Enum.map(fn message ->
      sender_wallet = Principal.normalize_wallet(message.sender_wallet)
      moderated? = message.website_visibility_state == "moderator_deleted"

      %{
        key: message.xmtp_message_id,
        author: author_label(message.sender_label, sender_wallet, message.sender_inbox_id),
        body: Log.website_body(message),
        stamp: format_stamp(message.sent_at),
        side: if(connected_wallet && sender_wallet == connected_wallet, do: :self, else: :other),
        sender_inbox_id: message.sender_inbox_id,
        sender_wallet: sender_wallet,
        sender_kind: normalize_sender_kind(message.sender_kind),
        website_state: if(moderated?, do: :moderator_deleted, else: :visible),
        can_delete?: moderator? and not moderated?,
        can_kick?: moderator? and sender_wallet != nil and sender_wallet != connected_wallet
      }
    end)
  end

  defp membership_change_message?(%MessageLog{message_snapshot: %{"kind" => :membership_change}}),
    do: true

  defp membership_change_message?(%MessageLog{
         message_snapshot: %{"kind" => "membership_change"}
       }),
       do: true

  defp membership_change_message?(%MessageLog{
         message_snapshot: %{"content_type_id" => "groupUpdated"}
       }),
       do: true

  defp membership_change_message?(%MessageLog{message_snapshot: %{"content_type_id" => type_id}}),
    do: type_id == "groupUpdated"

  defp membership_change_message?(%MessageLog{}), do: false

  defp membership(_state, nil, _wallet_address, _pending_request_id), do: :not_connected

  defp membership(state, principal, wallet_address, pending_request_id) do
    cond do
      joined?(state, wallet_address) ->
        :joined

      is_binary(pending_request_id) or pending_request_id_for_wallet(state, wallet_address) ->
        :pending_signature

      kicked?(state, wallet_address) ->
        :removed

      room_full?(state, principal) ->
        :blocked

      true ->
        :not_joined
    end
  end

  defp can_join?(:not_joined, %Principal{}), do: true
  defp can_join?(:removed, %Principal{}), do: true
  defp can_join?(_, _), do: false

  defp default_copy(:not_connected, nil, _seats_remaining), do: "Sign in to join this room."

  defp default_copy(:not_joined, principal, seats_remaining),
    do:
      "Connected as #{Principal.short(Principal.wallet(principal))}. #{seats_remaining} seats are open."

  defp default_copy(:pending_signature, _principal, _seats_remaining),
    do: "Check your wallet to finish joining."

  defp default_copy(:joined, principal, _seats_remaining),
    do: "Connected as #{Principal.short(Principal.wallet(principal))}. You are in the room."

  defp default_copy(:blocked, _principal, _seats_remaining),
    do: "This room is full right now. You can still read along."

  defp default_copy(:removed, _principal, seats_remaining),
    do:
      "You were removed from the room. Join again later if a seat opens. #{seats_remaining} seats are open."

  defp persist_streamed_message(%{repo: repo, room: room} = state, message) do
    sender_membership = fetch_membership_for_inbox(repo, room, message.sender_inbox_id)

    _ =
      Log.append_message(repo, room, message, %{
        wallet_address: sender_membership && sender_membership.wallet_address,
        kind: sender_membership && sender_membership.principal_kind,
        label: sender_membership && sender_membership.display_name
      })

    updated_room =
      room
      |> Room.changeset(%{last_activity_ns: message.sent_at_ns})
      |> repo.update!()

    %{state | room: updated_room}
  end

  defp fetch_message(client, message_id) do
    case Messages.get_by_id(client, message_id) do
      {:ok, %Types.Message{} = message} -> {:ok, message}
      {:ok, nil} -> {:error, :message_not_found}
      {:error, error} -> {:error, error}
    end
  end

  defp persist_room_snapshot(
         %{repo: repo, room: room} = state,
         %{id: _conversation_id} = conversation
       ) do
    updated_room =
      room
      |> Room.changeset(%{
        room_name: Map.get(conversation, :name),
        description: Map.get(conversation, :description),
        app_data: Map.get(conversation, :app_data),
        last_activity_ns: Map.get(conversation, :last_activity_ns),
        snapshot: room_snapshot(conversation)
      })
      |> repo.update!()

    %{state | room: updated_room, public_room: conversation}
  end

  defp room_snapshot(%{id: _conversation_id} = conversation) do
    %{
      name: Map.get(conversation, :name),
      description: Map.get(conversation, :description),
      app_data: Map.get(conversation, :app_data),
      created_at_ns: Map.get(conversation, :created_at_ns),
      last_activity_ns: Map.get(conversation, :last_activity_ns),
      image_url: Map.get(conversation, :image_url),
      added_by_inbox_id: Map.get(conversation, :added_by_inbox_id)
    }
  end

  defp invite_joined_member(
         %{public_room: public_room, room: room, repo: repo} = state,
         principal,
         client
       ) do
    with {:ok, updated_room} <- Groups.add_members(public_room, [client.inbox_id]),
         {:ok, _membership} <- upsert_membership(repo, room, principal, client.inbox_id, "joined") do
      next_state =
        state
        |> Map.put(:public_room, updated_room)
        |> Map.put(:room, repo.preload(room, :memberships, force: true))
        |> put_in([:clients_by_wallet, Principal.wallet(principal)], client)
        |> touch_membership_presence(principal, Principal.wallet(principal), client.inbox_id)
        |> persist_room_snapshot(updated_room)

      broadcast_refresh!(next_state)
      {:ok, build_panel(next_state, principal), next_state}
    end
  end

  defp remove_member(
         %{public_room: public_room, room: room, repo: repo} = state,
         wallet_address,
         inbox_id
       ) do
    if Enum.any?(public_room.members, &(&1.inbox_id == inbox_id)) do
      with {:ok, updated_room} <- Groups.remove_members(public_room, [inbox_id]),
           {:ok, _membership} <-
             upsert_membership(
               repo,
               room,
               %Principal{wallet_address: wallet_address},
               inbox_id,
               "kicked"
             ) do
        next_state =
          state
          |> Map.put(:public_room, updated_room)
          |> Map.put(:room, repo.preload(room, :memberships, force: true))
          |> persist_room_snapshot(updated_room)

        {:ok, next_state}
      end
    else
      {:error, :member_not_found}
    end
  end

  defp require_joined(state, principal, wallet_address) do
    cond do
      joined?(state, wallet_address) -> :ok
      kicked?(state, wallet_address) -> {:error, :kicked}
      room_full?(state, principal) -> {:error, :room_full}
      true -> {:error, :join_required}
    end
  end

  defp ensure_join_candidate(state, _principal, wallet_address) do
    case Map.fetch(state.clients_by_wallet, wallet_address) do
      {:ok, client} ->
        {:ok, client, state}

      :error ->
        create_fun =
          if existing_membership?(state.repo, state.room, wallet_address) do
            &Clients.create/3
          else
            &Clients.build/3
          end

        with {:ok, client} <-
               create_fun.(state.runtime_name, wallet_identifier(wallet_address), env: :dev) do
          {:ok, client, put_in(state.clients_by_wallet[wallet_address], client)}
        end
    end
  end

  defp ensure_registered_client(state, _principal, wallet_address) do
    case Map.fetch(state.clients_by_wallet, wallet_address) do
      {:ok, %Client{ready?: true} = client} ->
        {:ok, client, state}

      _ ->
        with {:ok, client} <-
               Clients.create(state.runtime_name, wallet_identifier(wallet_address), env: :dev) do
          {:ok, client, put_in(state.clients_by_wallet[wallet_address], client)}
        end
    end
  end

  defp fetch_cached_client(state, wallet_address) do
    case Map.fetch(state.clients_by_wallet, wallet_address) do
      {:ok, client} -> {:ok, client}
      :error -> {:error, :join_required}
    end
  end

  defp fetch_wallet_address(%Principal{} = principal) do
    case Principal.wallet(principal) do
      nil -> {:error, :wallet_required}
      wallet_address -> {:ok, wallet_address}
    end
  end

  defp fetch_wallet_address(_principal), do: {:error, :wallet_required}

  defp authorize_join(%{definition: definition}, %Principal{} = principal, claims) do
    definition.policy_module.allow_join(definition, principal, claims || %{})
  end

  defp fetch_moderator_wallet(%{definition: definition}, actor) do
    with {:ok, principal} <- resolve_actor_principal(actor),
         {:ok, wallet_address} <- fetch_wallet_address(principal),
         true <- moderator_wallet?(definition, wallet_address) do
      {:ok, wallet_address}
    else
      false -> {:error, :moderator_required}
      {:error, reason} -> {:error, reason}
    end
  end

  defp authorize_invite(:system), do: :ok
  defp authorize_invite(%Principal{}), do: :ok
  defp authorize_invite(_), do: {:error, :wallet_required}

  defp authorize_kick(state, actor) do
    fetch_moderator_wallet(state, actor)
    |> then(fn result -> if match?({:ok, _}, result), do: :ok, else: result end)
  end

  defp resolve_target_principal(%Principal{} = principal), do: {:ok, principal}
  defp resolve_target_principal(_target), do: {:error, :wallet_required}

  defp resolve_target_member(%{repo: repo, room: room}, %Principal{} = principal) do
    wallet_address = Principal.wallet(principal)
    resolve_target_member(%{repo: repo, room: room}, wallet_address)
  end

  defp resolve_target_member(%{repo: repo, room: room}, target_wallet_or_inbox)
       when is_binary(target_wallet_or_inbox) do
    normalized = String.downcase(String.trim(target_wallet_or_inbox))
    by_wallet = repo.get_by(RoomMembership, room_id: room.id, wallet_address: normalized)
    by_inbox = repo.get_by(RoomMembership, room_id: room.id, inbox_id: normalized)

    case by_wallet || by_inbox do
      %RoomMembership{} = membership ->
        {:ok, %{wallet_address: membership.wallet_address, inbox_id: membership.inbox_id}}

      nil ->
        {:error, :member_not_found}
    end
  end

  defp resolve_target_member(_state, _target), do: {:error, :member_not_found}

  defp validate_pending_signature(state, request_id, wallet_address, action) do
    case Map.get(state.pending_signatures, request_id) do
      %{wallet_address: ^wallet_address, action: ^action} -> :ok
      _ -> {:error, :signature_request_missing}
    end
  end

  defp tombstone_room_message(%{repo: repo, room: room}, message_id, moderator_wallet) do
    case Log.tombstone_message(repo, room, message_id, moderator_wallet) do
      {:ok, entry} -> {:ok, entry}
      {:error, :message_not_found} -> {:error, :message_not_found}
      {:error, _changeset} -> {:error, :message_not_found}
    end
  end

  defp room_full?(%{definition: definition, room: room, repo: repo}, %Principal{} = principal) do
    Principal.kind(principal) == :human and human_member_count(repo, room) >= definition.capacity
  end

  defp room_full?(_, _), do: false

  defp human_member_count(repo, %Room{} = room) do
    repo
    |> list_joined_memberships(room)
    |> Enum.count(&(&1.principal_kind == "human"))
  end

  defp active_human_member_count(repo, %Room{} = room, definition) do
    cutoff =
      DateTime.utc_now()
      |> DateTime.add(-div(definition.presence_timeout_ms, 1_000), :second)

    repo
    |> list_joined_memberships(room)
    |> Enum.count(fn membership ->
      membership.principal_kind == "human" and
        not is_nil(membership.last_seen_at) and
        DateTime.compare(membership.last_seen_at, cutoff) == :gt
    end)
  end

  defp joined?(%{repo: repo, room: room}, wallet_address) do
    case repo.get_by(RoomMembership,
           room_id: room.id,
           wallet_address: Principal.normalize_wallet(wallet_address)
         ) do
      %RoomMembership{membership_state: "joined"} -> true
      _ -> false
    end
  end

  defp existing_membership?(repo, %Room{} = room, wallet_address) do
    not is_nil(
      repo.get_by(RoomMembership,
        room_id: room.id,
        wallet_address: Principal.normalize_wallet(wallet_address)
      )
    )
  end

  defp kicked?(%{repo: repo, room: room}, wallet_address) do
    case repo.get_by(RoomMembership,
           room_id: room.id,
           wallet_address: Principal.normalize_wallet(wallet_address)
         ) do
      %RoomMembership{membership_state: "kicked"} -> true
      _ -> false
    end
  end

  defp pending_request_id_for_wallet(_state, nil), do: nil

  defp pending_request_id_for_wallet(state, wallet_address) do
    Enum.find_value(state.pending_signatures, fn {request_id, pending} ->
      if pending.wallet_address == wallet_address, do: request_id, else: nil
    end)
  end

  defp upsert_membership(
         repo,
         %Room{} = room,
         %Principal{} = principal,
         inbox_id,
         membership_state
       ) do
    attrs = %{
      room_id: room.id,
      wallet_address: Principal.wallet(principal),
      inbox_id: inbox_id,
      principal_kind: Atom.to_string(Principal.kind(principal) || :human),
      display_name: Principal.label(principal),
      membership_state: membership_state,
      last_seen_at: DateTime.utc_now(),
      metadata: principal.metadata || %{}
    }

    existing = repo.get_by(RoomMembership, room_id: room.id, wallet_address: attrs.wallet_address)

    case existing do
      nil ->
        %RoomMembership{}
        |> RoomMembership.changeset(attrs)
        |> repo.insert()

      membership ->
        membership
        |> RoomMembership.changeset(attrs)
        |> repo.update()
    end
  end

  defp touch_membership_presence(
         %{repo: repo, room: room} = state,
         principal,
         _wallet_address,
         inbox_id
       ) do
    _ = upsert_membership(repo, room, principal, inbox_id, "joined")
    %{state | room: repo.preload(room, :memberships, force: true)}
  end

  defp expire_stale_memberships(%{repo: repo, room: room, definition: definition} = state) do
    cutoff =
      DateTime.utc_now()
      |> DateTime.add(-div(definition.presence_timeout_ms, 1_000), :second)

    stale_memberships =
      RoomMembership
      |> where([membership], membership.room_id == ^room.id)
      |> where([membership], membership.membership_state == "joined")
      |> where(
        [membership],
        not is_nil(membership.last_seen_at) and membership.last_seen_at <= ^cutoff
      )
      |> repo.all()

    Enum.reduce(stale_memberships, state, fn membership, acc ->
      case remove_member(acc, membership.wallet_address, membership.inbox_id) do
        {:ok, updated_state} ->
          broadcast_refresh!(updated_state)
          updated_state

        {:error, _reason} ->
          acc
      end
    end)
  end

  defp list_joined_memberships(repo, %Room{id: room_id}) do
    RoomMembership
    |> where([membership], membership.room_id == ^room_id)
    |> where([membership], membership.membership_state == "joined")
    |> order_by([membership], asc: membership.inserted_at)
    |> repo.all()
  end

  defp fetch_membership_for_inbox(repo, %Room{} = room, inbox_id) do
    repo.get_by(RoomMembership, room_id: room.id, inbox_id: inbox_id)
  end

  defp author_label(label, _wallet_address, _inbox_id) when is_binary(label) and label != "",
    do: label

  defp author_label(_label, nil, inbox_id), do: Principal.short(inbox_id)
  defp author_label(_label, wallet_address, _inbox_id), do: Principal.short(wallet_address)

  defp format_stamp(%DateTime{} = datetime), do: Calendar.strftime(datetime, "%b %-d %H:%M")

  defp normalize_body(body) when is_binary(body), do: String.trim(body)
  defp normalize_body(_body), do: ""

  defp validate_body(""), do: {:error, :message_required}
  defp validate_body(body) when byte_size(body) > 2_000, do: {:error, :message_too_long}
  defp validate_body(_body), do: :ok

  defp wallet_identifier(wallet_address) do
    %Types.Identifier{
      identifier: Principal.normalize_wallet(wallet_address),
      identifier_kind: :ethereum
    }
  end

  defp moderator_wallet?(_definition, nil), do: false

  defp moderator_wallet?(definition, wallet_address) do
    Principal.normalize_wallet(wallet_address) in Enum.map(
      definition.moderator_wallets,
      &Principal.normalize_wallet/1
    )
  end

  defp configured_private_key(%{definition: definition}) do
    definition.agent_private_key
    |> Wallet.normalize_private_key()
  end

  defp schedule_presence_tick(state) do
    Process.send_after(self(), :presence_tick, state.definition.presence_check_interval_ms)
  end

  defp load_room(repo, room_key) do
    Room
    |> repo.get_by(room_key: room_key)
    |> case do
      nil -> nil
      room -> repo.preload(room, [:memberships])
    end
  end

  defp require_ready(%{mode: :ready} = state), do: {:ok, state}
  defp require_ready(_state), do: {:error, :room_unavailable}

  defp require_principal(%Principal{} = principal), do: {:ok, principal}
  defp require_principal(_principal), do: {:error, :wallet_required}

  defp resolve_actor_principal(%Principal{} = principal), do: {:ok, principal}
  defp resolve_actor_principal(:system), do: {:error, :moderator_required}
  defp resolve_actor_principal(_), do: {:error, :wallet_required}

  defp unavailable_state(state, reason) do
    %{
      state
      | mode: :unavailable,
        unavailable_reason: reason,
        relay_client: nil,
        public_room: nil,
        room: nil,
        clients_by_wallet: %{},
        pending_signatures: %{}
    }
  end

  defp actor_or_target(:system, target), do: target
  defp actor_or_target(%Principal{} = actor, _target), do: actor
  defp actor_or_target(_, target), do: target

  defp refresh_definition(%{definition_loader: {:mfa, module, function, args}} = state) do
    case apply(module, function, args) do
      %Xmtp.RoomDefinition{} = definition -> %{state | definition: definition}
      _ -> state
    end
  end

  defp refresh_definition(state), do: state

  defp broadcast_refresh!(state) do
    Phoenix.PubSub.broadcast(
      state.pubsub,
      Manager.topic(state.manager, state.definition.key),
      {:xmtp_public_room, :refresh}
    )
  end

  defp delete_room_data(state) do
    repo = state.repo

    case load_room(repo, state.definition.key) do
      %Room{} = room ->
        repo.delete_all(from(log in MessageLog, where: log.room_id == ^room.id))
        repo.delete_all(from(membership in RoomMembership, where: membership.room_id == ^room.id))
        repo.delete(room)
        :ok

      nil ->
        :ok
    end
  end

  defp normalize_sender_kind("agent"), do: :agent
  defp normalize_sender_kind(_), do: :human
end