From 70ad2cd22acbb3dfb4448dde417cbf41e76e7fa4 Mon Sep 17 00:00:00 2001 From: David Date: Fri, 23 Oct 2026 21:48:03 +0200 Subject: [PATCH 1/2] Keep time-dependent analytics reads scoped to affected sessions --- docs/analytics_sessions.md | 10 +- lib/mave_core/analytics/sessions.ex | 28 ++- lib/mave_core/analytics/sessions/query.ex | 196 ++++++++++++------ ...23200000_add_session_future_visibility.exs | 17 ++ test/mave_core/analytics/sessions_test.exs | 71 ++++++- 5 files changed, 244 insertions(+), 78 deletions(-) create mode 100644 priv/clickhouse_repo/migrations/20261023200000_add_session_future_visibility.exs diff --git a/docs/analytics_sessions.md b/docs/analytics_sessions.md index 1aa886e..0fc6a7e 100644 --- a/docs/analytics_sessions.md +++ b/docs/analytics_sessions.md @@ -91,8 +91,7 @@ counts to obtain period views. Sixteen deterministic session buckets bound each publication. Initial builds read the bucket's source events. Subsequent refreshes compare per-session counts/fingerprints, copy unchanged prepared rows into a new immutable -partition, and recalculate only changed sessions. Session rows, compact year summaries and day reports are copied together. More than 50,000 changed sessions or a due -future-event deadline uses the bounded full-bucket path. PostgreSQL's +partition, and recalculate only changed sessions. Session rows, compact year summaries and day reports are copied together. More than 50,000 changed sessions trigger the bounded full-bucket path. PostgreSQL's `analytics_session_buckets` publishes one active generation per space/bucket. An advisory transaction lock serializes publication and invalidation within a space. A failed ClickHouse insert never changes the active pointer. A retry @@ -102,7 +101,12 @@ Subsequent readers use the last complete bucket versions during refresh. Format versions prevent incompatible older generations from being read. Future-dated events are excluded from summaries until their time arrives; the published metadata records a refresh deadline so they become visible without -requiring another insert. +requiring another insert. A materialized per-session deadline also lets the +processor refresh only those sessions. Between refreshes, current overview and +date-aligned report queries combine compact rows with detailed calculations for +the affected sessions; one newly visible event does not force a detailed scan +of the entire space. Historical and partial-day reads retain precise segment +calculations. The `analytics_session_inventory` projection summarizes source row counts and field fingerprints. It lives inside the raw MergeTree table, so it follows diff --git a/lib/mave_core/analytics/sessions.ex b/lib/mave_core/analytics/sessions.ex index 9fb9bb9..fa6eca4 100644 --- a/lib/mave_core/analytics/sessions.ex +++ b/lib/mave_core/analytics/sessions.ex @@ -116,11 +116,15 @@ defmodule MaveCore.Analytics.Sessions do def range_data(_space_hash, _from, _to, _opts), do: {:error, :invalid_range} defp range_query(space, generations, from, to, granularity, mode) do - if mode == :series and midnight?(from) and - (midnight?(to) or reports_cover_time?(space, generations, to)) and - reports_complete?(space, generations), - do: Query.report_range(granularity), - else: Query.range(granularity, mode) + if midnight?(from) and reports_complete?(space, generations) do + cond do + midnight?(to) -> Query.report_range(granularity, mode) + reports_cover_time?(space, generations, to) -> Query.live_range(granularity, mode) + true -> Query.range(granularity, mode) + end + else + Query.range(granularity, mode) + end end defp midnight?(at), do: rem(DateTime.to_unix(at, :microsecond), 86_400_000_000) == 0 @@ -130,7 +134,7 @@ defmodule MaveCore.Analytics.Sessions do "SELECT count(*), max(latest_event_at), min(refresh_after) FROM analytics_session_buckets WHERE space_hash=$1 AND generation::text = ANY($2)", [space, generations] ) do - {:ok, %{rows: [[count, latest, nil]]}} -> + {:ok, %{rows: [[count, latest, _due]]}} -> count == Query.buckets() and (is_nil(latest) or NaiveDateTime.compare(naive(to), latest) != :lt) @@ -343,8 +347,8 @@ defmodule MaveCore.Analytics.Sessions do "SELECT generation::text, format_version, refresh_after FROM analytics_session_buckets WHERE space_hash = $1 AND bucket = $2", [space, bucket] ).rows do - [[previous, version, due]] -> - if version in [3, 4, Query.version()] and not refresh_due?(due) do + [[previous, version, _due]] -> + if version in [3, 4, 5, Query.version()] do {previous, changed, full} = changed_plan!(space, bucket, previous) {previous, changed, full, report_refresh_mode(version, full), @@ -611,9 +615,11 @@ defmodule MaveCore.Analytics.Sessions do historical? = not is_nil(latest) and NaiveDateTime.compare(naive(now), latest) == :lt newly_visible? = not is_nil(due) and NaiveDateTime.compare(naive(now), due) != :lt - if historical? or newly_visible?, - do: {Query.overview_at(), Query.unique_watch_time()}, - else: {Query.overview(), Query.unique_overview()} + cond do + historical? -> {Query.overview_at(), Query.unique_watch_time()} + newly_visible? -> {Query.live_overview(), Query.live_unique_overview()} + true -> {Query.overview(), Query.unique_overview()} + end end defp lock!(space_hash) do diff --git a/lib/mave_core/analytics/sessions/query.ex b/lib/mave_core/analytics/sessions/query.ex index fff5b17..3f17444 100644 --- a/lib/mave_core/analytics/sessions/query.ex +++ b/lib/mave_core/analytics/sessions/query.ex @@ -7,7 +7,7 @@ defmodule MaveCore.Analytics.Sessions.Query do @buckets 16 def buckets, do: @buckets - def version, do: 5 + def version, do: 6 def inventory do """ @@ -28,12 +28,13 @@ defmodule MaveCore.Analytics.Sessions.Query do WHERE space_hash = {$0:String} AND modulo(cityHash64(session_id), #{@buckets}) = {$1:UInt8} GROUP BY session_id ), previous AS ( - SELECT session_id, source_rows, fingerprint FROM analytics_sessions + SELECT session_id, source_rows, fingerprint, future_at FROM analytics_sessions WHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$2:UUID} ) SELECT if(source.source_rows = 0, previous.session_id, source.session_id) FROM source FULL OUTER JOIN previous ON source.session_id = previous.session_id WHERE source.source_rows != previous.source_rows OR source.fingerprint != previous.fingerprint + OR previous.future_at <= now64(6) LIMIT 50001 SETTINGS max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:build_memory_bytes, 1_073_741_824)}, join_use_nulls = 0, max_execution_time = 60 @@ -110,24 +111,45 @@ defmodule MaveCore.Analytics.Sessions.Query do "SELECT countIf(complete = 0) FROM analytics_session_reports PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)}" end - def report_range(granularity) when granularity in [:day, :week, :month] do + def report_range(granularity, mode \\ :series) + when granularity in [:day, :week, :month] and mode in [:series, :videos] do + range_query(report_pieces(granularity, mode), mode, "analytics_session_reports") + end + + def live_range(granularity, mode) + when granularity in [:day, :week, :month] and mode in [:series, :videos] do + pieces = + report_pieces(granularity, mode, true) <> + "\nUNION ALL\n" <> range_pieces(granularity, mode, true) + + range_query(pieces, mode, "analytics_sessions") + end + + defp report_pieces(granularity, mode, exclude_future \\ false) do key = - %{ - day: "toString(p.2)", - week: "toString(toMonday(p.2))", - month: "toString(toStartOfMonth(p.2))" - }[granularity] + if mode == :videos do + "p.1" + else + %{ + day: "toString(p.2)", + week: "toString(toMonday(p.2))", + month: "toString(toStartOfMonth(p.2))" + }[granularity] + end + + filter = + if exclude_future, + do: "AND session_id NOT IN (#{future_sessions("isNotNull(future_at)")})", + else: "" """ - WITH pieces AS ( - SELECT arrayMap(p -> (#{key},p.3,p.4), - arrayFilter(p -> (empty({$4:Array(String)}) OR p.1 IN {$4:Array(String)}) - AND toDateTime64(p.2,6,'UTC') >= {$2:DateTime64(6)} - AND toDateTime64(p.2,6,'UTC') < {$3:DateTime64(6)}, days)) AS pieces - FROM analytics_session_reports - PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} - ) - #{range_aggregation(:series, "analytics_session_reports")} + SELECT arrayMap(p -> (#{key},p.3,p.4), + arrayFilter(p -> (empty({$4:Array(String)}) OR p.1 IN {$4:Array(String)}) + AND toDateTime64(p.2,6,'UTC') >= {$2:DateTime64(6)} + AND toDateTime64(p.2,6,'UTC') < {$3:DateTime64(6)}, days)) AS pieces + FROM analytics_session_reports + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + #{filter} """ end @@ -169,10 +191,9 @@ defmodule MaveCore.Analytics.Sessions.Query do def revision do """ - WITH arrayMap(e -> e.2, arrayFilter(e -> e.2 >= created_at, events)) AS future_events, - arrayMap(e -> e.2, arrayFilter(e -> e.2 < created_at, events)) AS past_events + WITH arrayMap(e -> e.2, arrayFilter(e -> e.2 < created_at, events)) AS past_events SELECT sum(source_rows), toString(sumWithOverflow(fingerprint)), - minOrNullIf(arrayMin(future_events), notEmpty(future_events)), + min(future_at), maxOrNullIf(arrayMax(past_events), notEmpty(past_events)) FROM analytics_sessions WHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$2:UUID} @@ -212,10 +233,38 @@ defmodule MaveCore.Analytics.Sessions.Query do """ end - def overview_at do + def overview_at, do: overview_query(detail_metrics()) + def overview, do: overview_query(compact_metrics()) + + def live_overview do + affected = future_sessions("future_at <= {$5:DateTime64(6)}") + + overview_query( + compact_metrics("AND session_id NOT IN (#{affected})") <> + "\nUNION ALL\n" <> detail_metrics("AND future_at <= {$5:DateTime64(6)}") + ) + end + + defp compact_metrics(filter \\ "") do """ - /* mave_perf:sessions.overview */ - WITH metrics AS ( + SELECT + toMonth(first_at, 'UTC') AS month, + nullIf(source_device, '') AS device, + nullIf(source_browser, '') AS browser, + watched AS watch_year, + arraySum(arrayMap(d -> if(d.1 >= toDate({$3:DateTime64(6)}, 'UTC'), d.2, 0), days)) AS watch_month, + arraySum(arrayMap(d -> if(d.1 = toDate({$4:DateTime64(6)}, 'UTC'), d.2, 0), days)) AS watch_today + FROM analytics_session_years + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + WHERE year = toYear({$2:DateTime64(6)}, 'UTC') + #{filter} + """ + end + + defp detail_metrics(filter \\ "") do + """ + SELECT month, device, browser, watch_year, watch_month, watch_today + FROM ( SELECT arrayFilter(e -> e.2 >= {$2:DateTime64(6)} AND e.2 < {$5:DateTime64(6)}, events) AS ev, arrayFilter(s -> s.2 >= {$2:DateTime64(6)} AND s.3 < {$5:DateTime64(6)}, segments) AS seg, @@ -228,42 +277,17 @@ defmodule MaveCore.Analytics.Sessions.Query do arraySum(arrayMap(s -> if(s.2 >= {$4:DateTime64(6)}, greatest(toInt64(0), dateDiff('millisecond', s.2, s.3)), 0), seg)) AS watch_today FROM analytics_sessions PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + #{filter} WHERE notEmpty(ev) ) - SELECT kind, bucket_month, dimension, views, views_today, views_month, - total_watch_today, total_watch_month, total_watch_year - FROM ( - SELECT multiIf(grouping(month) = 0, 'month', grouping(device) = 0, 'device', - grouping(browser) = 0, 'browser', 'total') AS kind, - if(grouping(month) = 0, toNullable(month), NULL) AS bucket_month, - multiIf(grouping(device) = 0, device, grouping(browser) = 0, browser, NULL) AS dimension, - countIf(watch_year >= 1000) AS views, - countIf(watch_today >= 1000) AS views_today, - countIf(watch_month >= 1000) AS views_month, - sum(watch_today) AS total_watch_today, sum(watch_month) AS total_watch_month, sum(watch_year) AS total_watch_year - FROM metrics - GROUP BY GROUPING SETS ((), (month), (device), (browser)) - ) - QUALIFY kind IN ('total', 'month') OR - row_number() OVER (PARTITION BY kind ORDER BY views DESC, dimension ASC) <= 50 - SETTINGS max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:read_memory_bytes, 536_870_912)}, max_execution_time = 10 """ end - def overview do + defp overview_query(metrics) do """ /* mave_perf:sessions.overview */ WITH metrics AS ( - SELECT - toMonth(first_at, 'UTC') AS month, - nullIf(source_device, '') AS device, - nullIf(source_browser, '') AS browser, - watched AS watch_year, - arraySum(arrayMap(d -> if(d.1 >= toDate({$3:DateTime64(6)}, 'UTC'), d.2, 0), days)) AS watch_month, - arraySum(arrayMap(d -> if(d.1 = toDate({$4:DateTime64(6)}, 'UTC'), d.2, 0), days)) AS watch_today - FROM analytics_session_years - PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} - WHERE year = toYear({$2:DateTime64(6)}, 'UTC') + #{metrics} ) SELECT kind, bucket_month, dimension, views, views_today, views_month, total_watch_today, total_watch_month, total_watch_year @@ -286,11 +310,39 @@ defmodule MaveCore.Analytics.Sessions.Query do end def unique_overview do - """ - WITH intervals AS ( + compact_unique_query(""" SELECT unique_intervals AS iv FROM analytics_session_years PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} WHERE year = toYear({$2:DateTime64(6)}, 'UTC') + """) + end + + def live_unique_overview do + affected = future_sessions("future_at <= {$5:DateTime64(6)}") + + compact_unique_query(""" + SELECT unique_intervals AS iv FROM analytics_session_years + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + WHERE year = toYear({$2:DateTime64(6)}, 'UTC') AND session_id NOT IN (#{affected}) + UNION ALL + SELECT arrayMap(s -> (toDate(s.2, 'UTC'), toUInt64(greatest(0, floor(s.4))), + toUInt64(greatest(0, floor(s.4))) + least(toUInt64(3600), + toUInt64(greatest(0, ceil(s.5) - floor(s.4))), + toUInt64(greatest(0, intDiv(dateDiff('millisecond', s.2, s.3), 1000))))), + arrayFilter(s -> s.2 >= {$2:DateTime64(6)} AND s.3 < {$5:DateTime64(6)} + AND dateDiff('millisecond', s.2, s.3) < 3600000 + AND isFinite(s.4) AND isFinite(s.5) AND s.4 >= 0 AND s.5 > s.4, segments)) AS iv + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + AND future_at <= {$5:DateTime64(6)} + """) + end + + defp compact_unique_query(intervals) do + """ + WITH intervals AS ( + #{intervals} + ), lengths AS ( SELECT arrayReduce('intervalLengthSum', arrayMap(i -> i.2, iv), arrayMap(i -> i.3, iv)) AS year, arrayReduce('intervalLengthSum', @@ -388,6 +440,10 @@ defmodule MaveCore.Analytics.Sessions.Query do def range(granularity, mode) when granularity in [:day, :week, :month] and mode in [:series, :videos] do + range_query(range_pieces(granularity, mode), mode, "analytics_sessions") + end + + defp range_pieces(granularity, mode, only_future \\ false) do bucket = %{ day: "toDate(p.2, 'UTC')", @@ -404,23 +460,32 @@ defmodule MaveCore.Analytics.Sessions.Query do "if(p.7 < p.6, least(p.3, toDateTime64(addDays(toDate(p.2, 'UTC'), 1), 6, 'UTC')), p.3)", else: "p.3" + filter = if only_future, do: "AND isNotNull(future_at)", else: "" + """ - /* mave_perf:sessions.range */ - WITH filtered AS ( + SELECT arrayMap(p -> (#{key}, dateDiff('microsecond', p.2, p.3), + if(isFinite(p.8) AND p.8 > 0 AND isFinite(p.6) AND isFinite(p.7), + least(1.0, greatest(0.0, (p.6 + (p.7 - p.6) * dateDiff('microsecond', p.4, #{completion_at}) + / dateDiff('microsecond', p.4, p.5)) / p.8)), NULL)), #{rows}) AS pieces + FROM ( SELECT arrayMap(s -> (s.1, greatest(s.2, {$2:DateTime64(6)}), least(s.3, {$3:DateTime64(6)}), s.2, s.3, s.4, s.5, s.6), arrayFilter(s -> (empty({$4:Array(String)}) OR s.1 IN {$4:Array(String)}) AND s.3 > {$2:DateTime64(6)} AND s.2 < {$3:DateTime64(6)} AND s.3 > s.2, segments)) AS clipped FROM analytics_sessions PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} - ), pieces AS ( - SELECT arrayMap(p -> (#{key}, dateDiff('microsecond', p.2, p.3), - if(isFinite(p.8) AND p.8 > 0 AND isFinite(p.6) AND isFinite(p.7), - least(1.0, greatest(0.0, (p.6 + (p.7 - p.6) * dateDiff('microsecond', p.4, #{completion_at}) - / dateDiff('microsecond', p.4, p.5)) / p.8)), NULL)), #{rows}) AS pieces - FROM filtered WHERE notEmpty(clipped) + #{filter} + ) AS filtered WHERE notEmpty(clipped) + """ + end + + defp range_query(pieces, mode, table) do + """ + /* mave_perf:sessions.range */ + WITH pieces AS ( + #{pieces} ) - #{range_aggregation(mode, "analytics_sessions")} + #{range_aggregation(mode, table)} """ end @@ -491,6 +556,13 @@ defmodule MaveCore.Analytics.Sessions.Query do """ end + defp future_sessions(condition) do + """ + SELECT session_id FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} AND #{condition} + """ + end + # Wide event arrays require a byte-sized insert buffer. The default squasher # can retain hundreds of megabytes per copy, competing with source merges. defp copy_settings do diff --git a/priv/clickhouse_repo/migrations/20261023200000_add_session_future_visibility.exs b/priv/clickhouse_repo/migrations/20261023200000_add_session_future_visibility.exs new file mode 100644 index 0000000..8cfba58 --- /dev/null +++ b/priv/clickhouse_repo/migrations/20261023200000_add_session_future_visibility.exs @@ -0,0 +1,17 @@ +defmodule MaveCore.ClickHouseRepo.Migrations.AddSessionFutureVisibility do + use Ecto.Migration + + def up do + # A compact predicate avoids loading every event array when only a few + # sessions have crossed a visibility deadline. Existing writers remain valid. + execute(""" + ALTER TABLE analytics_sessions ADD COLUMN future_at Nullable(DateTime64(6)) + MATERIALIZED arrayReduce('minOrNull', + arrayMap(e -> e.2, arrayFilter(e -> e.2 >= created_at, events))) + """) + end + + def down do + execute("ALTER TABLE analytics_sessions DROP COLUMN future_at") + end +end diff --git a/test/mave_core/analytics/sessions_test.exs b/test/mave_core/analytics/sessions_test.exs index 53eac1b..df82180 100644 --- a/test/mave_core/analytics/sessions_test.exs +++ b/test/mave_core/analytics/sessions_test.exs @@ -3,6 +3,7 @@ defmodule MaveCore.Analytics.SessionsTest do alias MaveCore.{Analytics, ClickHouseRepo, Repo} alias MaveCore.Analytics.Sessions + alias MaveCore.Analytics.Sessions.Query alias MaveCore.Workers.AnalyticsSessionsWorker setup do @@ -478,18 +479,84 @@ defmodule MaveCore.Analytics.SessionsTest do end test "future-dated pauses become visible without another insert" do + [[unchanged_id]] = + ClickHouseRepo.query!(""" + SELECT number FROM numbers(2, 256) + WHERE modulo(cityHash64(toUUID(concat('00000000-0000-0000-0000-', leftPad(toString(number),12,'0')))),16)=1 + LIMIT 1 + """).rows + pair(1, "AAAAA", "normal", 20, 25, 0, 5) + pair(unchanged_id, "AAAAA", "unchanged", 20, 25, 0, 5) # Keep another ordinary session in the same bucket so an empty future # array cannot mask this session's scheduled refresh deadline. Process.put(:analytics_time, DateTime.utc_now()) pair(1, "AAAAA", "future", -15, -10, 0, 5) assert {:ok, _} = Sessions.refresh_changed(limit: 32) + {:ok, before_generations} = Sessions.generations("AAAAA") + + unchanged_sql = """ + SELECT session_id,created_at FROM analytics_sessions + WHERE generation IN {$0:Array(UUID)} + AND session_id!=toUUID('00000000-0000-0000-0000-000000000001') + """ + + unchanged = ClickHouseRepo.query!(unchanged_sql, [before_generations]).rows assert {:ok, before} = Sessions.data("AAAAA") - assert before.watch_time.year == 5000 + assert before.watch_time.year == 10_000 Process.sleep(16_000) assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + {:ok, after_generations} = Sessions.generations("AAAAA") + assert unchanged == ClickHouseRepo.query!(unchanged_sql, [after_generations]).rows assert {:ok, later} = Sessions.data("AAAAA") - assert later.watch_time.year == 10_000 + assert later.watch_time.year == 15_000 + end + + test "newly visible sessions retain exact overview, unique and range results" do + pair(1, "AAAAA", "normal", 20, 25, 0, 5) + pair(2, "AAAAA", "unchanged", 20, 25, 0, 5) + pair(1, "AAAAA", "future", -15, -10, 0, 5) + pair(3, "AAAAA", "future-only", -15, -10, 2, 7) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + {:ok, generations} = Sessions.generations("AAAAA") + + assert [[2]] = + ClickHouseRepo.query!( + "SELECT countIf(isNotNull(future_at)) FROM analytics_sessions WHERE generation IN {$0:Array(UUID)}", + [generations] + ).rows + + now = DateTime.add(Process.get(:analytics_time), 16) + date = DateTime.to_date(now) + year = DateTime.new!(Date.new!(date.year, 1, 1), ~T[00:00:00]) + month = DateTime.new!(Date.new!(date.year, date.month, 1), ~T[00:00:00]) + day = DateTime.new!(date, ~T[00:00:00]) + params = ["AAAAA", generations, year, month, day, now] + + assert Enum.sort(ClickHouseRepo.query!(Query.live_overview(), params).rows) == + Enum.sort(ClickHouseRepo.query!(Query.overview_at(), params).rows) + + assert ClickHouseRepo.query!(Query.live_unique_overview(), params).rows == + ClickHouseRepo.query!(Query.unique_watch_time(), params).rows + + assert {:ok, actual} = Sessions.data("AAAAA", now: now, include_unique: true) + assert actual.watch_time.year == 20_000 + assert actual.unique_watch_time.year == 15_000 + + for mode <- [:series, :videos], granularity <- [:day, :week, :month] do + range_params = ["AAAAA", generations, year, now, []] + detailed = ClickHouseRepo.query!(Query.range(granularity, mode), range_params).rows + hybrid = ClickHouseRepo.query!(Query.live_range(granularity, mode), range_params).rows + assert Enum.sort(hybrid) == Enum.sort(detailed) + + expected = + Map.new(detailed, fn [key, views, watched, completion] -> + {key, %{views: views, watch_seconds: watched, completion_rate: completion}} + end) + + assert {:ok, ^expected} = + Sessions.range_data("AAAAA", year, now, mode: mode, granularity: granularity) + end end test "cleanup retains published generations and removes only expired ones" do From 4df5963be3b3bf041cc567c0805b2c3ec3a2ef10 Mon Sep 17 00:00:00 2001 From: David Date: Fri, 23 Oct 2026 21:54:13 +0200 Subject: [PATCH 2/2] Preserve exact per-video completion across midnight --- docs/analytics_sessions.md | 2 +- lib/mave_core/analytics/sessions.ex | 8 +++--- lib/mave_core/analytics/sessions/query.ex | 30 +++++++++------------- test/mave_core/analytics/sessions_test.exs | 22 ++++++++++++++-- 4 files changed, 38 insertions(+), 24 deletions(-) diff --git a/docs/analytics_sessions.md b/docs/analytics_sessions.md index 0fc6a7e..94a1670 100644 --- a/docs/analytics_sessions.md +++ b/docs/analytics_sessions.md @@ -103,7 +103,7 @@ Future-dated events are excluded from summaries until their time arrives; the published metadata records a refresh deadline so they become visible without requiring another insert. A materialized per-session deadline also lets the processor refresh only those sessions. Between refreshes, current overview and -date-aligned report queries combine compact rows with detailed calculations for +date-aligned time-series queries combine compact rows with detailed calculations for the affected sessions; one newly visible event does not force a detailed scan of the entire space. Historical and partial-day reads retain precise segment calculations. diff --git a/lib/mave_core/analytics/sessions.ex b/lib/mave_core/analytics/sessions.ex index fa6eca4..6afd2e4 100644 --- a/lib/mave_core/analytics/sessions.ex +++ b/lib/mave_core/analytics/sessions.ex @@ -116,10 +116,12 @@ defmodule MaveCore.Analytics.Sessions do def range_data(_space_hash, _from, _to, _opts), do: {:error, :invalid_range} defp range_query(space, generations, from, to, granularity, mode) do - if midnight?(from) and reports_complete?(space, generations) do + # Daily completion maxima preserve series semantics. Per-video completion + # still needs segment endpoints, especially for rewinds across midnight. + if mode == :series and midnight?(from) and reports_complete?(space, generations) do cond do - midnight?(to) -> Query.report_range(granularity, mode) - reports_cover_time?(space, generations, to) -> Query.live_range(granularity, mode) + midnight?(to) -> Query.report_range(granularity) + reports_cover_time?(space, generations, to) -> Query.live_range(granularity) true -> Query.range(granularity, mode) end else diff --git a/lib/mave_core/analytics/sessions/query.ex b/lib/mave_core/analytics/sessions/query.ex index 3f17444..737f68b 100644 --- a/lib/mave_core/analytics/sessions/query.ex +++ b/lib/mave_core/analytics/sessions/query.ex @@ -111,31 +111,25 @@ defmodule MaveCore.Analytics.Sessions.Query do "SELECT countIf(complete = 0) FROM analytics_session_reports PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)}" end - def report_range(granularity, mode \\ :series) - when granularity in [:day, :week, :month] and mode in [:series, :videos] do - range_query(report_pieces(granularity, mode), mode, "analytics_session_reports") + def report_range(granularity) when granularity in [:day, :week, :month] do + range_query(report_pieces(granularity), :series, "analytics_session_reports") end - def live_range(granularity, mode) - when granularity in [:day, :week, :month] and mode in [:series, :videos] do + def live_range(granularity) when granularity in [:day, :week, :month] do pieces = - report_pieces(granularity, mode, true) <> - "\nUNION ALL\n" <> range_pieces(granularity, mode, true) + report_pieces(granularity, true) <> + "\nUNION ALL\n" <> range_pieces(granularity, :series, true) - range_query(pieces, mode, "analytics_sessions") + range_query(pieces, :series, "analytics_sessions") end - defp report_pieces(granularity, mode, exclude_future \\ false) do + defp report_pieces(granularity, exclude_future \\ false) do key = - if mode == :videos do - "p.1" - else - %{ - day: "toString(p.2)", - week: "toString(toMonday(p.2))", - month: "toString(toStartOfMonth(p.2))" - }[granularity] - end + %{ + day: "toString(p.2)", + week: "toString(toMonday(p.2))", + month: "toString(toStartOfMonth(p.2))" + }[granularity] filter = if exclude_future, diff --git a/test/mave_core/analytics/sessions_test.exs b/test/mave_core/analytics/sessions_test.exs index df82180..7ce3541 100644 --- a/test/mave_core/analytics/sessions_test.exs +++ b/test/mave_core/analytics/sessions_test.exs @@ -546,8 +546,11 @@ defmodule MaveCore.Analytics.SessionsTest do for mode <- [:series, :videos], granularity <- [:day, :week, :month] do range_params = ["AAAAA", generations, year, now, []] detailed = ClickHouseRepo.query!(Query.range(granularity, mode), range_params).rows - hybrid = ClickHouseRepo.query!(Query.live_range(granularity, mode), range_params).rows - assert Enum.sort(hybrid) == Enum.sort(detailed) + + if mode == :series do + hybrid = ClickHouseRepo.query!(Query.live_range(granularity), range_params).rows + assert Enum.sort(hybrid) == Enum.sort(detailed) + end expected = Map.new(detailed, fn [key, views, watched, completion] -> @@ -559,6 +562,21 @@ defmodule MaveCore.Analytics.SessionsTest do end end + test "per-video completion retains rewind endpoints across midnight" do + from = DateTime.new!(Date.add(Date.utc_today(), -2), ~T[00:00:00]) + to = DateTime.add(from, 2 * 86_400) + Process.put(:analytics_time, DateTime.add(from, 86_400)) + pair(1, "AAAAA", "rewind", -2, 2, 8, 2) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + assert {:ok, %{"rewind" => result}} = + Sessions.range_data("AAAAA", from, to, mode: :videos) + + assert result.views == 1 + assert result.watch_seconds == 4.0 + assert_in_delta result.completion_rate, 2 / 60, 1.0e-10 + end + test "cleanup retains published generations and removes only expired ones" do pair(1, "AAAAA", "first", 20, 25, 0, 5) assert {:ok, _} = Sessions.refresh_changed(limit: 32)