WIP

Signed-off-by: Thomas Citharel <tcit@tcit.fr>

lib/federation/activity_pub/actions/create.ex

@@ -14,7 +14,15 @@ defmodule Mobilizon.Federation.ActivityPub.Actions.Create do
     ]
 
   @type create_entities ::
-          :event | :comment | :discussion | :actor | :todo_list | :todo | :resource | :post
+          :event
+          | :comment
+          | :discussion
+          | :conversation
+          | :actor
+          | :todo_list
+          | :todo
+          | :resource
+          | :post
 
   @doc """
   Create an activity of type `Create`
@@ -50,18 +58,27 @@ defmodule Mobilizon.Federation.ActivityPub.Actions.Create do
     end
   end
 
+  @map_types %{
+    :event => Types.Events,
+    :comment => Types.Comments,
+    :discussion => Types.Discussions,
+    :conversation => Types.Conversations,
+    :actor => Types.Actors,
+    :todo_list => Types.TodoLists,
+    :todo => Types.Todos,
+    :resource => Types.Resources,
+    :post => Types.Posts
+  }
+
   @spec do_create(create_entities(), map(), map()) ::
           {:ok, Entity.t(), Activity.t()} | {:error, Ecto.Changeset.t() | atom()}
   defp do_create(type, args, additional) do
-    case type do
-      :event -> Types.Events.create(args, additional)
-      :comment -> Types.Comments.create(args, additional)
-      :discussion -> Types.Discussions.create(args, additional)
-      :actor -> Types.Actors.create(args, additional)
-      :todo_list -> Types.TodoLists.create(args, additional)
-      :todo -> Types.Todos.create(args, additional)
-      :resource -> Types.Resources.create(args, additional)
-      :post -> Types.Posts.create(args, additional)
+    mod = Map.get(@map_types, type)
+
+    if is_nil(mod) do
+      {:error, :type_not_supported}
+    else
+      mod.create(args, additional)
     end
   end
 

lib/federation/activity_pub/audience.ex

@@ -5,6 +5,7 @@ defmodule Mobilizon.Federation.ActivityPub.Audience do
 
   alias Mobilizon.{Actors, Discussions, Events, Share}
   alias Mobilizon.Actors.{Actor, Member}
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.{Comment, Discussion}
   alias Mobilizon.Events.{Event, Participant}
   alias Mobilizon.Federation.ActivityPub.Types.Entity
@@ -38,6 +39,10 @@ defmodule Mobilizon.Federation.ActivityPub.Audience do
     %{"to" => maybe_add_group_members([], actor), "cc" => []}
   end
 
+  def get_audience(%Conversation{participants: participants}) do
+    %{"to" => Enum.map(participants, & &1.url), "cc" => []}
+  end
+
   # Deleted comments are just like tombstones
   def get_audience(%Comment{deleted_at: deleted_at}) when not is_nil(deleted_at) do
     %{"to" => [@ap_public], "cc" => []}

lib/federation/activity_pub/fetcher.ex

@@ -177,7 +177,7 @@ defmodule Mobilizon.Federation.ActivityPub.Fetcher do
         {:error, :content_not_json}
 
       {:ok, %Tesla.Env{} = res} ->
-        Logger.debug("Resource returned bad HTTP code #{inspect(res)}")
+        Logger.debug("Resource returned bad HTTP code (#{res.status}) #{inspect(res)}")
         {:error, :http_error}
 
       {:error, err} ->

lib/federation/activity_pub/transmogrifier.ex

@@ -68,24 +68,26 @@ defmodule Mobilizon.Federation.ActivityPub.Transmogrifier do
   def handle_incoming(%{"type" => "Create", "object" => %{"type" => "Note"} = object}) do
     Logger.info("Handle incoming to create notes")
 
-    case Converter.Comment.as_to_model_data(object) do
-      %{visibility: visibility, event_id: event_id}
-      when visibility != :public and event_id != nil ->
-        Logger.info("Tried to reply to an event with a private comment - ignore")
-        :error
+    case Discussions.get_comment_from_url_with_preload(object["id"]) do
+      {:error, :comment_not_found} ->
+        case Converter.Comment.as_to_model_data(object) do
+          %{visibility: visibility} = object_data
+          when visibility === :private ->
+            Actions.Create.create(:conversation, object_data, false)
 
-      object_data when is_map(object_data) ->
-        case Discussions.get_comment_from_url_with_preload(object_data.url) do
-          {:error, :comment_not_found} ->
-            object_data
-            |> transform_object_data_for_discussion()
-            |> save_comment_or_discussion()
-
-          {:ok, %Comment{} = comment} ->
-            # Object already exists
-            {:ok, nil, comment}
+          object_data when is_map(object_data) ->
+            case Discussions.get_comment_from_url_with_preload(object_data.url) do
+              {:error, :comment_not_found} ->
+                object_data
+                |> transform_object_data_for_discussion()
+                |> save_comment_or_discussion()
+            end
         end
 
+      {:ok, %Comment{} = comment} ->
+        # Object already exists
+        {:ok, nil, comment}
+
       {:error, err} ->
         {:error, err}
     end

lib/federation/activity_pub/types/conversation.ex

@@ -0,0 +1,207 @@
+defmodule Mobilizon.Federation.ActivityPub.Types.Conversations do
+  @moduledoc false
+
+  # alias Mobilizon.Conversations.ConversationParticipant
+  alias Mobilizon.{Actors, Conversations, Discussions}
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.Conversation
+  alias Mobilizon.Discussions.Comment
+  alias Mobilizon.Federation.ActivityPub.Actor, as: ActivityPubActor
+  alias Mobilizon.Federation.ActivityPub.{Audience, Permission}
+  alias Mobilizon.Federation.ActivityPub.Types.Entity
+  alias Mobilizon.Federation.ActivityStream
+  alias Mobilizon.Federation.ActivityStream.Converter.Utils, as: ConverterUtils
+  alias Mobilizon.Federation.ActivityStream.Convertible
+  alias Mobilizon.GraphQL.API.Utils, as: APIUtils
+  alias Mobilizon.Service.Activity.Conversation, as: ConversationActivity
+  alias Mobilizon.Web.Endpoint
+  import Mobilizon.Federation.ActivityPub.Utils, only: [make_create_data: 2, make_update_data: 2]
+  require Logger
+
+  @behaviour Entity
+
+  @impl Entity
+  @spec create(map(), map()) ::
+          {:ok, Conversation.t(), ActivityStream.t()}
+          | {:error, :conversation_not_found | :last_comment_not_found | Ecto.Changeset.t()}
+  def create(%{conversation_id: conversation_id} = args, additional)
+      when not is_nil(conversation_id) do
+    Logger.debug("Creating a reply to a conversation #{inspect(args, pretty: true)}")
+    args = prepare_args(args)
+    Logger.debug("Creating a reply to a conversation #{inspect(args, pretty: true)}")
+
+    case Conversations.get_conversation(conversation_id) do
+      %Conversation{} = conversation ->
+        case Conversations.reply_to_conversation(conversation, args) do
+          {:ok, %Conversation{last_comment_id: last_comment_id} = conversation} ->
+            ConversationActivity.insert_activity(conversation, subject: "conversation_replied")
+            maybe_publish_graphql_subscription(conversation)
+
+            case Discussions.get_comment_with_preload(last_comment_id) do
+              %Comment{} = last_comment ->
+                comment_as_data = Convertible.model_to_as(last_comment)
+                audience = Audience.get_audience(conversation)
+                create_data = make_create_data(comment_as_data, Map.merge(audience, additional))
+                {:ok, conversation, create_data}
+
+              nil ->
+                {:error, :last_comment_not_found}
+            end
+
+          {:error, _, %Ecto.Changeset{} = err, _} ->
+            {:error, err}
+        end
+
+      nil ->
+        {:error, :discussion_not_found}
+    end
+  end
+
+  @impl Entity
+  def create(args, additional) do
+    with args when is_map(args) <- prepare_args(args) do
+      case Conversations.create_conversation(args) do
+        {:ok, %Conversation{} = conversation} ->
+          ConversationActivity.insert_activity(conversation, subject: "conversation_created")
+          conversation_as_data = Convertible.model_to_as(conversation)
+          audience = Audience.get_audience(conversation)
+          create_data = make_create_data(conversation_as_data, Map.merge(audience, additional))
+          {:ok, conversation, create_data}
+
+        {:error, _, %Ecto.Changeset{} = err, _} ->
+          {:error, err}
+      end
+    end
+  end
+
+  @impl Entity
+  @spec update(Conversation.t(), map(), map()) ::
+          {:ok, Conversation.t(), ActivityStream.t()} | {:error, Ecto.Changeset.t()}
+  def update(%Conversation{} = old_conversation, args, additional) do
+    case Conversations.update_conversation(old_conversation, args) do
+      {:ok, %Conversation{} = new_conversation} ->
+        # ConversationActivity.insert_activity(new_conversation,
+        #   subject: "conversation_renamed",
+        #   old_conversation: old_conversation
+        # )
+
+        conversation_as_data = Convertible.model_to_as(new_conversation)
+        audience = Audience.get_audience(new_conversation)
+        update_data = make_update_data(conversation_as_data, Map.merge(audience, additional))
+        {:ok, new_conversation, update_data}
+
+      {:error, %Ecto.Changeset{} = err} ->
+        {:error, err}
+    end
+  end
+
+  @impl Entity
+  @spec delete(Conversation.t(), Actor.t(), boolean, map()) ::
+          {:error, Ecto.Changeset.t()} | {:ok, ActivityStream.t(), Actor.t(), Conversation.t()}
+  def delete(
+        %Conversation{} = _conversation,
+        %Actor{} = _actor,
+        _local,
+        _additionnal
+      ) do
+    {:error, :not_applicable}
+  end
+
+  # @spec actor(Conversation.t()) :: Actor.t() | nil
+  # def actor(%ConversationParticipant{actor_id: actor_id}), do: Actors.get_actor(actor_id)
+
+  # @spec group_actor(Conversation.t()) :: Actor.t() | nil
+  # def group_actor(%Conversation{actor_id: actor_id}), do: Actors.get_actor(actor_id)
+
+  @spec permissions(Conversation.t()) :: Permission.t()
+  def permissions(%Conversation{}) do
+    %Permission{access: :member, create: :member, update: :moderator, delete: :moderator}
+  end
+
+  @spec maybe_publish_graphql_subscription(Conversation.t()) :: :ok
+  defp maybe_publish_graphql_subscription(%Conversation{} = conversation) do
+    Absinthe.Subscription.publish(Endpoint, conversation,
+      conversation_comment_changed: conversation.id
+    )
+
+    :ok
+  end
+
+  @spec prepare_args(map) :: map | {:error, :empty_participants}
+  defp prepare_args(args) do
+    {text, mentions, _tags} =
+      APIUtils.make_content_html(
+        args |> Map.get(:text, "") |> String.trim(),
+        # Can't put additional tags on a comment
+        [],
+        "text/html"
+      )
+
+    mentions =
+      (args |> Map.get(:mentions, []) |> prepare_mentions()) ++
+        ConverterUtils.fetch_mentions(mentions)
+
+    if Enum.empty?(mentions) do
+      {:error, :empty_participants}
+    else
+      event = Map.get(args, :event, get_event(Map.get(args, :event_id)))
+
+      participants =
+        (mentions ++
+           [
+             %{actor_id: args.actor_id},
+             %{
+               actor_id:
+                 if(is_nil(event),
+                   do: nil,
+                   else: event.attributed_to_id || event.organizer_actor_id
+                 )
+             }
+           ])
+        |> Enum.reduce(
+          [],
+          fn %{actor_id: actor_id}, acc ->
+            case Actors.get_actor(actor_id) do
+              nil -> acc
+              actor -> acc ++ [actor]
+            end
+          end
+        )
+        |> Enum.uniq_by(& &1.id)
+
+      args
+      |> Map.put(:text, text)
+      |> Map.put(:mentions, mentions)
+      |> Map.put(:participants, participants)
+    end
+  end
+
+  @spec prepare_mentions(list(String.t())) :: list(%{actor_id: String.t()})
+  defp prepare_mentions(mentions) do
+    Enum.reduce(mentions, [], &prepare_mention/2)
+  end
+
+  @spec prepare_mention(String.t() | map(), list()) :: list(%{actor_id: String.t()})
+  defp prepare_mention(%{actor_id: _} = mention, mentions) do
+    mentions ++ [mention]
+  end
+
+  defp prepare_mention(mention, mentions) do
+    case ActivityPubActor.find_or_make_actor_from_nickname(mention) do
+      {:ok, %Actor{id: actor_id}} ->
+        mentions ++ [%{actor_id: actor_id}]
+
+      {:error, _} ->
+        mentions
+    end
+  end
+
+  defp get_event(nil), do: nil
+
+  defp get_event(event_id) do
+    case Mobilizon.Events.get_event(event_id) do
+      {:ok, event} -> event
+      _ -> nil
+    end
+  end
+end

lib/federation/activity_pub/utils.ex

@@ -22,6 +22,7 @@ defmodule Mobilizon.Federation.ActivityPub.Utils do
 
   @actor_types ["Group", "Person", "Application"]
   @all_actor_types @actor_types ++ ["Organization", "Service"]
+  @ap_public_audience "https://www.w3.org/ns/activitystreams#Public"
 
   # Wraps an object into an activity
   @spec create_activity(map(), boolean()) :: {:ok, Activity.t()}
@@ -491,8 +492,7 @@ defmodule Mobilizon.Federation.ActivityPub.Utils do
       if public do
         Logger.debug("Making announce data for a public object")
 
-        {[actor.followers_url, object_actor_url],
-         ["https://www.w3.org/ns/activitystreams#Public"]}
+        {[actor.followers_url, object_actor_url], [@ap_public_audience]}
       else
         Logger.debug("Making announce data for a private object")
 
@@ -539,7 +539,7 @@ defmodule Mobilizon.Federation.ActivityPub.Utils do
       "actor" => url,
       "object" => activity,
       "to" => [actor.followers_url, actor.url],
-      "cc" => ["https://www.w3.org/ns/activitystreams#Public"]
+      "cc" => [@ap_public_audience]
     }
 
     if activity_id, do: Map.put(data, "id", activity_id), else: data

lib/federation/activity_stream/converter/comment.ex

@@ -47,9 +47,6 @@ defmodule Mobilizon.Federation.ActivityStream.Converter.Comment do
 
     case maybe_fetch_actor_and_attributed_to_id(object) do
       {:ok, %Actor{id: actor_id, domain: actor_domain}, attributed_to} ->
-        Logger.debug("Inserting full comment")
-        Logger.debug(inspect(object))
-
         data = %{
           text: object["content"],
           url: object["id"],
@@ -70,14 +67,7 @@ defmodule Mobilizon.Federation.ActivityStream.Converter.Comment do
           is_announcement: Map.get(object, "isAnnouncement", false)
         }
 
-        Logger.debug("Converted object before fetching parents")
-        Logger.debug(inspect(data))
-
-        data = maybe_fetch_parent_object(object, data)
-
-        Logger.debug("Converted object after fetching parents")
-        Logger.debug(inspect(data))
-        data
+        maybe_fetch_parent_object(object, data)
 
       {:error, err} ->
         {:error, err}
@@ -147,19 +137,22 @@ defmodule Mobilizon.Federation.ActivityStream.Converter.Comment do
   end
 
   @spec determine_to(CommentModel.t()) :: [String.t()]
-  defp determine_to(%CommentModel{} = comment) do
-    cond do
-      not is_nil(comment.attributed_to) ->
-        [comment.attributed_to.url]
+  defp determine_to(%CommentModel{visibility: :private, mentions: mentions} = _comment) do
+    Enum.map(mentions, fn mention -> mention.actor.url end)
+  end
 
-      comment.visibility == :public ->
-        ["https://www.w3.org/ns/activitystreams#Public"]
-
-      true ->
-        [comment.actor.followers_url]
+  defp determine_to(%CommentModel{visibility: :public} = comment) do
+    if is_nil(comment.attributed_to) do
+      ["https://www.w3.org/ns/activitystreams#Public"]
+    else
+      [comment.attributed_to.url]
     end
   end
 
+  defp determine_to(%CommentModel{} = comment) do
+    [comment.actor.followers_url]
+  end
+
   defp maybe_fetch_parent_object(object, data) do
     # We fetch the parent object
     Logger.debug("We're fetching the parent object")
@@ -170,9 +163,12 @@ defmodule Mobilizon.Federation.ActivityStream.Converter.Comment do
 
       case ActivityPub.fetch_object_from_url(object["inReplyTo"]) do
         # Reply to an event (Event)
-        {:ok, %Event{id: id}} ->
+        {:ok, %Event{id: id} = event} ->
           Logger.debug("Parent object is an event")
-          data |> Map.put(:event_id, id)
+
+          data
+          |> Map.put(:event_id, id)
+          |> Map.put(:event, event)
 
         # Reply to a comment (Comment)
         {:ok, %CommentModel{id: id} = comment} ->
@@ -182,6 +178,7 @@ defmodule Mobilizon.Federation.ActivityStream.Converter.Comment do
           |> Map.put(:in_reply_to_comment_id, id)
           |> Map.put(:origin_comment_id, comment |> CommentModel.get_thread_id())
           |> Map.put(:event_id, comment.event_id)
+          |> Map.put(:conversation_id, comment.conversation_id)
 
         # Reply to a discucssion (Discussion)
         {:ok,

lib/federation/activity_stream/converter/conversation.ex

@@ -0,0 +1,68 @@
+defmodule Mobilizon.Federation.ActivityStream.Converter.Conversation do
+  @moduledoc """
+  Comment converter.
+
+  This module allows to convert conversations from ActivityStream format to our own
+  internal one, and back.
+  """
+
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.Conversation
+  alias Mobilizon.Federation.ActivityPub.Actor, as: ActivityPubActor
+  alias Mobilizon.Federation.ActivityStream.{Converter, Convertible}
+  alias Mobilizon.Federation.ActivityStream.Converter.Conversation, as: ConversationConverter
+  alias Mobilizon.Storage.Repo
+  import Mobilizon.Service.Guards, only: [is_valid_string: 1]
+
+  require Logger
+
+  @behaviour Converter
+
+  defimpl Convertible, for: Conversation do
+    defdelegate model_to_as(comment), to: ConversationConverter
+  end
+
+  @doc """
+  Make an AS comment object from an existing `conversation` structure.
+  """
+  @impl Converter
+  @spec model_to_as(Conversation.t()) :: map
+  def model_to_as(%Conversation{} = conversation) do
+    conversation = Repo.preload(conversation, [:participants, last_comment: [:actor]])
+
+    %{
+      "type" => "Note",
+      "to" => Enum.map(conversation.participants, & &1.url),
+      "cc" => [],
+      "content" => conversation.last_comment.text,
+      "mediaType" => "text/html",
+      "actor" => conversation.last_comment.actor.url,
+      "id" => conversation.last_comment.url,
+      "publishedAt" => conversation.inserted_at
+    }
+  end
+
+  @impl Converter
+  @spec as_to_model_data(map) :: map() | {:error, atom()}
+  def as_to_model_data(%{"type" => "Note", "name" => name} = object) when is_valid_string(name) do
+    with %{actor_id: actor_id, creator_id: creator_id} <- extract_actors(object) do
+      %{actor_id: actor_id, creator_id: creator_id, title: name, url: object["id"]}
+    end
+  end
+
+  @spec extract_actors(map()) ::
+          %{actor_id: String.t(), creator_id: String.t()} | {:error, atom()}
+  defp extract_actors(%{"actor" => creator_url, "attributedTo" => actor_url} = _object)
+       when is_valid_string(creator_url) and is_valid_string(actor_url) do
+    with {:ok, %Actor{id: creator_id, suspended: false}} <-
+           ActivityPubActor.get_or_fetch_actor_by_url(creator_url),
+         {:ok, %Actor{id: actor_id, suspended: false}} <-
+           ActivityPubActor.get_or_fetch_actor_by_url(actor_url) do
+      %{actor_id: actor_id, creator_id: creator_id}
+    else
+      {:error, error} -> {:error, error}
+      {:ok, %Actor{url: ^creator_url}} -> {:error, :creator_suspended}
+      {:ok, %Actor{url: ^actor_url}} -> {:error, :actor_suspended}
+    end
+  end
+end

lib/federation/web_finger/web_finger.ex

@@ -242,12 +242,15 @@ defmodule Mobilizon.Federation.WebFinger do
   @spec domain_from_federated_actor(String.t()) :: {:ok, String.t()} | {:error, :host_not_found}
   defp domain_from_federated_actor(actor) do
     case String.split(actor, "@") do
+      [_name, ""] ->
+        {:error, :host_not_found}
+
       [_name, domain] ->
         {:ok, domain}
 
       _e ->
         host = URI.parse(actor).host
-        if is_nil(host), do: {:error, :host_not_found}, else: {:ok, host}
+        if is_nil(host) or host == "", do: {:error, :host_not_found}, else: {:ok, host}
     end
   end
 

lib/graphql/api/comments.ex

@@ -4,7 +4,8 @@ defmodule Mobilizon.GraphQL.API.Comments do
   """
 
   alias Mobilizon.Actors.Actor
