Implement how many people are currently watching a video

README.md

@@ -115,6 +115,18 @@ An example response:
 }
 ```
 
+### `/api/v1/watching` (POST or GET)
+
+Watching is a very simple request to determine how many people are currently watching a video. You can specify the video(s) by using the identifier or by querying for metadata.
+
+The response will show you how many viewers are currently watching your request, here's an example response:
+
+```json
+{
+  "watching": 231
+}
+```
+
 ### `/api/v1/sources` (POST or GET)
 
 Retrieve the number of plays per source using the video's `identifier` and/or `query` for metadata. A source refers to the location where your video is placed, which can be particularly useful when embedding the same video across multiple pages/sites.

server/lib/mave_metrics/api.ex

@@ -6,11 +6,26 @@ defmodule MaveMetrics.API do
   import Ecto.Query, warn: false
   import EctoCase
   alias MaveMetrics.{Repo, Video, Duration}
+  alias MaveMetricsWeb.Presence
 
   @default_timeframe "7 days"
   @default_interval "12 months"
   @default_minimum_watch_seconds 1
 
+  def get_watching(%{"video" => query}) do
+    video_ids =
+      Video
+      |> where([v], fragment("? @> ?", v.metadata, ^query))
+      |> select([v], v.id)
+      |> Repo.all()
+
+    if video_ids == [] do
+      0
+    else
+      video_ids |> Enum.map(&map_size(Presence.list("video:#{&1}"))) |> Enum.sum()
+    end
+  end
+
   def get_plays(%{"video" => query}, interval, timeframe, minimum_watch_seconds) do
     interval = interval || @default_interval
     timeframe = timeframe || @default_timeframe

server/lib/mave_metrics/application.ex

@@ -18,6 +18,8 @@ defmodule MaveMetrics.Application do
       {Task.Supervisor, name: MaveMetrics.TaskSupervisor},
       # Start the PubSub system
       {Phoenix.PubSub, name: MaveMetrics.PubSub},
+      # Presence
+      MaveMetricsWeb.Presence,
       # Start Finch
       {Finch, name: MaveMetrics.Finch},
       # Start cluster

server/lib/mave_metrics_web/channels/session_channel.ex

@@ -1,5 +1,6 @@
 defmodule MaveMetricsWeb.SessionChannel do
   use MaveMetricsWeb, :channel
+  alias MaveMetricsWeb.Presence
 
   alias MaveMetrics.{Stats, Keys, Pipeline}
 
@@ -69,6 +70,8 @@ defmodule MaveMetricsWeb.SessionChannel do
       video_id: video.id
     })
 
+    Presence.track(self(), "video:#{video.id}", "#{session.id}", %{})
+
     {
       :noreply,
       socket
@@ -97,6 +100,8 @@ defmodule MaveMetricsWeb.SessionChannel do
       video_id: video_id
     })
 
+    Presence.track(self(), "video:#{video_id}", "#{session_id}", %{})
+
     {:noreply, socket}
   end
 
@@ -116,6 +121,8 @@ defmodule MaveMetricsWeb.SessionChannel do
       video_id: video_id
     })
 
+    Presence.untrack(self(), "video:#{video_id}", "#{session_id}")
+
     {:noreply, socket}
   end
 
@@ -145,6 +152,8 @@ defmodule MaveMetricsWeb.SessionChannel do
       session_id: session_id,
       video_id: video_id
     })
+
+    Presence.untrack(self(), "video:#{video_id}", "#{session_id}")
   end
 
   defp on_disconnect(_, _) do

server/lib/mave_metrics_web/controllers/api/watching_controller.ex

@@ -0,0 +1,27 @@
+defmodule MaveMetricsWeb.API.WatchingController do
+  use MaveMetricsWeb, :controller
+
+  alias MaveMetrics.API
+
+  def get_watching(conn, _params) do
+    {:ok, body, _conn} = Plug.Conn.read_body(conn)
+    params = Jason.decode!(body)
+    conn |> watching(params)
+  end
+
+  def watching(conn, %{"query" => filters} = params) do
+    result =
+      API.get_watching(filters)
+
+    conn
+    |> json(%{watching: result})
+    |> halt
+  end
+
+  def watching(conn, _params) do
+    conn
+    |> put_status(400)
+    |> json(%{error: "Requires a valid JSON query struct."})
+    |> halt
+  end
+end

server/lib/mave_metrics_web/presence.ex

@@ -0,0 +1,5 @@
+defmodule MaveMetricsWeb.Presence do
+  use Phoenix.Presence,
+    otp_app: :mave_metrics,
+    pubsub_server: MaveMetrics.PubSub
+end

server/lib/mave_metrics_web/router.ex

@@ -20,6 +20,8 @@ defmodule MaveMetricsWeb.Router do
   scope "/api/v1", MaveMetricsWeb do
     pipe_through([:api, :require_api_authentication?])
 
+    post("/watching", API.WatchingController, :watching)
+    get("/watching", API.WatchingController, :get_watching)
     post("/views", API.ViewsController, :views)
     get("/views", API.ViewsController, :get_views)
     post("/sources", API.SourcesController, :sources)