feat: GraphQL subscriptions

Author: Betree

apps/cf/lib/comments/comments.ex

@@ -37,10 +37,25 @@ defmodule CF.Comments do
   source fetcher should be moved to a job.
   """
   def add_comment(user, video_id, params, source_url \\ nil, source_fetch_callback \\ nil) do
-    # TODO [Security] What if reply_to_id refer to a comment that is on a different statement ?
     UserPermissions.check!(user, :create, :comment)
     source_url = source_url && Source.prepare_url(source_url)
 
+    # Handle the case where reply_to_id refer to a comment that is on a different statement
+    if Map.get(params, :reply_to_id) do
+      reply_to = Repo.get!(Comment, Map.get(params, :reply_to_id))
+
+      cond do
+        is_nil(reply_to) ->
+          raise "Reply to comment not found"
+
+        reply_to.statement_id != params.statement_id ->
+          raise "Reply to comment on a different statement"
+
+        true ->
+          true
+      end
+    end
+
     # Load source from DB or create a changeset to make a new one
     source =
       source_url &&

apps/cf/lib/statements/statements.ex

@@ -18,7 +18,7 @@ defmodule CF.Statements do
   """
   def update!(user_id, statement = %Statement{is_removed: false}, changes) do
     UserPermissions.check!(user_id, :update, :statement)
-    changeset = Statement.changeset(statement, changes)
+    changeset = Statement.changeset_update(statement, changes)
 
     if changeset.changes == %{} do
       Result.ok(statement)

apps/cf_graphql/lib/application.ex

@@ -9,7 +9,8 @@ defmodule CF.Graphql.Application do
       # Start the PubSub system
       {Phoenix.PubSub, name: CF.Graphql.PubSub},
       # Start the endpoint when the application starts
-      {CF.GraphQLWeb.Endpoint, []}
+      {CF.GraphQLWeb.Endpoint, []},
+      {Absinthe.Subscription, CF.GraphQLWeb.Endpoint}
     ]
 
     # See https://hexdocs.pm/elixir/Supervisor.html

apps/cf_graphql/lib/endpoint.ex

@@ -1,5 +1,8 @@
 defmodule CF.GraphQLWeb.Endpoint do
   use Phoenix.Endpoint, otp_app: :cf_graphql
+  use Absinthe.Phoenix.Endpoint
+
+  socket("/socket", CF.GraphQLWeb.UserSocket, websocket: true, longpoll: false)
 
   plug(Plug.RequestId)
   plug(Plug.Logger)

apps/cf_graphql/lib/resolvers/comments.ex

@@ -3,6 +3,10 @@ defmodule CF.Graphql.Resolvers.Comments do
   import Ecto.Query
   alias DB.Repo
   alias DB.Schema.Vote