-  alias Mobilizon.Discussions.Comment
+  alias Mobilizon.Conversations.Conversation
+  alias Mobilizon.Discussions.{Comment, Discussion}
   alias Mobilizon.Federation.ActivityPub.{Actions, Activity}
   alias Mobilizon.GraphQL.API.Utils
 
@@ -53,6 +54,22 @@ defmodule Mobilizon.GraphQL.API.Comments do
     )
   end
 
+  @doc """
+  Creates a conversation (or reply to a conversation)
+  """
+  @spec create_conversation(map()) ::
+          {:ok, Activity.t(), Conversation.t()}
+          | {:error, :entity_tombstoned | atom | Ecto.Changeset.t()}
+  def create_conversation(args) do
+    args = extract_pictures_from_comment_body(args)
+
+    Actions.Create.create(
+      :conversation,
+      args,
+      true
+    )
+  end
+
   @spec extract_pictures_from_comment_body(map()) :: map()
   defp extract_pictures_from_comment_body(%{text: text, actor_id: actor_id} = args) do
     pictures = Utils.extract_pictures_from_body(text, actor_id)

lib/graphql/api/events.ex

@@ -4,8 +4,8 @@ defmodule Mobilizon.GraphQL.API.Events do
   """
 
   alias Mobilizon.Actors.Actor
+  alias Mobilizon.Discussions.Comment
   alias Mobilizon.Events.Event
-
   alias Mobilizon.Federation.ActivityPub.{Actions, Activity, Utils}
   alias Mobilizon.GraphQL.API.Utils, as: APIUtils
 
@@ -36,6 +36,12 @@ defmodule Mobilizon.GraphQL.API.Events do
     Actions.Delete.delete(event, actor, true)
   end
 
+  @spec send_private_message_to_participants(map()) ::
+          {:ok, Activity.t(), Comment.t()} | {:error, atom() | Ecto.Changeset.t()}
+  def send_private_message_to_participants(args) do
+    Actions.Create.create(:comment, args, true)
+  end
+
   @spec prepare_args(map) :: map
   defp prepare_args(args) do
     organizer_actor = Map.get(args, :organizer_actor)

lib/graphql/api/search.ex

@@ -116,13 +116,9 @@ defmodule Mobilizon.GraphQL.API.Search do
   @spec process_from_username(String.t()) :: Page.t(Actor.t())
   defp process_from_username(search) do
     case ActivityPubActor.find_or_make_actor_from_nickname(search) do
-      {:ok, %Actor{type: :Group} = actor} ->
+      {:ok, %Actor{} = actor} ->
         %Page{total: 1, elements: [actor]}
 
-      # Don't return anything else than groups
-      {:ok, %Actor{}} ->
-        %Page{total: 0, elements: []}
-
       {:error, _err} ->
         Logger.debug(fn -> "Unable to find or make actor '#{search}'" end)
 

lib/graphql/authorization.ex

@@ -16,11 +16,13 @@ defmodule Mobilizon.GraphQL.Authorization do
   @impl true
   def has_user_access?(%User{}, _scope, _rule), do: true
 
+  @impl true
   def has_user_access?(%ApplicationToken{scope: scope} = _current_app_token, _struct, rule)
       when rule != :forbid_app_access do
     AppScope.has_app_access?(scope, rule)
   end
 
+  @impl true
   def has_user_access?(_current_user, _scoped_struct, _rule), do: false
 
   @impl true

lib/graphql/resolvers/conversation.ex

@@ -0,0 +1,269 @@
+defmodule Mobilizon.GraphQL.Resolvers.Conversation do
+  @moduledoc """
+  Handles the group-related GraphQL calls.
+  """
+
+  alias Mobilizon.{Actors, Conversations}
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.{Conversation, ConversationParticipant, ConversationView}
+  alias Mobilizon.Events.Event
+  alias Mobilizon.GraphQL.API.Comments
+  alias Mobilizon.Storage.Page
+  alias Mobilizon.Users.User
+  alias Mobilizon.Web.Endpoint
+  # alias Mobilizon.Users.User
+  import Mobilizon.Web.Gettext, only: [dgettext: 2]
+
+  @spec find_conversations_for_event(Event.t(), map(), Absinthe.Resolution.t()) ::
+          {:ok, Page.t(ConversationView.t())} | {:error, :unauthenticated}
+  def find_conversations_for_event(
+        %Event{id: event_id, attributed_to_id: attributed_to_id},
+        %{page: page, limit: limit},
+        %{
+          context: %{
+            current_actor: %Actor{id: actor_id}
+          }
+        }
+      )
+      when not is_nil(attributed_to_id) do
+    if Actors.is_member?(actor_id, attributed_to_id) do
+      {:ok,
+       event_id
+       |> Conversations.find_conversations_for_event(actor_id, page, limit)
+       |> conversation_participant_to_view()}
+    else
+      {:ok, %Page{total: 0, elements: []}}
+    end
+  end
+
+  @spec find_conversations_for_event(Event.t(), map(), Absinthe.Resolution.t()) ::
+          {:ok, Page.t(ConversationView.t())} | {:error, :unauthenticated}
+  def find_conversations_for_event(
+        %Event{id: event_id, organizer_actor_id: organizer_actor_id},
+        %{page: page, limit: limit},
+        %{
+          context: %{
+            current_actor: %Actor{id: actor_id}
+          }
+        }
+      ) do
+    if organizer_actor_id == actor_id do
+      {:ok,
+       event_id
+       |> Conversations.find_conversations_for_event(actor_id, page, limit)
+       |> conversation_participant_to_view()}
+    else
+      {:ok, %Page{total: 0, elements: []}}
+    end
+  end
+
+  def list_conversations(%Actor{id: actor_id}, %{page: page, limit: limit}, %{
+        context: %{
+          current_actor: %Actor{id: _current_actor_id}
+        }
+      }) do
+    {:ok,
+     actor_id
+     |> Conversations.list_conversation_participants_for_actor(page, limit)
+     |> conversation_participant_to_view()}
+  end
+
+  def list_conversations(%User{id: user_id}, %{page: page, limit: limit}, %{
+        context: %{
+          current_actor: %Actor{id: _current_actor_id}
+        }
+      }) do
+    {:ok,
+     user_id
+     |> Conversations.list_conversation_participants_for_user(page, limit)
+     |> conversation_participant_to_view()}
+  end
+
+  def unread_conversations_count(%Actor{id: actor_id}, _args, %{
+        context: %{
+          current_user: %User{} = user
+        }
+      }) do
+    case User.owns_actor(user, actor_id) do
+      {:is_owned, %Actor{}} ->
+        {:ok, Conversations.count_unread_conversation_participants_for_person(actor_id)}
+
+      _ ->
+        {:error, :unauthorized}
+    end
+  end
+
+  def get_conversation(_parent, %{id: conversation_participant_id}, %{
+        context: %{
+          current_actor: %Actor{id: performing_actor_id}
+        }
+      }) do
+    case Conversations.get_conversation_participant(conversation_participant_id) do
+      nil ->
+        {:error, :not_found}
+
+      %ConversationParticipant{actor_id: actor_id} = conversation_participant ->
+        if actor_id == performing_actor_id or Actors.is_member?(performing_actor_id, actor_id) do
+          {:ok, conversation_participant_to_view(conversation_participant)}
+        else
+          {:error, :not_found}
+        end
+    end
+  end
+
+  def get_comments_for_conversation(
+        %ConversationView{origin_comment_id: origin_comment_id, actor_id: conversation_actor_id},
+        %{page: page, limit: limit},
+        %{
+          context: %{
+            current_actor: %Actor{id: performing_actor_id}
+          }
+        }
+      ) do
+    if conversation_actor_id == performing_actor_id or
+         Actors.is_member?(performing_actor_id, conversation_actor_id) do
+      {:ok,
+       Mobilizon.Discussions.get_comments_in_reply_to_comment_id(origin_comment_id, page, limit)}
+    else
+      {:error, :unauthorized}
+    end
+  end
+
+  def create_conversation(
+        _parent,
+        %{actor_id: actor_id} = args,
+        %{
+          context: %{
+            current_actor: %Actor{} = current_actor
+          }
+        }
+      ) do
+    if authorized_to_reply?(
+         Map.get(args, :conversation_id),
+         Map.get(args, :attributed_to_id),
+         current_actor.id
+       ) do
+      case Comments.create_conversation(args) do
+        {:ok, _activity, %Conversation{} = conversation} ->
+          Absinthe.Subscription.publish(
+            Endpoint,
+            Conversations.count_unread_conversation_participants_for_person(current_actor.id),
+            person_unread_conversations_count: current_actor.id
+          )
+
+          conversation_participant_actor =
+            args |> Map.get(:attributed_to_id, actor_id) |> Actors.get_actor()
+
+          {:ok, conversation_to_view(conversation, conversation_participant_actor)}
+
+        {:error, :empty_participants} ->
+          {:error, dgettext("errors", "Conversation needs to mention at least one participant")}
+      end
+    else
+      {:error, :unauthorized}
+    end
+  end
+
+  def update_conversation(_parent, %{conversation_id: conversation_participant_id, read: read}, %{
+        context: %{
+          current_actor: %Actor{id: current_actor_id}
+        }
+      }) do
+    with {:no_participant,
+          %ConversationParticipant{actor_id: actor_id} = conversation_participant} <-
+           {:no_participant,
+            Conversations.get_conversation_participant(conversation_participant_id)},
+         {:valid_actor, true} <-
+           {:valid_actor,
+            actor_id == current_actor_id or
+              Actors.is_member?(current_actor_id, actor_id)},
+         {:ok, %ConversationParticipant{} = conversation_participant} <-
+           Conversations.update_conversation_participant(conversation_participant, %{
+             unread: !read
+           }) do
+      Absinthe.Subscription.publish(
+        Endpoint,
+        Conversations.count_unread_conversation_participants_for_person(actor_id),
+        person_unread_conversations_count: actor_id
+      )
+
+      {:ok, conversation_participant_to_view(conversation_participant)}
+    else
+      {:no_participant, _} ->
+        {:error, :not_found}
+
+      {:valid_actor, _} ->
+        {:error, :unauthorized}
+    end
+  end
+
+  def delete_conversation(_, _, _), do: :ok
+
+  defp conversation_participant_to_view(%Page{elements: elements} = page) do
+    %Page{page | elements: Enum.map(elements, &conversation_participant_to_view/1)}
+  end
+
+  defp conversation_participant_to_view(%ConversationParticipant{} = conversation_participant) do
+    value =
+      conversation_participant
+      |> Map.from_struct()
+      |> Map.merge(Map.from_struct(conversation_participant.conversation))
+      |> Map.delete(:conversation)
+      |> Map.put(
+        :participants,
+        Enum.map(
+          conversation_participant.conversation.participants,
+          &conversation_participant_to_actor/1
+        )
+      )
+      |> Map.put(:conversation_participant_id, conversation_participant.id)
+
+    struct(ConversationView, value)
+  end
+
+  defp conversation_to_view(
+         %Conversation{id: conversation_id} = conversation,
+         %Actor{id: actor_id} = actor,
+         unread \\ true
+       ) do
+    value =
+      conversation
+      |> Map.from_struct()
+      |> Map.put(:actor, actor)
+      |> Map.put(:unread, unread)
+      |> Map.put(
+        :conversation_participant_id,
+        Conversations.get_participant_by_conversation_and_actor(conversation_id, actor_id).id
+      )
+
+    struct(ConversationView, value)
+  end
+
+  defp conversation_participant_to_actor(%Actor{} = actor), do: actor
+
+  defp conversation_participant_to_actor(%ConversationParticipant{} = conversation_participant),
+    do: conversation_participant.actor
+
+  @spec authorized_to_reply?(String.t() | nil, String.t() | nil, String.t()) :: boolean()
+  # Not a reply
+  defp authorized_to_reply?(conversation_id, _attributed_to_id, _current_actor_id)
+       when is_nil(conversation_id),
+       do: true
+
+  # We are authorized to reply if we are one of the participants, or if we a a member of a participant group
+  defp authorized_to_reply?(conversation_id, attributed_to_id, current_actor_id) do
+    case Conversations.get_conversation(conversation_id) do
+      nil ->
+        false
+
+      %Conversation{participants: participants} ->
+        participant_ids = Enum.map(participants, fn participant -> to_string(participant.id) end)
+
+        current_actor_id in participant_ids or
+          Enum.any?(participant_ids, fn participant_id ->
+            Actors.is_member?(current_actor_id, participant_id) and
+              attributed_to_id == participant_id
+          end)
+    end
+  end
+end

lib/graphql/resolvers/participant.ex

@@ -2,9 +2,12 @@ defmodule Mobilizon.GraphQL.Resolvers.Participant do
   @moduledoc """
   Handles the participation-related GraphQL calls.
   """
+  # alias Mobilizon.Conversations.ConversationParticipant
   alias Mobilizon.{Actors, Config, Crypto, Events}
   alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.{Conversation, ConversationView}
   alias Mobilizon.Events.{Event, Participant}
+  alias Mobilizon.GraphQL.API.Comments
   alias Mobilizon.GraphQL.API.Participations
   alias Mobilizon.Service.Export.Participants.{CSV, ODS, PDF}
   alias Mobilizon.Users.User
@@ -346,6 +349,60 @@ defmodule Mobilizon.GraphQL.Resolvers.Participant do
 
   def export_event_participants(_, _, _), do: {:error, :unauthorized}
 
+  def send_private_messages_to_participants(
+        _parent,
+        %{roles: roles, event_id: event_id, actor_id: actor_id} =
+          args,
+        %{
+          context: %{
+            current_user: %User{locale: _locale},
+            current_actor: %Actor{id: current_actor_id}
+          }
+        }
+      ) do
+    participant_actors =
+      event_id
+      |> Events.list_all_participants_for_event(roles)
+      |> Enum.map(& &1.actor)
+
+    mentions =
+      participant_actors
+      |> Enum.map(& &1.id)
+      |> Enum.uniq()
+      |> Enum.map(&%{actor_id: &1, event_id: event_id})
+
+    args =
+      Map.merge(args, %{
+        mentions: mentions,
+        visibility: :private
+      })
+
+    with {:member, true} <-
+           {:member,
+            current_actor_id == actor_id or Actors.is_member?(current_actor_id, actor_id)},
+         {:ok, _activity, %Conversation{} = conversation} <- Comments.create_conversation(args) do
+      {:ok, conversation_to_view(conversation, Actors.get_actor(actor_id))}
+    else
+      {:member, false} ->
+        {:error, :unauthorized}
+
+      {:error, err} ->
+        {:error, err}
+    end
+  end
+
+  def send_private_messages_to_participants(_parent, _args, _resolution),
+    do: {:error, :unauthorized}
+
+  defp conversation_to_view(%Conversation{} = conversation, %Actor{} = actor) do
+    value =
+      conversation
+      |> Map.from_struct()
+      |> Map.put(:actor, actor)
+
+    struct(ConversationView, value)
+  end
+
   @spec valid_email?(String.t() | nil) :: boolean
   defp valid_email?(email) when is_nil(email), do: false
 

lib/graphql/schema.ex

@@ -55,6 +55,7 @@ defmodule Mobilizon.GraphQL.Schema do
   import_types(Schema.Users.ActivitySetting)
   import_types(Schema.FollowedGroupActivityType)
   import_types(Schema.AuthApplicationType)
+  import_types(Schema.ConversationType)
 
   @desc "A struct containing the id of the deleted object"
   object :deleted_object do
@@ -165,6 +166,7 @@ defmodule Mobilizon.GraphQL.Schema do
     import_fields(:todo_list_queries)
     import_fields(:todo_queries)
     import_fields(:discussion_queries)
+    import_fields(:conversation_queries)
     import_fields(:resource_queries)
     import_fields(:post_queries)
     import_fields(:statistics_queries)
@@ -189,6 +191,7 @@ defmodule Mobilizon.GraphQL.Schema do
     import_fields(:todo_list_mutations)
     import_fields(:todo_mutations)
     import_fields(:discussion_mutations)
+    import_fields(:conversation_mutations)
     import_fields(:resource_mutations)
     import_fields(:post_mutations)
     import_fields(:actor_mutations)
@@ -204,6 +207,7 @@ defmodule Mobilizon.GraphQL.Schema do
   subscription do
     import_fields(:person_subscriptions)
     import_fields(:discussion_subscriptions)
+    import_fields(:conversation_subscriptions)
   end
 
   @spec middleware(list(module()), any(), map()) :: list(module())

lib/graphql/schema/actors/person.ex

@@ -7,7 +7,7 @@ defmodule Mobilizon.GraphQL.Schema.Actors.PersonType do
   import Absinthe.Resolution.Helpers, only: [dataloader: 2]
 
   alias Mobilizon.Events
-  alias Mobilizon.GraphQL.Resolvers.{Media, Person}
+  alias Mobilizon.GraphQL.Resolvers.{Conversation, Media, Person}
   alias Mobilizon.GraphQL.Schema
 
   import_types(Schema.Events.FeedTokenType)
@@ -136,6 +136,25 @@ defmodule Mobilizon.GraphQL.Schema.Actors.PersonType do
       arg(:limit, :integer, default_value: 10, description: "The limit of follows per page")
       resolve(&Person.person_follows/3)
     end
+
+    @desc "The list of conversations this person has"
+    field(:conversations, :paginated_conversation_list,
+      meta: [private: true, rule: :"read:profile:conversations"]
+    ) do
+      arg(:page, :integer,
+        default_value: 1,
+        description: "The page in the conversations list"
+      )
+
+      arg(:limit, :integer, default_value: 10, description: "The limit of conversations per page")
+      resolve(&Conversation.list_conversations/3)
+    end
+
+    field(:unread_conversations_count, :integer,
+      meta: [private: true, rule: :"read:profile:conversations"]
+    ) do
+      resolve(&Conversation.unread_conversations_count/3)
+    end
   end
 
   @desc """
