Add watch time, completion and trends to the data top list

lib/mave_core/analytics.ex

@@ -98,85 +98,6 @@ defmodule MaveCore.Analytics do
      }}
   end
 
-  @doc """
-  Returns the space's most viewed video embeds this year as
-  `%{embed_hash: hash, views: count}` maps, ordered by views.
-  """
-  def space_top_videos(space_hash, limit \\ 25)
-
-  def space_top_videos(space_hash, limit)
-      when is_binary(space_hash) and is_integer(limit) and limit > 0 do
-    with {:ok, space_hash} <- parse_space_hash(space_hash) do
-      if SharedStorageSpace.hash?(space_hash) do
-        {:ok, []}
-      else
-        do_space_top_videos(space_hash, limit)
-      end
-    end
-  end
-
-  def space_top_videos(_space_hash, _limit), do: {:error, :invalid_space_hash}
-
-  defp do_space_top_videos(space_hash, limit) do
-    now = DateTime.utc_now()
-    year_from = DateTime.new!(Date.new!(now.year, 1, 1), ~T[00:00:00], "Etc/UTC")
-
-    db_name = clickhouse_database_name()
-    threshold = @min_watch_seconds * 1000
-
-    sql = """
-    /* mave_perf:spaces.top_videos */
-    WITH ordered AS (
-      SELECT
-        embed_hash,
-        session_id,
-        timestamp,
-        name,
-        leadInFrame(timestamp) OVER (
-          PARTITION BY embed_hash, session_id
-          ORDER BY timestamp
-          ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-        ) AS next_ts,
-        leadInFrame(name) OVER (
-          PARTITION BY embed_hash, session_id
-          ORDER BY timestamp
-          ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-        ) AS next_name
-      FROM #{db_name}.events
-      PREWHERE space_hash = #{string_literal(space_hash)}
-        AND timestamp >= #{datetime_literal(year_from)}
-        AND timestamp < #{datetime_literal(now)}
-      WHERE name IN ('play', 'pause')
-    ),
-    session_aggregates AS (
-      SELECT
-        embed_hash,
-        session_id,
-        sumIf(
-          greatest(toInt64(0), toUnixTimestamp64Milli(next_ts) - toUnixTimestamp64Milli(timestamp)),
-          name = 'play' AND next_name = 'pause'
-        ) AS watch_ms
-      FROM ordered
-      GROUP BY embed_hash, session_id
-    )
-    SELECT
-      embed_hash,
-      countIf(watch_ms >= #{threshold}) AS views
-    FROM session_aggregates
-    GROUP BY embed_hash
-    HAVING views > 0
-    ORDER BY views DESC, embed_hash ASC
-    LIMIT #{limit}
-    """
-
-    query_clickhouse(sql, {:error, :analytics_unavailable}, fn rows ->
-      {:ok,
-       Enum.map(rows, fn [embed_hash, views] ->
-         %{embed_hash: embed_hash, views: to_int(views)}
-       end)}
-    end)
-  end
-
   def space_monthly_watch_time(space_hash) when is_binary(space_hash) do
     with {:ok, space_hash} <- parse_space_hash(space_hash) do
       if SharedStorageSpace.hash?(space_hash) do

lib/mave_core/analytics/overview.ex

@@ -81,6 +81,90 @@ defmodule MaveCore.Analytics.Overview do
     }
   end
 