+  alias DB.Schema.Comment
+  alias DB.Schema.Statement
+  alias CF.Comments
+  alias CF.Graphql.Subscriptions
 
   def score(comment, _args, _info) do
     batch({__MODULE__, :comments_scores}, comment.id, fn results ->
@@ -18,4 +22,75 @@ defmodule CF.Graphql.Resolvers.Comments do
     |> Repo.all()
     |> Enum.into(%{})
   end
+
+  # Mutations
+
+  def create(_root, args = %{statement_id: statement_id}, %{context: %{user: user}}) do
+    # Get statement to find video_id
+    statement = Repo.get!(Statement, statement_id)
+    video_id = statement.video_id
+    reply_to_id = Map.get(args, :reply_to_id)
+
+    params = %{
+      "statement_id" => statement_id,
+      "text" => Map.get(args, :text),
+      "reply_to_id" => reply_to_id,
+      "approve" => Map.get(args, :approve)
+    }
+
+    source_url = Map.get(args, :source)
+
+    # Comments.add_comment returns the comment directly or {:error, reason}
+    case Comments.add_comment(user, video_id, params, source_url) do
+      {:error, reason} ->
+        {:error, reason}
+
+      comment ->
+        # Preload associations for GraphQL response
+        comment = Repo.preload(comment, [:source, :user, :statement])
+
+        Subscriptions.publish_comment_added(comment, video_id)
+        {:ok, comment}
+    end
+  end
+
+  def delete(_root, %{id: id}, %{context: %{user: user}}) do
+    comment = Repo.get!(Comment, id) |> Repo.preload(:statement)
+    video_id = comment.statement.video_id
+
+    case Comments.delete_comment(user, video_id, comment) do
+      nil ->
+        {:ok, %{id: id, statement_id: comment.statement_id, reply_to_id: comment.reply_to_id}}
+
+      _ ->
+        Subscriptions.publish_comment_removed(comment, video_id)
+        {:ok, %{id: id, statement_id: comment.statement_id, reply_to_id: comment.reply_to_id}}
+    end
+  end
+
+  def vote(_root, %{comment_id: comment_id, value: value}, %{context: %{user: user}}) do
+    # Get comment and preload statement to access video_id
+    comment = Repo.get!(Comment, comment_id) |> Repo.preload(:statement)
+    video_id = comment.statement.video_id
+
+    case Comments.vote!(user, video_id, comment_id, value) do
+      {:ok, comment, vote, prev_value} ->
+        # Calculate score diff (same logic as comments_channel.ex)
+        diff = value_diff(prev_value, vote.value)
+
+        # Publish score diff via subscription
+        Subscriptions.publish_comment_score_diff(comment, diff, video_id)
+
+        # Preload associations for GraphQL response
+        comment = Repo.preload(comment, [:source, :user, :statement])
+        {:ok, comment}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  # Helper function to calculate vote value diff (matches comments_channel.ex logic)
+  defp value_diff(0, new_value), do: new_value
+  defp value_diff(prev_value, new_value), do: new_value - prev_value
 end

apps/cf_graphql/lib/resolvers/statements.ex

@@ -5,8 +5,16 @@ defmodule CF.Graphql.Resolvers.Statements do
 
   alias Kaur.Result
 
+  alias Ecto.Multi
   alias DB.Repo
   alias DB.Schema.Statement
+  alias CF.Accounts.UserPermissions
+  alias CF.Actions.ActionCreator
+  alias CF.Graphql.Subscriptions
+  alias CF.Algolia.StatementsIndex
+  alias CF.Statements
+
+  import CF.Actions.ActionCreator, only: [action_remove: 2]
 
   # Queries
 
@@ -16,4 +24,67 @@ defmodule CF.Graphql.Resolvers.Statements do
     |> Repo.paginate(page: offset, page_size: limit)
     |> Result.ok()
   end
+
+  # Mutations
+
+  def create(_root, args = %{video_id: video_id, text: _text, time: _time}, %{
+        context: %{user: user}
+      }) do
+    user_id = user.id
+    UserPermissions.check!(user_id, :create, :statement)
+
+    # Absinthe automatically converts GraphQL camelCase to snake_case
+    changeset = Statement.changeset(%Statement{video_id: video_id}, args)
+
+    Multi.new()
+    |> Multi.insert(:statement, changeset)
+    |> Multi.run(:action_create, fn _repo, %{statement: statement} ->
+      Repo.insert(ActionCreator.action_create(user_id, statement))
+    end)
+    |> Repo.transaction()
+    |> case do
+      {:ok, %{statement: statement}} ->
+        Subscriptions.publish_statement_added(statement)
+        StatementsIndex.save_object(statement)
+        {:ok, statement}
+
+      {:error, _operation, reason, _changes} ->
+        {:error, reason}
+    end
+  end
+
+  def update(_root, args = %{id: id}, %{context: %{user: user}}) when not is_nil(id) do
+    user_id = user.id
+    statement = Repo.get_by!(Statement, id: id, is_removed: false)
+
+    case Statements.update!(user_id, statement, args) do
+      {:ok, updated_statement} ->
+        Subscriptions.publish_statement_updated(updated_statement)
+        StatementsIndex.save_object(updated_statement)
+        {:ok, updated_statement}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
+  def delete(_root, %{id: id}, %{context: %{user: user}}) do
+    user_id = user.id
+    UserPermissions.check!(user_id, :remove, :statement)
+    statement = Repo.get_by!(Statement, id: id, is_removed: false)
+
+    Multi.new()
+    |> Multi.update(:statement, Statement.changeset_remove(statement))
+    |> Multi.insert(:action_remove, action_remove(user_id, statement))
+    |> Repo.transaction()
+    |> case do
+      {:ok, _} ->
+        Subscriptions.publish_statement_removed(id, statement.video_id)
+        StatementsIndex.delete_object(statement)
+        {:ok, %{id: id}}
+
+      {:error, _operation, reason, _changes} ->
+        {:error, reason}
+    end
+  end
 end