@@ -353,5 +372,16 @@ defmodule Mobilizon.GraphQL.Schema.Actors.PersonType do
         {:ok, topic: [args.group, args.person_id]}
       end)
     end
+
+    @desc "Notify when a person unread conversations count changed"
+    field(:person_unread_conversations_count, :integer,
+      meta: [private: true, rule: :"read:profile:conversations"]
+    ) do
+      arg(:person_id, non_null(:id), description: "The person's ID")
+
+      config(fn args, _ ->
+        {:ok, topic: [args.person_id]}
+      end)
+    end
   end
 end

lib/graphql/schema/conversation.ex

@@ -0,0 +1,132 @@
+defmodule Mobilizon.GraphQL.Schema.ConversationType do
+  @moduledoc """
+  Schema representation for conversation
+  """
+  use Absinthe.Schema.Notation
+
+  # import Absinthe.Resolution.Helpers, only: [dataloader: 1]
+
+  # alias Mobilizon.Actors
+  alias Mobilizon.GraphQL.Resolvers.Conversation
+
+  @desc "A conversation"
+  object :conversation do
+    meta(:authorize, :user)
+    interfaces([:activity_object])
+    field(:id, :id, description: "Internal ID for this conversation")
+
+    field(:conversation_participant_id, :id,
+      description: "Internal ID for the conversation participant"
+    )
+
+    field(:last_comment, :comment, description: "The last comment of the conversation")
+
+    field :comments, :paginated_comment_list do
+      arg(:page, :integer, default_value: 1)
+      arg(:limit, :integer, default_value: 10)
+      resolve(&Conversation.get_comments_for_conversation/3)
+      description("The comments for the conversation")
+    end
+
+    field(:participants, list_of(:person),
+      # resolve: dataloader(Actors),
+      description: "The list of participants to the conversation"
+    )
+
+    field(:event, :event, description: "The event this conversation is associated to")
+
+    field(:actor, :person,
+      # resolve: dataloader(Actors),
+      description: "The actor concerned by the conversation"
+    )
+
+    field(:unread, :boolean, description: "Whether this conversation is unread")
+
+    field(:inserted_at, :datetime, description: "When was this conversation's created")
+    field(:updated_at, :datetime, description: "When was this conversation's updated")
+  end
+
+  @desc "A paginated list of conversations"
+  object :paginated_conversation_list do
+    meta(:authorize, :user)
+    field(:elements, list_of(:conversation), description: "A list of conversations")
+    field(:total, :integer, description: "The total number of conversations in the list")
+  end
+
+  object :conversation_queries do
+    @desc "Get a conversation"
+    field :conversation, type: :conversation do
+      arg(:id, :id, description: "The conversation's ID")
+
+      middleware(Rajska.QueryAuthorization,
+        permit: :user,
+        scope: Mobilizon.Conversations.Conversation,
+        rule: :"read:conversations",
+        args: %{id: :id}
+      )
+
+      resolve(&Conversation.get_conversation/3)
+    end
+  end
+
+  object :conversation_mutations do
+    @desc "Post a private message"
+    field :post_private_message, type: :conversation do
+      arg(:text, non_null(:string), description: "The conversation's first comment body")
+      arg(:actor_id, non_null(:id), description: "The profile ID to create the conversation as")
+      arg(:attributed_to_id, :id, description: "The group ID to attribute the conversation to")
+      arg(:conversation_id, :id, description: "The conversation ID to reply to")
+      arg(:language, :string, description: "The comment language", default_value: "und")
+      arg(:mentions, list_of(:string), description: "A list of federated usernames to mention")
+
+      middleware(Rajska.QueryAuthorization,
+        permit: :user,
+        scope: Mobilizon.Conversations.ConversationParticipant,
+        rule: :"write:conversation:create",
+        args: %{actor_id: :actor_id}
+      )
+
+      resolve(&Conversation.create_conversation/3)
+    end
+
+    @desc "Update a conversation"
+    field :update_conversation, type: :conversation do
+      arg(:conversation_id, non_null(:id), description: "The conversation's ID")
+      arg(:read, non_null(:boolean), description: "Whether the conversation is read or not")
+
+      middleware(Rajska.QueryAuthorization,
+        permit: :user,
+        scope: Mobilizon.Conversations.Conversation,
+        rule: :"write:conversation:update",
+        args: %{id: :conversation_id}
+      )
+
+      resolve(&Conversation.update_conversation/3)
+    end
+
+    @desc "Delete a conversation"
+    field :delete_conversation, type: :conversation do
+      arg(:conversation_id, non_null(:id), description: "The conversation's ID")
+
+      middleware(Rajska.QueryAuthorization,
+        permit: :user,
+        scope: Mobilizon.Conversations.Conversation,
+        rule: :"write:conversation:delete",
+        args: %{id: :conversation_id}
+      )
+
+      resolve(&Conversation.delete_conversation/3)
+    end
+  end
+
+  object :conversation_subscriptions do
+    @desc "Notify when a conversation changed"
+    field :conversation_comment_changed, :conversation do
+      arg(:id, non_null(:id), description: "The conversation's ID")
+
+      config(fn args, _ ->
+        {:ok, topic: args.id}
+      end)
+    end
+  end
+end