+  @doc """
+  Returns the space's most viewed videos in `period`, ordered by views, as
+  maps with `:embed_hash`, `:views`, `:watch_ms`, `:avg_completion` and
+  `:series` (views per hour, day or month, oldest first).
+  """
+  def top_videos(space_hash, period, limit \\ 25, now \\ DateTime.utc_now())
+
+  def top_videos(space_hash, period, limit, now)
+      when is_binary(space_hash) and period in @periods and is_integer(limit) and limit > 0 do
+    window = window(period, now)
+
+    with {:ok, space_hash} <- parse_space_hash(space_hash) do
+      if SharedStorageSpace.hash?(space_hash),
+        do: {:ok, []},
+        else: query_top_videos(space_hash, window, limit)
+    end
+  end
+
+  def top_videos(_space_hash, _period, _limit, _now), do: {:error, :invalid_request}
+
+  defp query_top_videos(space_hash, window, limit) do
+    sql = """
+    /* mave_perf:spaces.top_videos */
+    #{views_cte(space_hash, %{window | previous_from: window.from})},
+    top AS (
+      SELECT
+        embed_hash,
+        count() AS views,
+        sum(watch_ms) AS watch_ms,
+        ifNotFinite(avgIf(completion, video_duration > 0), 0) AS completion
+      FROM views
+      GROUP BY embed_hash
+      ORDER BY views DESC, embed_hash ASC
+      LIMIT #{limit}
+    )
+    SELECT
+      'video' AS kind,
+      embed_hash,
+      '' AS key,
+      toUInt64(views) AS views,
+      toInt64(watch_ms) AS watch_ms,
+      toFloat64(completion) AS value
+    FROM top
+    UNION ALL
+    SELECT
+      'bucket' AS kind,
+      embed_hash,
+      #{bucket_key_sql(window.bucket)} AS key,
+      toUInt64(count()) AS views,
+      toInt64(0) AS watch_ms,
+      toFloat64(0) AS value
+    FROM views
+    WHERE embed_hash IN (SELECT embed_hash FROM top)
+    GROUP BY embed_hash, key
+    """
+
+    query_clickhouse(sql, {:error, :analytics_unavailable}, fn rows ->
+      {:ok, build_top_videos(rows, window)}
+    end)
+  end
+
+  defp build_top_videos(rows, window) do
+    {videos, buckets} = Enum.split_with(rows, &(hd(&1) == "video"))
+
+    series =
+      Enum.group_by(buckets, &Enum.at(&1, 1), fn [_, _, key, views, _, _] ->
+        {key, to_int(views)}
+      end)
+
+    videos
+    |> Enum.map(fn [_, hash, _, views, watch_ms, completion] ->
+      counts = Map.new(Map.get(series, hash, []))
+
+      %{
+        embed_hash: hash,
+        views: to_int(views),
+        watch_ms: to_int(watch_ms),
+        avg_completion: to_float(completion),
+        series: Enum.map(window.buckets, &Map.get(counts, bucket_key(&1), 0))
+      }
+    end)
+    |> Enum.sort_by(&{-&1.views, &1.embed_hash})
+  end
+
   defp query_overview(space_hash, period, window) do
     sql = """
     /* mave_perf:spaces.overview */
@@ -96,7 +180,7 @@ defmodule MaveCore.Analytics.Overview do
     UNION ALL
     SELECT
       'series' AS kind,
-      formatDateTime(toDateTime(#{bucket_sql(window.bucket)}(started_at, 'UTC'), 'UTC'), '%Y-%m-%d %H:%i:%S', 'UTC') AS key,
+      #{bucket_key_sql(window.bucket)} AS key,
       toUInt64(count()) AS views,
       toInt64(sum(watch_ms)) AS watch_ms,
       toFloat64(0) AS value
@@ -317,6 +401,10 @@ defmodule MaveCore.Analytics.Overview do
     %{dt | year: div(index, 12), month: rem(index, 12) + 1}
   end
 
+  defp bucket_key_sql(bucket) do
+    "formatDateTime(toDateTime(#{bucket_sql(bucket)}(started_at, 'UTC'), 'UTC'), '%Y-%m-%d %H:%i:%S', 'UTC')"
+  end
+
   defp bucket_sql(:hour), do: "toStartOfHour"
   defp bucket_sql(:day), do: "toStartOfDay"
   defp bucket_sql(:month), do: "toStartOfMonth"

lib/mave_core_web/dashboard_routes.ex

@@ -24,7 +24,7 @@ defmodule MaveCoreWeb.DashboardRoutes do
   end
 
   def data_top_list_path(space_or_user, opts \\ []) do
-    path_for(space_or_user, "/data/top-list", period_params(opts), opts)
+    path_for(space_or_user, "/data/top-list", period_params(opts) ++ sort_params(opts), opts)
   end
 
   def settings_path(space_or_user, tab \\ nil, opts \\ []) do
@@ -65,6 +65,13 @@ defmodule MaveCoreWeb.DashboardRoutes do
     end
   end
 
+  defp sort_params(opts) do
+    case Keyword.get(opts, :sort) do
+      sort when is_binary(sort) and sort != "" -> [sort: sort]
+      _ -> []
+    end
+  end
+
   defp maybe_put_param(params, _key, false, _value), do: params
   defp maybe_put_param(params, key, true, value), do: Keyword.put(params, key, value)