lib/graphql/schema/discussions/comment.ex

@@ -56,6 +56,8 @@ defmodule Mobilizon.GraphQL.Schema.Discussions.CommentType do
       description: "Whether this comment needs to be announced to participants"
     )
 
+    field(:conversation, :conversation, description: "The conversation this comment is part of")
+
     field(:language, :string, description: "The comment language")
   end
 

lib/graphql/schema/event.ex

@@ -8,7 +8,7 @@ defmodule Mobilizon.GraphQL.Schema.EventType do
   import Absinthe.Resolution.Helpers, only: [dataloader: 1, dataloader: 2]
 
   alias Mobilizon.{Actors, Addresses, Discussions}
-  alias Mobilizon.GraphQL.Resolvers.{Event, Media, Tag}
+  alias Mobilizon.GraphQL.Resolvers.{Conversation, Event, Media, Tag}
   alias Mobilizon.GraphQL.Schema
 
   import_types(Schema.AddressType)
@@ -113,6 +113,18 @@ defmodule Mobilizon.GraphQL.Schema.EventType do
     field(:options, :event_options, description: "The event options")
     field(:metadata, list_of(:event_metadata), description: "A key-value list of metadata")
     field(:language, :string, description: "The event language")
+
+    field(:conversations, :paginated_conversation_list,
+      description: "The list of conversations started on this event"
+    ) do
+      arg(:page, :integer,
+        default_value: 1,
+        description: "The page in the paginated conversation list"
+      )
+
+      arg(:limit, :integer, default_value: 10, description: "The limit of conversations per page")
+      resolve(&Conversation.find_conversations_for_event/3)
+    end
   end
 
   @desc "The list of visibility options for an event"

lib/graphql/schema/events/participant.ex

@@ -159,5 +159,34 @@ defmodule Mobilizon.GraphQL.Schema.Events.ParticipantType do
 
       resolve(&Participant.export_event_participants/3)
     end
+
+    @desc "Send private messages to participants"
+    field :send_event_private_message, :conversation do
+      arg(:event_id, non_null(:id),
+        description: "The ID from the event for which to export participants"
+      )
+
+      arg(:roles, list_of(:participant_role_enum),
+        default_value: [],
+        description: "The participant roles to include"
+      )
+
+      arg(:text, non_null(:string), description: "The private message body")
+
+      arg(:actor_id, non_null(:id),
+        description: "The profile ID to create the private message as"
+      )
+
+      arg(:language, :string, description: "The private message language", default_value: "und")
+
+      middleware(Rajska.QueryAuthorization,
+        permit: :user,
+        scope: Mobilizon.Events.Event,
+        rule: :"write:event:participants:private_message",
+        args: %{id: :event_id}
+      )
+
+      resolve(&Participant.send_private_messages_to_participants/3)
+    end
   end
 end

lib/graphql/schema/user.ex

@@ -8,7 +8,7 @@ defmodule Mobilizon.GraphQL.Schema.UserType do
 
   alias Mobilizon.Events
   alias Mobilizon.GraphQL.Resolvers.Application, as: ApplicationResolver
-  alias Mobilizon.GraphQL.Resolvers.{Media, User}
+  alias Mobilizon.GraphQL.Resolvers.{Conversation, Media, User}
   alias Mobilizon.GraphQL.Resolvers.Users.ActivitySettings
   alias Mobilizon.GraphQL.Schema
 
@@ -191,6 +191,19 @@ defmodule Mobilizon.GraphQL.Schema.UserType do
     ) do
       resolve(&ApplicationResolver.get_user_applications/3)
     end
+
+    @desc "The list of conversations this person has"
+    field(:conversations, :paginated_conversation_list,
+      meta: [private: true, rule: :"read:profile:conversations"]
+    ) do
+      arg(:page, :integer,
+        default_value: 1,
+        description: "The page in the conversations list"
+      )
+
+      arg(:limit, :integer, default_value: 10, description: "The limit of conversations per page")
+      resolve(&Conversation.list_conversations/3)
+    end
   end
 
   @desc "The list of roles an user can have"

lib/mobilizon/activities/activities.ex

@@ -17,10 +17,24 @@ defmodule Mobilizon.Activities do
     very_high: 50
   )
 
-  @activity_types ["event", "post", "discussion", "resource", "group", "member", "comment"]
+  @activity_types [
+    "event",
+    "post",
+    "conversation",
+    "discussion",
+    "resource",
+    "group",
+    "member",
+    "comment"
+  ]
   @event_activity_subjects ["event_created", "event_updated", "event_deleted", "comment_posted"]
   @participant_activity_subjects ["event_new_participation"]
   @post_activity_subjects ["post_created", "post_updated", "post_deleted"]
+  @conversation_activity_subjects [
+    "conversation_created",
+    "conversation_replied",
+    "conversation_event_announcement"
+  ]
   @discussion_activity_subjects [
     "discussion_created",
     "discussion_replied",
@@ -49,6 +63,7 @@ defmodule Mobilizon.Activities do
   @settings_activity_subjects ["group_created", "group_updated"]
 
   @subjects @event_activity_subjects ++
+              @conversation_activity_subjects ++
               @participant_activity_subjects ++
               @post_activity_subjects ++
               @discussion_activity_subjects ++
@@ -61,6 +76,7 @@ defmodule Mobilizon.Activities do
     "actor",
     "post",
     "discussion",
+    "conversation",
     "resource",
     "member",
     "group",

lib/mobilizon/actors/actor.ex

@@ -10,6 +10,7 @@ defmodule Mobilizon.Actors.Actor do
   alias Mobilizon.{Actors, Addresses, Config, Crypto, Mention, Share}
   alias Mobilizon.Actors.{ActorOpenness, ActorType, ActorVisibility, Follower, Member}
   alias Mobilizon.Addresses.Address
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.Comment
   alias Mobilizon.Events.{Event, FeedToken, Participant}
   alias Mobilizon.Medias.File
@@ -196,6 +197,11 @@ defmodule Mobilizon.Actors.Actor do
     has_many(:owner_shares, Share, foreign_key: :owner_actor_id)
     many_to_many(:memberships, __MODULE__, join_through: Member)
 
+    many_to_many(:conversations, Conversation,
+      join_through: "conversation_participants",
+      join_keys: [conversation_id: :id, participant_id: :id]
+    )
+
     timestamps()
   end
 

lib/mobilizon/conversations/conversation.ex

@@ -0,0 +1,57 @@
+defmodule Mobilizon.Conversations.Conversation do
+  @moduledoc """
+  Represents a conversation
+  """
+
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  alias Ecto.Changeset
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.ConversationParticipant
+  alias Mobilizon.Discussions.Comment
+  alias Mobilizon.Events.Event
+
+  @type t :: %__MODULE__{
+          id: String.t(),
+          origin_comment: Comment.t(),
+          last_comment: Comment.t(),
+          participants: list(Actor.t())
+        }
+
+  @required_attrs [:origin_comment_id, :last_comment_id]
+  @optional_attrs [:event_id]
+  @attrs @required_attrs ++ @optional_attrs
+
+  schema "conversations" do
+    belongs_to(:origin_comment, Comment)
+    belongs_to(:last_comment, Comment)
+    belongs_to(:event, Event)
+    has_many(:comments, Comment)
+
+    many_to_many(:participants, Actor,
+      join_through: ConversationParticipant,
+      join_keys: [conversation_id: :id, actor_id: :id],
+      on_replace: :delete
+    )
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @doc false
+  @spec changeset(t | Ecto.Schema.t(), map) :: Ecto.Changeset.t()
+  def changeset(%__MODULE__{} = conversation, attrs) do
+    conversation
+    |> cast(attrs, @attrs)
+    |> maybe_set_participants(attrs)
+    |> validate_required(@required_attrs)
+  end
+
+  defp maybe_set_participants(%Changeset{} = changeset, %{participants: participants})
+       when length(participants) > 0 do
+    put_assoc(changeset, :participants, participants)
+  end
+
+  defp maybe_set_participants(%Changeset{} = changeset, _), do: changeset
+end

lib/mobilizon/conversations/conversation_participant.ex

@@ -0,0 +1,40 @@
+defmodule Mobilizon.Conversations.ConversationParticipant do
+  @moduledoc """
+  Represents a conversation participant
+  """
+
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.Conversation
+
+  @type t :: %__MODULE__{
+          conversation: Conversation.t(),
+          actor: Actor.t(),
+          unread: boolean()
+        }
+
+  @required_attrs [:actor_id, :conversation_id]
+  @optional_attrs [:unread]
+  @attrs @required_attrs ++ @optional_attrs
+
+  schema "conversation_participants" do
+    belongs_to(:conversation, Conversation)
+    belongs_to(:actor, Actor)
+    field(:unread, :boolean, default: true)
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @doc false
+  @spec changeset(t | Ecto.Schema.t(), map) :: Ecto.Changeset.t()
+  def changeset(%__MODULE__{} = conversation, attrs) do
+    conversation
+    |> cast(attrs, @attrs)
+    |> validate_required(@required_attrs)
+    |> foreign_key_constraint(:conversation_id)
+    |> foreign_key_constraint(:actor_id)
+  end
+end

lib/mobilizon/conversations/conversation_view.ex

@@ -0,0 +1,22 @@
+defmodule Mobilizon.Conversations.ConversationView do
+  @moduledoc """
+  Represents a conversation view for GraphQL API
+  """
+
+  defstruct [
+    :id,
+    :conversation_participant_id,
+    :origin_comment,
+    :origin_comment_id,
+    :last_comment,
+    :last_comment_id,
+    :event,
+    :event_id,
+    :actor,
+    :actor_id,
+    :unread,
+    :inserted_at,
+    :updated_at,
+    :participants
+  ]
+end

lib/mobilizon/conversations/conversations.ex

@@ -0,0 +1,344 @@
+defmodule Mobilizon.Conversations do
+  @moduledoc """
+  The conversations context
+  """
+
+  import Ecto.Query
+
+  alias Ecto.Changeset
+  alias Ecto.Multi
+  alias Mobilizon.Actors.{Actor, Member}
+  alias Mobilizon.Conversations.{Conversation, ConversationParticipant}
+  alias Mobilizon.Discussions.Comment
+  alias Mobilizon.Events.Event
+  alias Mobilizon.Storage.{Page, Repo}
+
+  @conversation_preloads [
+    :origin_comment,
+    :last_comment,
+    :event,
+    :participants
+  ]
+
+  @comment_preloads [
+    :actor,
+    :event,
+    :attributed_to,
+    :in_reply_to_comment,
+    :origin_comment,
+    :replies,
+    :tags,
+    :mentions,
+    :media
+  ]
+
+  @doc """
+  Get a conversation by it's ID
+  """
+  @spec get_conversation(String.t() | integer()) :: Conversation.t() | nil
+  def get_conversation(conversation_id) do
+    Conversation
+    |> Repo.get(conversation_id)
+    |> Repo.preload(@conversation_preloads)
+  end
+
+  @doc """
+  Get a conversation by it's ID
+  """
+  @spec get_conversation_participant(String.t() | integer()) :: Conversation.t() | nil
+  def get_conversation_participant(conversation_participant_id) do
+    preload_conversation_participant_details()
+    |> where([cp], cp.id == ^conversation_participant_id)
+    |> Repo.one()
+  end
+
+  def get_participant_by_conversation_and_actor(conversation_id, actor_id) do
+    preload_conversation_participant_details()
+    |> where([cp], cp.conversation_id == ^conversation_id and cp.actor_id == ^actor_id)
+    |> Repo.one()
+  end
+
+  defp preload_conversation_participant_details do
+    ConversationParticipant
+    |> join(:inner, [cp], c in Conversation, on: cp.conversation_id == c.id)
+    |> join(:left, [_cp, c], e in Event, on: c.event_id == e.id)
+    |> join(:inner, [cp], a in Actor, on: cp.actor_id == a.id)
+    |> join(:inner, [_cp, c], lc in Comment, on: c.last_comment_id == lc.id)
+    |> join(:inner, [_cp, c], oc in Comment, on: c.origin_comment_id == oc.id)
+    |> join(:inner, [_cp, c], p in ConversationParticipant, on: c.id == p.conversation_id)
+    |> join(:inner, [_cp, _c, _e, _a, _lc, _oc, p], ap in Actor, on: p.actor_id == ap.id)
+    |> preload([_cp, c, e, a, lc, oc, p, ap],
+      actor: a,
+      conversation:
+        {c, event: e, last_comment: lc, origin_comment: oc, participants: {p, actor: ap}}
+    )
+  end
+
+  @doc """
+  Get a paginated list of conversations for an actor
+  """
+  @spec find_conversations_for_actor(Actor.t(), integer | nil, integer | nil) ::
+          Page.t(Conversation.t())
+  def find_conversations_for_actor(%Actor{id: actor_id}, page \\ nil, limit \\ nil) do
+    Conversation
+    |> where([c], c.actor_id == ^actor_id)
+    |> preload(^@conversation_preloads)
+    |> order_by(desc: :updated_at)
+    |> Page.build_page(page, limit)
+  end
+
+  @spec find_conversations_for_event(
+          String.t() | integer,
+          String.t() | integer,
+          integer | nil,
+          integer | nil
+        ) :: Page.t(ConversationParticipant.t())
+  def find_conversations_for_event(event_id, actor_id, page \\ nil, limit \\ nil) do
+    ConversationParticipant
+    |> join(:inner, [cp], c in Conversation, on: cp.conversation_id == c.id)
+    |> join(:left, [_cp, c], e in Event, on: c.event_id == e.id)
+    |> join(:inner, [cp], a in Actor, on: cp.actor_id == a.id)
+    |> join(:inner, [_cp, c], lc in Comment, on: c.last_comment_id == lc.id)
+    |> join(:inner, [_cp, c], oc in Comment, on: c.origin_comment_id == oc.id)
+    |> join(:inner, [_cp, c], p in ConversationParticipant, on: c.id == p.conversation_id)
+    |> join(:inner, [_cp, _c, _e, _a, _lc, _oc, p], ap in Actor, on: p.actor_id == ap.id)
+    |> where([_cp, c], c.event_id == ^event_id)
+    |> where([cp], cp.actor_id == ^actor_id)
+    |> preload([_cp, c, e, a, lc, oc, p, ap],
+      actor: a,
+      conversation:
+        {c, event: e, last_comment: lc, origin_comment: oc, participants: {p, actor: ap}}
+    )
+    |> Page.build_page(page, limit)
+  end
+
+  @spec list_conversation_participants_for_actor(
+          integer | String.t(),
+          integer | nil,
+          integer | nil
+        ) ::
+          Page.t(ConversationParticipant.t())
+  def list_conversation_participants_for_actor(actor_id, page \\ nil, limit \\ nil) do
+    subquery =
+      ConversationParticipant
+      |> distinct([cp], cp.conversation_id)
+      |> join(:left, [cp], m in Member, on: cp.actor_id == m.parent_id)
+      |> where([cp], cp.actor_id == ^actor_id)
+      |> or_where(
+        [_cp, m],
+        m.actor_id == ^actor_id and m.role in [:creator, :administrator, :moderator]
+      )
+
+    subquery
+    |> subquery()
+    |> order_by([cp], desc: cp.unread, desc: cp.updated_at)
+    |> preload([:actor, conversation: [:last_comment, :participants]])
+    |> Page.build_page(page, limit)
+  end
+
+  @spec list_conversation_participants_for_user(
+          integer | String.t(),
+          integer | nil,
+          integer | nil
+        ) ::
+          Page.t(ConversationParticipant.t())
+  def list_conversation_participants_for_user(user_id, page \\ nil, limit \\ nil) do
+    ConversationParticipant
+    |> join(:inner, [cp], a in Actor, on: cp.actor_id == a.id)
+    |> where([_cp, a], a.user_id == ^user_id)
+    |> preload([:actor, conversation: [:last_comment, :participants]])
+    |> Page.build_page(page, limit)
+  end
+
+  @spec list_conversation_participants_for_conversation(integer | String.t()) ::
+          list(ConversationParticipant.t())
+  def list_conversation_participants_for_conversation(conversation_id) do
+    ConversationParticipant
+    |> where([cp], cp.conversation_id == ^conversation_id)
+    |> Repo.all()
+  end
+
+  @spec count_unread_conversation_participants_for_person(integer | String.t()) ::
+          non_neg_integer()
+  def count_unread_conversation_participants_for_person(actor_id) do
+    ConversationParticipant
+    |> where([cp], cp.actor_id == ^actor_id and cp.unread == true)
+    |> Repo.aggregate(:count)
+  end
+
+  @doc """
+  Creates a conversation.
+  """
+  @spec create_conversation(map()) ::
+          {:ok, Conversation.t()} | {:error, atom(), Changeset.t(), map()}
+  def create_conversation(attrs) do
+    with {:ok, %{comment: %Comment{} = _comment, conversation: %Conversation{} = conversation}} <-
+           Multi.new()
+           |> Multi.insert(
+             :comment,
+             Comment.changeset(
+               %Comment{},
+               Map.merge(attrs, %{
+                 actor_id: attrs.actor_id,
+                 attributed_to_id: attrs.actor_id,
+                 visibility: :private
+               })
+             )
+           )
+           |> Multi.insert(:conversation, fn %{
+                                               comment: %Comment{
+                                                 id: comment_id,
+                                                 origin_comment_id: origin_comment_id
+                                               }
+                                             } ->
+             Conversation.changeset(
+               %Conversation{},
+               Map.merge(attrs, %{
+                 last_comment_id: comment_id,
+                 origin_comment_id: origin_comment_id || comment_id,
+                 participants: attrs.participants
+               })
+             )
+           end)
+           |> Multi.update(:update_comment, fn %{
+                                                 comment: %Comment{} = comment,
+                                                 conversation: %Conversation{id: conversation_id}
+                                               } ->
+             Comment.changeset(
+               comment,
+               %{conversation_id: conversation_id}
+             )
+           end)
+           |> Multi.update_all(
+             :conversation_participants,
+             fn %{
+                  conversation: %Conversation{
+                    id: conversation_id
+                  }
+                } ->
+               ConversationParticipant
+               |> where(
+                 [cp],
+                 cp.conversation_id == ^conversation_id and cp.actor_id == ^attrs.actor_id
+               )
+               |> update([cp], set: [unread: false, updated_at: ^NaiveDateTime.utc_now()])
+             end,
+             []
+           )
+           |> Repo.transaction(),
+         %Conversation{} = conversation <- Repo.preload(conversation, @conversation_preloads) do
+      {:ok, conversation}
+    end
+  end
+
+  @doc """
+  Create a response to a conversation
+  """
+  @spec reply_to_conversation(Conversation.t(), map()) ::
+          {:ok, Conversation.t()} | {:error, atom(), Ecto.Changeset.t(), map()}
+  def reply_to_conversation(%Conversation{id: conversation_id} = conversation, attrs \\ %{}) do
+    attrs =
+      Map.merge(attrs, %{
+        conversation_id: conversation_id,
+        actor_id: Map.get(attrs, :creator_id, Map.get(attrs, :actor_id)),
+        origin_comment_id: conversation.origin_comment_id,
+        in_reply_to_comment_id: conversation.last_comment_id,
+        visibility: :private
+      })
+
+    changeset =
+      Comment.changeset(
+        %Comment{},
+        attrs
+      )
+
+    with {:ok, %{comment: %Comment{} = comment, conversation: %Conversation{} = conversation}} <-
+           Multi.new()
+           |> Multi.insert(
+             :comment,
+             changeset
+           )
+           |> Multi.update(:conversation, fn %{comment: %Comment{id: comment_id}} ->
+             Conversation.changeset(
+               conversation,
+               %{last_comment_id: comment_id}
+             )
+           end)
+           |> Multi.update_all(
+             :conversation_participants,
+             fn %{
+                  conversation: %Conversation{
+                    id: conversation_id
+                  }
+                } ->
+               ConversationParticipant
+               |> where(
+                 [cp],
+                 cp.conversation_id == ^conversation_id and cp.actor_id != ^attrs.actor_id
+               )
+               |> update([cp], set: [unread: true, updated_at: ^NaiveDateTime.utc_now()])
+             end,
+             []
+           )
+           |> Multi.update_all(
+             :conversation_participants_author,
+             fn %{
+                  conversation: %Conversation{
+                    id: conversation_id
+                  }
+                } ->
+               ConversationParticipant
+               |> where(
+                 [cp],
+                 cp.conversation_id == ^conversation_id and cp.actor_id == ^attrs.actor_id
+               )
+               |> update([cp], set: [unread: false, updated_at: ^NaiveDateTime.utc_now()])
+             end,
+             []
+           )
+           |> Repo.transaction(),
+         # Conversation is not updated
+         %Comment{} = comment <- Repo.preload(comment, @comment_preloads) do
+      {:ok, %Conversation{conversation | last_comment: comment}}
+    end
+  end
+
+  @doc """
+  Update a conversation.
+  """
+  @spec update_conversation(Conversation.t(), map()) ::
+          {:ok, Conversation.t()} | {:error, Changeset.t()}
+  def update_conversation(%Conversation{} = conversation, attrs \\ %{}) do
+    conversation
+    |> Conversation.changeset(attrs)
+    |> Repo.update()
+  end
+
+  @doc """
+  Delete a conversation.
+  """
+  @spec delete_conversation(Conversation.t()) ::
+          {:ok, %{comments: {integer() | nil, any()}}} | {:error, :comments, Changeset.t(), map()}
+  def delete_conversation(%Conversation{id: conversation_id}) do
+    Multi.new()
+    |> Multi.delete_all(:comments, fn _ ->
+      where(Comment, [c], c.conversation_id == ^conversation_id)
+    end)
+    # |> Multi.delete(:conversation, conversation)
+    |> Repo.transaction()
+  end
+
+  @doc """
+  Update a conversation participant. Only their read status for now
+  """
+  @spec update_conversation_participant(ConversationParticipant.t(), map()) ::
+          {:ok, ConversationParticipant.t()} | {:error, Changeset.t()}
+  def update_conversation_participant(
+        %ConversationParticipant{} = conversation_participant,
+        attrs \\ %{}
+      ) do
+    conversation_participant
+    |> ConversationParticipant.changeset(attrs)
+    |> Repo.update()
+  end
+end

lib/mobilizon/discussions/comment.ex

@@ -9,6 +9,7 @@ defmodule Mobilizon.Discussions.Comment do
   import Mobilizon.Storage.Ecto, only: [maybe_add_published_at: 1]
 
   alias Mobilizon.Actors.Actor
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.{Comment, CommentVisibility, Discussion}
   alias Mobilizon.Events.{Event, Tag}
   alias Mobilizon.Medias.Media
@@ -49,7 +50,9 @@ defmodule Mobilizon.Discussions.Comment do
     :local,
     :is_announcement,
     :discussion_id,
-    :language
+    :conversation_id,
+    :language,
+    :visibility
   ]
   @attrs @required_attrs ++ @optional_attrs
 
@@ -71,6 +74,7 @@ defmodule Mobilizon.Discussions.Comment do
     belongs_to(:in_reply_to_comment, Comment, foreign_key: :in_reply_to_comment_id)
     belongs_to(:origin_comment, Comment, foreign_key: :origin_comment_id)
     belongs_to(:discussion, Discussion, type: :binary_id)
+    belongs_to(:conversation, Conversation)
     has_many(:replies, Comment, foreign_key: :origin_comment_id)
     many_to_many(:tags, Tag, join_through: "comments_tags", on_replace: :delete)
     has_many(:mentions, Mention)
@@ -80,7 +84,7 @@ defmodule Mobilizon.Discussions.Comment do
   end
 
   @doc """
-  Returns the id of the first comment in the discussion.
+  Returns the id of the first comment in the discussion or conversation.
   """
   @spec get_thread_id(t) :: integer
   def get_thread_id(%__MODULE__{id: id, origin_comment_id: origin_comment_id}) do
@@ -181,7 +185,7 @@ defmodule Mobilizon.Discussions.Comment do
     Tag.changeset(%Tag{}, tag)
   end
 
-  defp process_mention(tag) do
-    Mention.changeset(%Mention{}, tag)
+  defp process_mention(mention) do
+    Mention.changeset(%Mention{}, mention)
   end
 end

lib/mobilizon/discussions/discussions.ex

@@ -42,9 +42,9 @@ defmodule Mobilizon.Discussions do
     :origin_comment,
     :replies,
     :tags,
-    :mentions,
     :discussion,
-    :media
+    :media,
+    mentions: [:actor]
   ]
 
   @discussion_preloads [
@@ -76,6 +76,7 @@ defmodule Mobilizon.Discussions do
     Comment
     |> join(:left, [c], r in Comment, on: r.origin_comment_id == c.id)
     |> where([c, _], is_nil(c.in_reply_to_comment_id))
+    |> where([c], c.visibility in ^@public_visibility)
     # TODO: This was added because we don't want to count deleted comments in total_replies.
     # However, it also excludes all top-level comments with deleted replies from being selected
     # |> where([_, r], is_nil(r.deleted_at))
@@ -197,9 +198,13 @@ defmodule Mobilizon.Discussions do
   """
   @spec update_comment(Comment.t(), map) :: {:ok, Comment.t()} | {:error, Changeset.t()}
   def update_comment(%Comment{} = comment, attrs) do
-    comment
-    |> Comment.update_changeset(attrs)
-    |> Repo.update()
+    with {:ok, %Comment{} = comment} <-
+           comment
+           |> Comment.update_changeset(attrs)
+           |> Repo.update(),
+         %Comment{} = comment <- Repo.preload(comment, @comment_preloads) do
+      {:ok, comment}
+    end
   end
 
   @doc """
@@ -272,6 +277,19 @@ defmodule Mobilizon.Discussions do
     |> Page.build_page(page, limit)
   end
 
+  @doc """
+  Get all the comments contained into a discussion
+  """
+  @spec get_comments_in_reply_to_comment_id(integer, integer | nil, integer | nil) ::
+          Page.t(Comment.t())
+  def get_comments_in_reply_to_comment_id(origin_comment_id, page \\ nil, limit \\ nil) do
+    Comment
+    |> where([c], c.id == ^origin_comment_id)
+    |> or_where([c], c.origin_comment_id == ^origin_comment_id)
+    |> order_by(asc: :inserted_at)
+    |> Page.build_page(page, limit)
+  end
+
   @doc """
   Counts local comments under events
   """

lib/mobilizon/events/event.ex

@@ -13,6 +13,7 @@ defmodule Mobilizon.Events.Event do
   alias Mobilizon.{Addresses, Events, Medias, Mention}
   alias Mobilizon.Addresses.Address
 
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.Comment
 
   alias Mobilizon.Events.{
@@ -126,6 +127,7 @@ defmodule Mobilizon.Events.Event do
     has_many(:sessions, Session)
     has_many(:mentions, Mention)
     has_many(:comments, Comment)
+    has_many(:conversations, Conversation)
     many_to_many(:contacts, Actor, join_through: "event_contacts", on_replace: :delete)
     many_to_many(:tags, Tag, join_through: "events_tags", on_replace: :delete)
     many_to_many(:participants, Actor, join_through: Participant)

lib/mobilizon/events/events.ex

@@ -871,6 +871,21 @@ defmodule Mobilizon.Events do
     |> Page.build_page(page, limit)
   end
 
+  @doc """
+  Returns the whole list of participants for an event.
+  Default behaviour is to not return :not_approved or :not_confirmed participants
+  """
+  @spec list_all_participants_for_event(String.t(), list(atom())) :: list(Participant.t())
+  def list_all_participants_for_event(
+        id,
+        roles \\ []
+      ) do
+    id
+    |> participants_for_event_query(roles)
+    |> preload([:actor, :event])
+    |> Repo.all()
+  end
+
   @spec list_actors_participants_for_event(String.t()) :: [Actor.t()]
   def list_actors_participants_for_event(id) do
     id

lib/mobilizon/mentions/mention.ex

@@ -32,8 +32,8 @@ defmodule Mobilizon.Mention do
 
   @doc false
   @spec changeset(t | Ecto.Schema.t(), map) :: Ecto.Changeset.t()
-  def changeset(event, attrs) do
-    event
+  def changeset(mention, attrs) do
+    mention
     |> cast(attrs, @attrs)
     # TODO: Enforce having either event_id or comment_id
     |> validate_required(@required_attrs)

lib/mobilizon/reports/reports.ex

@@ -21,7 +21,14 @@ defmodule Mobilizon.Reports do
   def get_report(id) do
     Report
     |> Repo.get(id)
-    |> Repo.preload([:reported, :reporter, :manager, :events, :comments, :notes])
+    |> Repo.preload([
+      :reported,
+      :reporter,
+      :manager,
+      :events,
+      :notes,
+      comments: [conversation: [:participants]]
+    ])
   end
 
   @doc """

lib/service/activity/conversation.ex

@@ -0,0 +1,90 @@
+defmodule Mobilizon.Service.Activity.Conversation do
+  @moduledoc """
+  Insert a conversation activity
+  """
+  alias Mobilizon.Conversations
+  alias Mobilizon.Conversations.{Conversation, ConversationParticipant}
+  alias Mobilizon.Discussions.Comment
+  alias Mobilizon.Events.Event
+  alias Mobilizon.Service.Activity
+  alias Mobilizon.Service.Workers.LegacyNotifierBuilder
+
+  @behaviour Activity
+
+  @impl Activity
+  def insert_activity(conversation, options \\ [])
+
+  def insert_activity(
+        %Conversation{} = conversation,
+        options
+      ) do
+    subject = Keyword.fetch!(options, :subject)
+
+    send_participant_notifications(subject, conversation, conversation.last_comment, options)
+  end
+
+  def insert_activity(_, _), do: {:ok, nil}
+
+  @impl Activity
+  def get_object(conversation_id) do
+    Conversations.get_conversation(conversation_id)
+  end
+
+  # An actor is mentionned
+  @spec send_participant_notifications(String.t(), Discussion.t(), Comment.t(), Keyword.t()) ::
+          {:ok, Oban.Job.t()} | {:ok, :skipped}
+  defp send_participant_notifications(
+         subject,
+         %Conversation{
+           id: conversation_id
+         } = conversation,
+         %Comment{actor_id: actor_id},
+         _options
+       )
+       when subject in [
+              "conversation_created",
+              "conversation_replied",
+              "conversation_event_announcement"
+            ] do
+    # We need to send each notification individually as the conversation URL varies for each participant
+
+    conversation_id
+    |> Conversations.list_conversation_participants_for_conversation()
+    |> Enum.each(fn %ConversationParticipant{id: conversation_participant_id} =
+                      conversation_participant ->
+      LegacyNotifierBuilder.enqueue(
+        :legacy_notify,
+        %{
+          "subject" => subject,
+          "subject_params" =>
+            Map.merge(
+              %{
+                conversation_id: conversation_id,
+                conversation_participant_id: conversation_participant_id
+              },
+              event_subject_params(conversation)
+            ),
+          "type" => :conversation,
+          "object_type" => :conversation,
+          "author_id" => actor_id,
+          "object_id" => to_string(conversation_id),
+          "participant" => Map.take(conversation_participant, [:id, :actor_id])
+        }
+      )
+    end)
+
+    {:ok, :enqueued}
+  end
+
+  defp send_participant_notifications(_, _, _, _), do: {:ok, :skipped}
+
+  defp event_subject_params(%Conversation{
+         event: %Event{id: conversation_event_id, title: conversation_event_title}
+       }),
+       do: %{
+         conversation_event_id: conversation_event_id,
+         conversation_event_title: conversation_event_title
+       }
+
+  defp event_subject_params(_), do: %{}
+end

lib/service/activity/renderer/conversation.ex

@@ -0,0 +1,73 @@
+defmodule Mobilizon.Service.Activity.Renderer.Conversation do
+  @moduledoc """
+  Render a conversation activity
+  """
+  alias Mobilizon.Activities.Activity
+  alias Mobilizon.Actors.Actor
+  alias Mobilizon.Service.Activity.Renderer
+  alias Mobilizon.Web.Endpoint
+  alias Mobilizon.Web.Router.Helpers, as: Routes
+  import Mobilizon.Web.Gettext, only: [dgettext: 3]
+
+  @behaviour Renderer
+
+  @impl Renderer
+  def render(%Activity{} = activity, options) do
+    locale = Keyword.get(options, :locale, "en")
+    Gettext.put_locale(locale)
+    profile = profile(activity)
+
+    case activity.subject do
+      :conversation_created ->
+        %{
+          body:
+            dgettext(
+              "activity",
+              "%{profile} sent you a message",
+              %{
+                profile: profile
+              }
+            ),
+          url: conversation_url(activity)
+        }
+
+      :conversation_replied ->
+        %{
+          body:
+            dgettext(
+              "activity",
+              "%{profile} replied to your message",
+              %{
+                profile: profile
+              }
+            ),
+          url: conversation_url(activity)
+        }
+
+      :conversation_event_announcement ->
+        %{
+          body:
+            dgettext(
+              "activity",
+              "%{profile} sent a private message about event %{event}",
+              %{
+                profile: profile,
+                event: event_title(activity)
+              }
+            ),
+          url: conversation_url(activity)
+        }
+    end
+  end
+
+  defp conversation_url(activity) do
+    Routes.page_url(
+      Endpoint,
+      :conversation,
+      activity.subject_params["conversation_id"]
+    )
+  end
+
+  defp profile(activity), do: Actor.display_name_and_username(activity.author)
+  defp event_title(activity), do: activity.subject_params["conversation_event_title"]
+end

lib/service/activity/renderer/renderer.ex

@@ -51,17 +51,25 @@ defmodule Mobilizon.Service.Activity.Renderer do
     res
   end
 
+  @types_map %{
+    discussion: Discussion,
+    conversation: Conversation,
+    event: Event,
+    group: Group,
+    member: Member,
+    post: Post,
+    resource: Resource,
+    comment: Comment
+  }
+
   @spec do_render(Activity.t(), Keyword.t()) :: common_render()
   defp do_render(%Activity{type: type} = activity, options) do
-    case type do
-      :discussion -> Discussion.render(activity, options)
-      :event -> Event.render(activity, options)
-      :group -> Group.render(activity, options)
-      :member -> Member.render(activity, options)
-      :post -> Post.render(activity, options)
-      :resource -> Resource.render(activity, options)
-      :comment -> Comment.render(activity, options)
-      _ -> nil
+    case Map.get(@types_map, type) do
+      nil ->
+        nil
+
+      mod ->
+        mod.render(activity, options)
     end
   end
 end

lib/service/notifier/email.ex

@@ -70,6 +70,9 @@ defmodule Mobilizon.Service.Notifier.Email do
   @always_direct_subjects [
     :participation_event_comment,
     :event_comment_mention,
+    :conversation_mention,
+    :conversation_created,
+    :conversation_replied,
     :discussion_mention,
     :event_new_comment
   ]
@@ -175,6 +178,9 @@ defmodule Mobilizon.Service.Notifier.Email do
     "member_updated" => false,
     "user_email_password_updated" => true,
     "event_comment_mention" => true,
+    "conversation_mention" => true,
+    "conversation_created" => true,
+    "conversation_replied" => true,
     "discussion_mention" => true,
     "event_new_comment" => true
   }

lib/service/notifier/filter.ex

@@ -33,6 +33,10 @@ defmodule Mobilizon.Service.Notifier.Filter do
   defp map_activity_to_activity_setting(%Activity{subject: :event_comment_mention}),
     do: "event_comment_mention"
 
+  defp map_activity_to_activity_setting(%Activity{subject: subject})
+       when subject in [:conversation_mention, :conversation_created, :conversation_replied],
+       do: to_string(subject)
+
   defp map_activity_to_activity_setting(%Activity{subject: :discussion_mention}),
     do: "discussion_mention"
 

lib/service/notifier/push.ex

@@ -64,6 +64,7 @@ defmodule Mobilizon.Service.Notifier.Push do
     "member_updated" => false,
     "user_email_password_updated" => false,
     "event_comment_mention" => true,
+    "conversation_mention" => true,
     "discussion_mention" => false,
     "event_new_comment" => false
   }

lib/service/workers/legacy_notifier_builder.ex

@@ -8,6 +8,7 @@ defmodule Mobilizon.Service.Workers.LegacyNotifierBuilder do
   alias Mobilizon.Actors.Actor
   alias Mobilizon.Events.{Event, Participant}
   alias Mobilizon.Service.Notifier
+  require Logger
 
   use Mobilizon.Service.Workers.Helper, queue: "activity"
 
@@ -15,6 +16,7 @@ defmodule Mobilizon.Service.Workers.LegacyNotifierBuilder do
   def perform(%Job{args: args}) do
     {"legacy_notify", args} = Map.pop(args, "op")
     activity = build_activity(args)
+    Logger.debug("Handling activity #{activity.subject} to notify in LegacyNotifierBuilder")
 
     if args["subject"] == "participation_event_comment" do
       notify_anonymous_participants(get_in(args, ["subject_params", "event_id"]), activity)
@@ -22,7 +24,7 @@ defmodule Mobilizon.Service.Workers.LegacyNotifierBuilder do
 
     args
     |> users_to_notify(author_id: args["author_id"], group_id: Map.get(args, "group_id"))
-    |> Enum.each(&Notifier.notify(&1, activity, single_activity: true))
+    |> Enum.each(&notify_user(&1, activity))
   end
 
   defp build_activity(args) do
@@ -48,6 +50,15 @@ defmodule Mobilizon.Service.Workers.LegacyNotifierBuilder do
     users_from_actor_ids(mentionned_actor_ids, Keyword.fetch!(options, :author_id))
   end
 
+  @spec users_to_notify(map(), Keyword.t()) :: list(Users.t())
+  defp users_to_notify(
+         %{"subject" => subject, "participant" => %{"actor_id" => actor_id}},
+         options
+       )
+       when subject in ["conversation_created", "conversation_replied"] do
+    users_from_actor_ids([actor_id], Keyword.fetch!(options, :author_id))
+  end
+
   defp users_to_notify(
          %{"subject" => "discussion_mention", "mentions" => mentionned_actor_ids},
          options
@@ -114,4 +125,9 @@ defmodule Mobilizon.Service.Workers.LegacyNotifierBuilder do
       )
     end)
   end
+
+  defp notify_user(user, activity) do
+    Logger.debug("Notifying #{user.email} for activity #{activity.subject}")
+    Notifier.notify(user, activity, single_activity: true)
+  end
 end

lib/web/auth/context.ex

@@ -44,7 +44,7 @@ defmodule Mobilizon.Web.Auth.Context do
 
     context = if is_nil(user_agent), do: context, else: Map.put(context, :user_agent, user_agent)
 
-    put_private(conn, :absinthe, %{context: context})
+    Absinthe.Plug.put_options(conn, context: context)
   end
 
   defp set_user_context({conn, context}, %User{id: user_id, email: user_email} = user) do

lib/web/cache/activity_pub.ex

@@ -3,9 +3,10 @@ defmodule Mobilizon.Web.Cache.ActivityPub do
   ActivityPub related cache.
   """
 
-  alias Mobilizon.{Actors, Discussions, Events, Posts, Resources, Todos, Tombstone}
+  alias Mobilizon.{Actors, Conversations, Discussions, Events, Posts, Resources, Todos, Tombstone}
   alias Mobilizon.Actors.Actor, as: ActorModel
   alias Mobilizon.Actors.Member
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.{Comment, Discussion}
   alias Mobilizon.Events.Event
   alias Mobilizon.Federation.ActivityPub.{Actor, Relay}
@@ -184,6 +185,23 @@ defmodule Mobilizon.Web.Cache.ActivityPub do
     end)
   end
 
+  @doc """
+  Gets a conversation participant by it's ID, with all associations loaded.
+  """
+  @spec get_conversation_by_id_with_preload(String.t()) ::
+          {:commit, Todo.t()} | {:ignore, nil}
+  def get_conversation_by_id_with_preload(id) do
+    Cachex.fetch(@cache, "conversation_participant_" <> id, fn "conversation_participant_" <> id ->
+      case Conversations.get_conversation_participant(id) do
+        %Conversation{} = conversation ->
+          {:commit, conversation}
+
+        nil ->
+          {:ignore, nil}
+      end
+    end)
+  end
+
   @doc """
   Gets a member by its UUID, with all associations loaded.
   """

lib/web/cache/cache.ex

@@ -4,6 +4,7 @@ defmodule Mobilizon.Web.Cache do
   """
 
   alias Mobilizon.Actors.{Actor, Member}
+  alias Mobilizon.Conversations.Conversation
   alias Mobilizon.Discussions.{Comment, Discussion}
   alias Mobilizon.Events.Event
   alias Mobilizon.Posts.Post
@@ -27,6 +28,10 @@ defmodule Mobilizon.Web.Cache do
   defdelegate get_todo_list_by_uuid_with_preload(uuid), to: ActivityPub
   @spec get_todo_by_uuid_with_preload(binary) :: {:commit, Todo.t()} | {:ignore, nil}
   defdelegate get_todo_by_uuid_with_preload(uuid), to: ActivityPub
+
+  @spec get_conversation_by_id_with_preload(binary) ::
+          {:commit, Conversation.t()} | {:ignore, nil}
+  defdelegate get_conversation_by_id_with_preload(uuid), to: ActivityPub
   @spec get_member_by_uuid_with_preload(binary) :: {:commit, Member.t()} | {:ignore, nil}
   defdelegate get_member_by_uuid_with_preload(uuid), to: ActivityPub
   @spec get_post_by_slug_with_preload(binary) :: {:commit, Post.t()} | {:ignore, nil}

lib/web/channels/graphql_socket.ex

@@ -13,9 +13,7 @@ defmodule Mobilizon.Web.GraphQLSocket do
     with {:ok, authed_socket} <-
            Guardian.Phoenix.Socket.authenticate(socket, Mobilizon.Web.Auth.Guardian, token),
          resource <- Guardian.Phoenix.Socket.current_resource(authed_socket) do
-      set_context(authed_socket, resource)
-
-      {:ok, authed_socket}
+      {:ok, set_context(authed_socket, resource)}
     else
       {:error, _} ->
         :error
@@ -24,8 +22,17 @@ defmodule Mobilizon.Web.GraphQLSocket do
 
   def connect(_args, _socket), do: :error
 
-  @spec id(any) :: nil
-  def id(_socket), do: nil
+  @spec id(Phoenix.Socket.t()) :: String.t() | nil
+  def id(%Phoenix.Socket{assigns: assigns}) do
+    context = Keyword.get(assigns.absinthe.opts, :context)
+    current_user = Map.get(context, :current_user)
+
+    if current_user do
+      "user_socket:#{current_user.id}"
+    else
+      nil
+    end
+  end
 
   @spec set_context(Phoenix.Socket.t(), User.t() | ApplicationToken.t()) :: Phoenix.Socket.t()
   defp set_context(socket, %User{} = user) do

lib/web/controllers/page_controller.ex

@@ -85,6 +85,12 @@ defmodule Mobilizon.Web.PageController do
     render_or_error(conn, &checks?/3, status, :todo, todo)
   end
 
+  @spec conversation(Plug.Conn.t(), map()) :: Plug.Conn.t() | {:error, :not_found}
+  def conversation(conn, %{"id" => slug}) do
+    {status, conversation} = Cache.get_conversation_by_id_with_preload(slug)
+    render_or_error(conn, &checks?/3, status, :conversation, conversation)
+  end
+
   @typep collections :: :resources | :posts | :discussions | :events | :todos
 
   @spec resources(Plug.Conn.t(), map()) :: Plug.Conn.t()

lib/web/router.ex

@@ -132,6 +132,7 @@ defmodule Mobilizon.Web.Router do
     get("/@:name/discussions", PageController, :discussions)
     get("/@:name/events", PageController, :events)
     get("/p/:slug", PageController, :post)
+    get("/conversations/:id", PageController, :conversation)
     get("/@:name/c/:slug", PageController, :discussion)
   end
 
@@ -176,6 +177,7 @@ defmodule Mobilizon.Web.Router do
 
     forward("/", Absinthe.Plug.GraphiQL,
       schema: Mobilizon.GraphQL.Schema,
+      socket: Mobilizon.Web.GraphQLSocket,
       interface: :playground
     )
   end

lib/web/templates/email/activity/_conversation_activity_item.html.heex

@@ -0,0 +1,20 @@
+<%= case @activity.subject do %>
+  <% :conversation_created -> %>
+    <%= dgettext("activity", "%{profile} mentionned you in a %{conversation}.", %{
+      profile: "<b>#{escaped_display_name_and_username(@activity.author)}</b>",
+      conversation:
+        "<a href=\"#{Routes.page_url(Mobilizon.Web.Endpoint,
+        :conversation,
+        @activity.subject_params["conversation_participant_id"]) |> URI.decode()}\">conversation</a>"
+    })
+    |> raw %>
+  <% :conversation_replied -> %>
+    <%= dgettext("activity", "%{profile} replied you in a %{conversation}.", %{
+      profile: "<b>#{escaped_display_name_and_username(@activity.author)}</b>",
+      conversation:
+        "<a href=\"#{Routes.page_url(Mobilizon.Web.Endpoint,
+        :conversation,
+        @activity.subject_params["conversation_participant_id"]) |> URI.decode()}\">conversation</a>"
+    })
+    |> raw %>
+<% end %>

lib/web/templates/email/activity/_conversation_activity_item.text.eex

@@ -0,0 +1,11 @@
+<%= case @activity.subject do %><% :conversation_created -> %><%= dgettext("activity", "%{profile} mentionned you in a conversation.",
+    %{
+        profile: Mobilizon.Actors.Actor.display_name_and_username(@activity.author),
+    }
+) %>
+<%= Routes.page_url(Mobilizon.Web.Endpoint, :conversation, @activity.subject_params["conversation_participant_id"]) |> URI.decode() %><% :conversation_replied -> %><%= dgettext("activity", "%{profile} replied you in a conversation.",
+    %{
+        profile: Mobilizon.Actors.Actor.display_name_and_username(@activity.author),
+    }
+) %>
+<%= Routes.page_url(Mobilizon.Web.Endpoint, :conversation, @activity.subject_params["conversation_participant_id"]) |> URI.decode() %><% end %>

lib/web/templates/email/email_direct_activity.html.heex

@@ -167,6 +167,10 @@
                                   <%= render("activity/_discussion_activity_item.html",
                                     activity: activity
                                   ) %>
+                                <% :conversation -> %>
+                                  <%= render("activity/_conversation_activity_item.html",
+                                    activity: activity
+                                  ) %>
                                 <% :event -> %>
                                   <%= render("activity/_event_activity_item.html",
                                     activity: activity

lib/web/templates/email/email_direct_activity.text.eex

@@ -15,7 +15,7 @@
 <% end %>
 
 <%= for activity <- Enum.take(group_activities, 5) do %>
-* <%= case activity.type do %><% :discussion -> %><%= render("activity/_discussion_activity_item.text", activity: activity) %><% :event -> %><%= render("activity/_event_activity_item.text", activity: activity) %><% :group -> %><%= render("activity/_group_activity_item.text", activity: activity) %>
+* <%= case activity.type do %><% :discussion -> %><%= render("activity/_discussion_activity_item.text", activity: activity) %><% :conversation -> %><%= render("activity/_conversation_activity_item.text", activity: activity) %><% :event -> %><%= render("activity/_event_activity_item.text", activity: activity) %><% :group -> %><%= render("activity/_group_activity_item.text", activity: activity) %>
 <% :member -> %><%= render("activity/_member_activity_item.text", activity: activity) %><% :post -> %><%= render("activity/_post_activity_item.text", activity: activity) %><% :resource -> %><%= render("activity/_resource_activity_item.text", activity: activity) %><% :comment -> %><%= render("activity/_comment_activity_item.text", activity: activity) %><% end %>
 <%= unless @single_activity do %><%= datetime_to_string(activity.inserted_at, @locale, :short) %><% end %>
 <% end %>