diff --git a/.github/scripts/install-media-deps.sh b/.github/scripts/install-media-deps.sh new file mode 100644 index 0000000..9c120b5 --- /dev/null +++ b/.github/scripts/install-media-deps.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +set -euo pipefail + +if [[ "$#" -eq 0 ]]; then + echo 'Specify the media test dependencies' >&2 + exit 1 +fi + +# The hosted-runner mirror can stall before tests even start. Use the signed +# public Ubuntu archive and avoid unrelated recommended desktop packages. +if [[ -f /etc/apt/apt-mirrors.txt ]]; then + printf '%s\n' 'http://archive.ubuntu.com/ubuntu/' | sudo tee /etc/apt/apt-mirrors.txt >/dev/null +fi +sudo apt-get -o Acquire::Retries=3 update +sudo apt-get -o Acquire::Retries=3 install --yes --no-install-recommends "$@" diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml index dc88314..ef77bc0 100644 --- a/.github/workflows/security.yml +++ b/.github/workflows/security.yml @@ -24,13 +24,13 @@ concurrency: jobs: media-sandbox: runs-on: ubuntu-24.04 - timeout-minutes: 10 + timeout-minutes: 20 steps: - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: persist-credentials: false - name: Install sandbox test dependencies - run: sudo apt-get update && sudo apt-get install --yes build-essential libseccomp-dev ffmpeg python3 + run: bash .github/scripts/install-media-deps.sh build-essential libseccomp-dev ffmpeg python3 - name: Build and test the shared streaming broker run: | artifacts=$(mktemp -d) @@ -120,8 +120,7 @@ jobs: key: test-${{ runner.os }}-otp29.1.1-elixir1.20.4-${{ hashFiles('mix.lock') }} - name: Prepare disposable media dependencies run: | - sudo apt-get update - sudo apt-get install --yes ffmpeg build-essential + bash .github/scripts/install-media-deps.sh ffmpeg build-essential docker build --pull --target server --tag mave-ci-minio deploy/minio docker run --detach --name mave-ci-minio --publish 127.0.0.1:9010:9000 \ --env MINIO_ROOT_USER=minioadmin --env MINIO_ROOT_PASSWORD=minioadmin \ diff --git a/README.md b/README.md index 614eba2..2e41d77 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,10 @@ local MinIO support and deployment adapter integration. register it in [StepRegistry](lib/mave_core/flow/step_registry.ex), and use it in your pipeline. New step implementations require rebuilding Core; see [contributing](CONTRIBUTING.md) for local development and tests. +- **Analytics at scale:** prepare playback sessions in bounded background batches, + then enable the reader for selected spaces. See + [prepared analytics operations](docs/analytics_sessions.md) for backfill, + activation, resource budgets and capacity checks. ## Prefer managed hosting? diff --git a/config/config.exs b/config/config.exs index 235c888..7112c18 100644 --- a/config/config.exs +++ b/config/config.exs @@ -37,6 +37,8 @@ config :mave_core, Oban, {Oban.Plugins.Cron, crontab: [ {"* * * * *", MaveCore.Workers.FlowRecoveryWorker}, + {"* * * * *", MaveCore.Workers.AnalyticsSessionsWorker}, + {"* * * * *", MaveCore.Workers.AnalyticsSessionsCleanupWorker}, {"23 * * * *", MaveCore.Workers.CliAuthorizationPruneWorker} ]} ], @@ -49,6 +51,7 @@ config :mave_core, Oban, flow_media: 10, flow_imports: 10, flow_low: 5, + analytics: 1, webhooks: 20 ] @@ -253,3 +256,6 @@ config :mime, :types, %{ # Import environment specific config. This must remain at the bottom # of this file so it overrides the configuration defined above. import_config "#{config_env()}.exs" + +# Opt-in until the historical session backfill has been verified. +config :mave_core, :analytics_sessions, enabled: false diff --git a/config/runtime.exs b/config/runtime.exs index 018ef35..44ef42f 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -729,3 +729,8 @@ if config_env() == :prod do System.get_env("UA_INSPECTOR_DATABASE_PATH") || Application.app_dir(:mave_core, "priv/ua_inspector") end + +# Background preparation and reader activation can be rolled out independently. +config :mave_core, + :analytics_sessions, + MaveCore.Analytics.Sessions.Config.from_env!(System.get_env()) diff --git a/config/test.exs b/config/test.exs index a22ff9a..89922d5 100644 --- a/config/test.exs +++ b/config/test.exs @@ -16,7 +16,7 @@ config :mave_core, MaveCore.Repo, password: "postgres", hostname: postgres_host, port: String.to_integer(postgres_port), - database: "mave_core_test", + database: System.get_env("MAVE_TEST_POSTGRES_DATABASE") || "mave_core_test", pool: Ecto.Adapters.SQL.Sandbox, pool_size: 10 @@ -57,8 +57,8 @@ config :phoenix_live_view, :test_warnings, missing_form_id: :raise # Sort query params output of verified routes for robust url comparisons config :mave_core, MaveCore.ClickHouseRepo, hostname: System.get_env("CLICKHOUSE_HOST") || "localhost", - port: 8123, - database: "mave_metrics_test", + port: String.to_integer(System.get_env("CLICKHOUSE_PORT") || "8123"), + database: System.get_env("MAVE_TEST_CLICKHOUSE_DATABASE") || "mave_metrics_test", username: "default", password: "password", scheme: "http", diff --git a/docs/analytics_sessions.md b/docs/analytics_sessions.md new file mode 100644 index 0000000..1aa886e --- /dev/null +++ b/docs/analytics_sessions.md @@ -0,0 +1,242 @@ +# Prepared analytics sessions + +Prepared sessions separate event ingestion from dashboard query work. The raw +`events` table remains authoritative and is not deleted or rewritten by the +session processor. Enable the reader/worker with `MAVE_ANALYTICS_SESSIONS=true` +after migrations and the initial backfill have been verified. It defaults off. Query thread and memory ceilings can be configured for the +installation, for example: + +```elixir +config :mave_core, :analytics_sessions, + enabled: true, + query_threads: 2, + build_memory_bytes: 1_073_741_824, + read_memory_bytes: 536_870_912 +``` + +Readers and background processing can be switched independently. An unset +`processor_enabled` follows `enabled` for compatibility. Set it explicitly to +prepare/catch up while readers still use the original implementation, or to +pause processing while retaining the last complete prepared results. +Scheduled cleanup runs while either the reader or processor is enabled; pausing +the processor stops rebuilds, not reclamation of superseded generations. + +```elixir +config :mave_core, :analytics_sessions, + enabled: true, + processor_enabled: true, + read_spaces: ["AAAAA", "BBBBB"] +``` + +`read_spaces` defaults to `:all`. It only selects the prepared reader; other +spaces retain the original reader, and the processor can still prepare all +spaces. An empty list selects no prepared readers. It does not alter access +control. Direct `Sessions` calls remain explicit prepared-data operations. + +The worker can be restricted separately with `process_spaces: ["AAAAA"]`, or +`MAVE_ANALYTICS_PROCESS_SPACES`. It defaults to all spaces, accepts `*` for all +and an empty value for none. Use this during a capacity-limited rollout so the +worker does not begin preparing the entire installation. Explicit per-space +maintenance/backfill calls remain independent of this worker setting. + +The corresponding runtime variables are `MAVE_ANALYTICS_SESSIONS`, +`MAVE_ANALYTICS_PROCESSOR_ENABLED`, `MAVE_ANALYTICS_READ_SPACES` (comma-separated +hashes, `*` for all, empty for none), `MAVE_ANALYTICS_BUILD_MEMORY_BYTES`, +`MAVE_ANALYTICS_READ_MEMORY_BYTES`, `MAVE_ANALYTICS_QUERY_THREADS`, +`MAVE_ANALYTICS_REFRESH_BUCKET_LIMIT` (default 4), and +`MAVE_ANALYTICS_CLEANUP_SPACE_LIMIT` (default 32), +`MAVE_ANALYTICS_CLEANUP_GENERATION_LIMIT` (default 64 per space/round), and +`MAVE_ANALYTICS_READER_GRACE_SECONDS` (default 300; minimum 120). +Invalid values fail startup rather than silently changing the activation scope. + +Limits remain bounded by ClickHouse's own server settings; increasing these +values does not increase available server memory. + +## Storage and publication + +`analytics_sessions` stores one row per space/session in an immutable generation. +It retains ordered events and paired play/pause segments, including video IDs, +positions, duration and source dimensions. A session shared by multiple videos +stays in the same bucket. This preserves legacy space-level counting while +allowing per-video reports to group separately. + +The build records the source window's event ordinal before collecting arrays, +so equal timestamps retain the playback pairing used by the legacy window +queries. Timestamp ties have no independent event ID in the source schema; +the prepared layer does not invent a new ordering policy for them. Large +per-video arrays are combined element by element to avoid the aggregate +function's nested-array element size limit. + +`analytics_session_years` stores compact per-session calendar-year totals, +first-event source dimensions and watch totals combined per day for the common +overview. Repeated same-day segments occupy one day entry while retaining the +original segment-start attribution and per-session view thresholds. It also +stores deduplicated integer watch intervals per day for the common unique-watch +query. Deduplication preserves the earliest-day attribution across periods; +historical within-day reads retain the original segment calculation. Copying +an older generation combines its day entries, and format upgrades rebuild the +compact year summaries from stored sessions without reordering raw events. + +`analytics_session_reports` stores compact per-session/video/day watch totals +and completion maxima. Calendar-aligned period series use these summaries, +including a current partial day only when the snapshot has no future events. +Historical partial-day boundaries and per-video mode use bounded segment +queries. Sessions containing intervals longer than the supported report window +are marked incomplete and use the segment path instead of expanding unbounded +day arrays. Version-3 snapshots can gain reports by reusing their existing +session data. All three tables are built before a generation is published. Date-range +queries retain the full segment representation; they do not sum daily view +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 +`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 +uses a new generation, so partial inserts cannot double-count published data. +Initial readers wait for all sixteen bucket pointers, including empty buckets. +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. + +The `analytics_session_inventory` projection summarizes source row counts and +field fingerprints. It lives inside the raw MergeTree table, so it follows +normal inserts and source mutations without an application-side dual write. +Late events are detected by their arrival in the source, not by their event +timestamp. A snapshot records the count/fingerprint of the exact rows read by +its build; changes racing with that build remain different at the next poll. +Identical replacement rows need no rebuild; same-count changes to event fields +change the fingerprint. As with any non-cryptographic checksum, collisions are +theoretically possible. + +The `analytics_session_revisions` projection stores the same aggregate +revision per session. It identifies exactly which sessions need recalculation +without sorting all historical events again. These projections contain +aggregate metadata; they do not reorder the raw event reads. This matters for +legacy events with identical timestamps and no independent sequence number. +Both projections are maintained by ClickHouse inserts and source mutations. + +The projections are optimizations, not correctness dependencies. Existing parts +without them are scanned until they are explicitly materialized. Initial migration +does not perform an unbounded historical backfill during deployment. + +## Operation + +Run commands from the hosting application (for example the SaaS application): + +```sh +mix ecto.migrate +mix analytics.sessions.backfill --materialize-inventory --materialize-session-revisions --limit 16 --rounds 100 +# Optional: prepare one tenant without changing the reader activation scope. +mix analytics.sessions.backfill --space AAAAA --limit 1 --rounds 16 +``` + +Apply both the PostgreSQL and ClickHouse migrations through the installation's +usual migration command. The optional projection materialization can take time +and disk space; run it as an explicit maintenance/backfill step. Backfill can be +stopped and resumed. The command publishes at most `--limit` buckets per round +and stops when caught up or when `--rounds` is exhausted. It does not enable the +feature or alter event retention. + +Run only one backfill coordinator at a time. Pause the scheduled processor while +running a manual backfill if aggregate background concurrency must stay at one. +The space option bounds rebuilds, but inventory discovery can still read the +whole source until the projection is materialized. Do not include projection +materialization or historical backfill in application startup or deployment +health checks. + +The minute worker runs on the `analytics` queue with one job per node and a +distributed Oban uniqueness constraint while active. It prioritizes unpublished +and oldest buckets. Source changes coalesce into bucket rebuilds rather than +creating one job per event. The default round handles four buckets; tune the +configurable batch limit against measured build duration and freshness targets. +Copying unchanged rows uses a single stream and bounded byte-sized blocks, +with a separate 512 MiB maximum within the configured build limit. This leaves +more headroom for raw-event merges. Very large individual rows still need to +fit in the copy budget; failure preserves the published generation. +Keep rounds short enough that cleanup on the same queue can run regularly. + +The minute cleanup worker visits up to 32 spaces per round, oldest first. It +collects eligible retired generations per space. Each immutable generation has +its own partition (keyed by space and generation), so cleanup drops whole +partitions instead of rewriting surviving rows. Different generations never +merge together. Automatic merges are disabled on these three immutable tables; +a generation has a bounded set of insert parts and is retired as a unit. This +avoids background merge amplification during repeated refreshes. PostgreSQL records when a version +was replaced; the five-minute default reader grace period starts then, not when the version +was originally built. Discovery gives unpublished crash remnants a fresh grace +period. Active generations are excluded again under the publication lock. +Deletion must succeed on all three tables before retirement records are removed; +failures are safe to retry. ClickHouse can retain inactive files after a drop; +measure physical bytes as well as active rows when planning disk headroom. +The number of partitions grows with nonempty buckets and retained generations; +include that metadata cost when testing installations with many spaces. +Prepared read queries have at most 20-second client +deadlines; consumers of `generations/1` must use the returned pointers promptly, +within the configured grace period. Long-running exports need their own pinned +snapshot or a larger configured grace period. More retained generations consume +more disk; do not increase the grace period without measuring storage growth. + +Raw source deletion/copy callers must call `Sessions.invalidate/1` under their +existing lifecycle authorization after changing the source. This hides prepared +data immediately and removes its generations; remaining source data rebuilds +from the next inventory. In particular, retired tenant identifiers must not be +reused before invalidation succeeds. Normal event arrival needs no callback. + +## Query interfaces + +- `Sessions.data(hash)` returns the dashboard fields without calculating unique + watch time. `include_unique: true` adds the existing API field. +- `Sessions.range_data(hash, from, to, granularity: :day | :week | :month)` returns + a map keyed by date strings, with `""` holding whole-period totals. `to` is + exclusive; the caller converts an inclusive last date to next-day midnight. +- `mode: :videos, embed_hashes: hashes` returns per-video totals for a bounded + export batch of at most 250 hashes. Empty hashes select all videos. +- `Sessions.generations(hash)` exposes the immutable generation list for shared + future reports. Every query must also constrain `space_hash`. + +Range metrics preserve sessions across midnight and clip segments at range +boundaries. Whole-period views are calculated independently of daily views; +they must not be obtained by adding daily counts. Completion excludes unknown +durations. Legacy overview queries retain their existing year/date attribution. +Unique watch time unions watched integer intervals instead of allocating one +row per watched second, retaining earliest-observation attribution. + +An incomplete initial backfill returns `:preparing`; database failures return +`:analytics_unavailable`. The API reports 503 with Retry-After; the dashboard +shows preparation state and polls again. Neither substitutes fabricated zero +metrics for an unavailable prepared result. + +## Analytics feature integration + +Top-list and period-overview features should consume the prepared generation +and segment interfaces instead of adding `leadInFrame` scans of raw events. +Date-range/CSV code can use `range_data/4`; retain its existing metadata joins, +authorization, output escaping, and bounded export pagination. Device/browser +and additional source panels should share a session aggregation in one query. +The stored tuples include OS, source URL, duration and video positions for +engagement, completion and website reports. This does not merge those feature +branches or change their public API contracts. + +Restoring raw events without the matching prepared data requires invalidating +the published metadata and rerunning the backfill. Prepared data is derived and +can always be rebuilt from the raw source. + +Publication remains bucket-scoped, but normal recomputation is session-scoped. +Unchanged event arrays and year summaries are copied without sorting again; +changed sessions retain the same raw-event pairing query as full backfills. +Large change sets and due future events use the full-bucket fallback. Measure +both paths, including very large individual sessions. + +Measure sustained ingestion and reclamation, not only isolated read latency. +Repeated immutable generations temporarily amplify storage even when their +contents are reused. Cleanup and processing have independent bounded budgets; +measure complete sweep time, inactive parts and retained generations at the +intended tenant count. Do not enable broadly when processing lag or physical +storage grows without bound. Deployment-specific budgets and rollout decisions +belong to the hosting application. diff --git a/lib/mave_core/analytics.ex b/lib/mave_core/analytics.ex index ec1406f..81e4507 100644 --- a/lib/mave_core/analytics.ex +++ b/lib/mave_core/analytics.ex @@ -2,6 +2,7 @@ defmodule MaveCore.Analytics do @moduledoc false alias Ecto.Adapters.SQL + alias MaveCore.Analytics.Sessions alias MaveCore.ClickHouseRepo alias MaveCore.SharedStorageSpace @@ -15,13 +16,25 @@ defmodule MaveCore.Analytics do if SharedStorageSpace.hash?(space_hash) do {:ok, empty_space_data(space_hash)} else - do_space_data(space_hash) + space_metrics(space_hash) end end end def space_data(_), do: {:error, :invalid_space_hash} + defp space_metrics(space_hash) do + if Sessions.enabled?(space_hash), + do: Sessions.data(space_hash, include_unique: true), + else: do_space_data(space_hash) + end + + defp monthly_metrics(space_hash) do + if Sessions.enabled?(space_hash), + do: Sessions.monthly_watch_time(space_hash), + else: do_space_monthly_watch_time(space_hash) + end + defp do_space_data(space_hash) do now = DateTime.utc_now() @@ -98,12 +111,19 @@ defmodule MaveCore.Analytics do }} end + @doc "Dashboard totals without the optional unique-watch calculation." + def space_overview(space_hash) do + if Sessions.enabled?(space_hash), + do: Sessions.data(space_hash), + else: space_data(space_hash) + 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 {:ok, %{watch_time: %{month: 0}}} else - do_space_monthly_watch_time(space_hash) + monthly_metrics(space_hash) end end end diff --git a/lib/mave_core/analytics/sessions.ex b/lib/mave_core/analytics/sessions.ex new file mode 100644 index 0000000..9fb9bb9 --- /dev/null +++ b/lib/mave_core/analytics/sessions.ex @@ -0,0 +1,686 @@ +defmodule MaveCore.Analytics.Sessions do + @moduledoc """ + Versioned playback sessions, rebuilt in bounded buckets after source changes. + + PostgreSQL publishes the active generation only after its ClickHouse insert + completes. Readers keep using the previous complete generation during refresh. + Inventory revisions are measured from the rows actually used by the build, so + an insert racing with a refresh is picked up by the following inventory poll. + """ + alias MaveCore.Analytics.Sessions.Query + alias MaveCore.{ClickHouseRepo, Repo, SharedStorageSpace} + + def enabled?, do: Application.get_env(:mave_core, :analytics_sessions, [])[:enabled] == true + + def enabled?(space_hash) do + config = Application.get_env(:mave_core, :analytics_sessions, []) + spaces = Keyword.get(config, :read_spaces, :all) + enabled?() and (spaces == :all or (is_list(spaces) and space_hash in spaces)) + end + + def processing_enabled? do + config = Application.get_env(:mave_core, :analytics_sessions, []) + Keyword.get(config, :processor_enabled, enabled?()) == true + end + + def data(space_hash, opts \\ []) do + with :ok <- validate_hash(space_hash) do + if SharedStorageSpace.hash?(space_hash) do + {:ok, empty(space_hash)} + else + read_data(space_hash, opts) + end + end + end + + defp read_data(space_hash, opts) do + now = Keyword.get(opts, :now, DateTime.utc_now()) + + with {:ok, {generations, latest, due}} <- read_snapshot(space_hash), + params = params(space_hash, generations, now), + {query, unique_query} = overview_query(now, latest, due), + {:ok, %{rows: rows}} <- ch(query, params, 15_000), + {:ok, unique} <- maybe_unique(params, opts, unique_query) do + {:ok, rows |> map_overview(space_hash) |> Map.put(:unique_watch_time, unique)} + else + {:empty, _} -> {:ok, empty(space_hash)} + {:error, :preparing} -> {:error, :preparing} + _ -> {:error, :analytics_unavailable} + end + end + + def monthly_watch_time(space_hash) do + now = DateTime.utc_now() + date = DateTime.to_date(now) + from = DateTime.new!(Date.new!(date.year, date.month, 1), ~T[00:00:00]) + + with {:ok, generations} <- generations(space_hash), + {:ok, %{rows: [[value]]}} <- + ch(Query.monthly_watch_time(), [space_hash, generations, from, now], 6_000) do + {:ok, %{watch_time: %{month: value}}} + else + {:empty, _} -> {:ok, %{watch_time: %{month: 0}}} + {:error, :preparing} -> {:error, :preparing} + _ -> {:error, :analytics_unavailable} + end + end + + defp maybe_unique(params, opts, query) do + if Keyword.get(opts, :include_unique, false) do + case ch(query, params, 20_000) do + {:ok, %{rows: [[today, month, year]]}} -> + {:ok, %{today: today, month: month, year: year}} + + _ -> + {:error, :analytics_unavailable} + end + else + {:ok, nil} + end + end + + @doc """ + Bounded UTC range totals and buckets from prepared segments, never raw events. + `to` is exclusive. Use `mode: :videos` for CSV batches/top lists and + `embed_hashes: [...]` to restrict the work to a set of videos. + """ + def range_data(space_hash, from, to, opts \\ []) + + def range_data(space_hash, %DateTime{} = from, %DateTime{} = to, opts) do + granularity = Keyword.get(opts, :granularity, :day) + mode = Keyword.get(opts, :mode, :series) + hashes = Keyword.get(opts, :embed_hashes, []) + + with :ok <- validate_hash(space_hash), + true <- granularity in [:day, :week, :month] and mode in [:series, :videos], + true <- DateTime.diff(to, from) in 1..(732 * 86_400), + true <- + is_list(hashes) and length(hashes) <= 250 and + Enum.all?(hashes, &(is_binary(&1) and byte_size(&1) in 1..128)), + {:ok, generations} <- generations(space_hash), + sql = range_query(space_hash, generations, from, to, granularity, mode), + {:ok, %{rows: rows}} <- + ch(sql, [space_hash, generations, from, to, hashes], 20_000) do + {:ok, + Map.new(rows, fn [key, views, watched, completion] -> + {key, %{views: views, watch_seconds: watched, completion_rate: completion}} + end)} + else + {:empty, _} -> {:ok, %{}} + false -> {:error, :invalid_range} + {:error, :preparing} -> {:error, :preparing} + _ -> {:error, :analytics_unavailable} + end + end + + 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) + end + + defp midnight?(at), do: rem(DateTime.to_unix(at, :microsecond), 86_400_000_000) == 0 + + defp reports_cover_time?(space, generations, to) do + case Repo.query( + "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]]}} -> + count == Query.buckets() and + (is_nil(latest) or NaiveDateTime.compare(naive(to), latest) != :lt) + + _ -> + false + end + end + + defp reports_complete?(space, generations) do + case ch(Query.report_coverage(), [space, generations], 5_000) do + {:ok, %{rows: [[0]]}} -> true + _ -> false + end + end + + @doc "Active immutable generations for custom reports on the session table." + def generations(space_hash) do + if SharedStorageSpace.hash?(space_hash), + do: {:empty, space_hash}, + else: do_generations(space_hash) + end + + defp do_generations(space_hash) do + with :ok <- validate_hash(space_hash), + {:ok, %{rows: rows}} <- + Repo.query( + "SELECT generation::text FROM analytics_session_buckets WHERE space_hash = $1 AND format_version = $2 ORDER BY bucket", + [space_hash, Query.version()] + ) do + generation_result(space_hash, rows) + end + end + + defp generation_result(space, rows) do + if length(rows) == Query.buckets(), + do: {:ok, List.flatten(rows)}, + else: source_readiness(space) + end + + defp source_readiness(space) do + case ch("SELECT 1 FROM events PREWHERE space_hash = {$0:String} LIMIT 1", [space]) do + {:ok, %{rows: []}} -> {:empty, space} + {:ok, _} -> {:error, :preparing} + _ -> {:error, :analytics_unavailable} + end + end + + @doc "Refresh changed buckets. Safe to retry after partial execution or a crash." + def refresh_changed(opts \\ []) do + limit = Keyword.get(opts, :limit, setting(:refresh_bucket_limit, 4)) + target = Keyword.get(opts, :space_hash) + scope = Keyword.get(opts, :spaces, :all) + + with :ok <- validate_optional_hash(target), + :ok <- validate_scope(scope), + {:ok, %{rows: inventory}} <- ch(Query.inventory(), [], 30_000), + {:ok, %{rows: published}} <- + Repo.query( + "SELECT space_hash, bucket, source_rows, fingerprint, updated_at, format_version, refresh_after FROM analytics_session_buckets" + ) do + current = + published + |> Enum.filter(fn [_, _, _, _, _, version, _] -> version == Query.version() end) + |> Map.new(fn [space, bucket, count, hash, _, _, _] -> + {{space, bucket}, {count, hash}} + end) + + ages = + Map.new(published, fn [space, bucket, _, _, updated, _, _] -> + {{space, bucket}, to_string(updated)} + end) + + due = + Map.new(published, fn [space, bucket, _, _, _, _, at] -> + {{space, bucket}, refresh_due?(at)} + end) + + source = + Map.new(inventory, fn [space, bucket, count, hash] -> {{space, bucket}, {count, hash}} end) + + spaces = Enum.uniq(Enum.map(Map.keys(source), &elem(&1, 0)) ++ Enum.map(published, &hd/1)) + spaces = Enum.filter(spaces, &(validate_hash(&1) == :ok)) + spaces = if target, do: Enum.filter(spaces, &(&1 == target)), else: spaces + spaces = if scope == :all, do: spaces, else: Enum.filter(spaces, &(&1 in scope)) + + changes = + for space <- Enum.sort(spaces), + not SharedStorageSpace.hash?(space), + bucket <- 0..(Query.buckets() - 1), + revision = Map.get(source, {space, bucket}, {0, "0"}), + Map.get(current, {space, bucket}) != revision or Map.get(due, {space, bucket}, false), + do: {space, bucket, revision} + + {count, errors} = + changes + |> Enum.sort_by(fn {space, bucket, _} -> + {Map.get(ages, {space, bucket}, ""), space, bucket} + end) + |> Enum.take(limit) + |> Enum.reduce({0, 0}, &rebuild_one/2) + + if errors == 0, do: {:ok, count}, else: {:error, {:buckets_failed, errors}} + end + end + + defp validate_scope(:all), do: :ok + + defp validate_scope(spaces) when is_list(spaces) do + if Enum.all?(spaces, &(validate_hash(&1) == :ok)), + do: :ok, + else: {:error, :invalid_space_hash} + end + + defp validate_scope(_spaces), do: {:error, :invalid_space_hash} + + defp rebuild_one({space, bucket, revision}, {count, errors}) do + case refresh_bucket(space, bucket, revision) do + {:ok, _} -> {count + 1, errors} + {:error, _} -> {count, errors + 1} + end + end + + @doc false + def refresh_bucket(space_hash, bucket, revision) do + with :ok <- validate_hash(space_hash), true <- bucket in 0..(Query.buckets() - 1) do + Repo.transaction(fn -> refresh_locked(space_hash, bucket, revision) end, timeout: 180_000) + else + _ -> {:error, :invalid_bucket} + end + end + + defp refresh_locked(space, bucket, revision) do + lock!(space) + + existing = + Repo.query!( + "SELECT source_rows, fingerprint, format_version, refresh_after FROM analytics_session_buckets WHERE space_hash = $1 AND bucket = $2", + [space, bucket] + ).rows + + if unchanged?(existing, revision), do: :unchanged, else: build_bucket!(space, bucket) + end + + defp build_bucket!(space_hash, bucket) do + generation = Ecto.UUID.generate() + {previous, changed, full, report_full, overview_full} = build_plan!(space_hash, bucket) + params = [space_hash, bucket, generation, changed, full] + + with :ok <- + copy_previous( + space_hash, + bucket, + generation, + previous, + changed, + report_full, + overview_full + ), + {:ok, _} <- ch(Query.build(), params, 90_000), + {:ok, _} <- + ch( + Query.build_overview(), + [space_hash, bucket, generation, changed, overview_full], + 90_000 + ), + {:ok, _} <- + ch( + Query.build_reports(), + [space_hash, bucket, generation, changed, report_full], + 90_000 + ), + {:ok, %{rows: [[count, fingerprint, refresh_after, latest]]}} <- + ch(Query.revision(), Enum.take(params, 3)) do + retire_previous!(space_hash, bucket) + + # Publication is the only mutable pointer. A failed build leaves this row + # untouched, and a later attempt uses a different generation UUID. + Repo.query!( + """ + INSERT INTO analytics_session_buckets + (space_hash, bucket, generation, source_rows, fingerprint, format_version, refresh_after, latest_event_at, updated_at) + VALUES ($1, $2, $3::text::uuid, $4, $5, $6, $7, $8, now()) + ON CONFLICT (space_hash, bucket) DO UPDATE SET + generation = EXCLUDED.generation, source_rows = EXCLUDED.source_rows, + fingerprint = EXCLUDED.fingerprint, format_version = EXCLUDED.format_version, + refresh_after = EXCLUDED.refresh_after, latest_event_at = EXCLUDED.latest_event_at, updated_at = EXCLUDED.updated_at + """, + [ + space_hash, + bucket, + generation, + count, + fingerprint, + Query.version(), + naive(refresh_after), + naive(latest) + ] + ) + + generation + else + _ -> + discard_generation(space_hash, generation) + Repo.rollback(:analytics_unavailable) + end + end + + defp build_plan!(space, bucket) do + case Repo.query!( + "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, changed, full} = changed_plan!(space, bucket, previous) + + {previous, changed, full, report_refresh_mode(version, full), + overview_refresh_mode(version, full)} + else + {nil, [], 1, 1, 1} + end + + [] -> + {nil, [], 1, 1, 1} + end + end + + defp overview_refresh_mode(version, _full) when version in [3, 4], do: 1 + defp overview_refresh_mode(_version, full), do: full + + defp report_refresh_mode(3, _full), do: 1 + defp report_refresh_mode(_version, full), do: full + + defp changed_plan!(space, bucket, previous) do + case ch(Query.changed_sessions(), [space, bucket, previous], 90_000) do + {:ok, %{rows: rows}} when length(rows) <= 50_000 -> + {previous, Enum.map(rows, fn [id] -> uuid_string(id) end), 0} + + {:ok, _} -> + {nil, [], 1} + + _ -> + Repo.rollback(:analytics_unavailable) + end + end + + defp uuid_string(<<_::128>> = value) do + {:ok, uuid} = Ecto.UUID.load(value) + uuid + end + + defp uuid_string(value), do: value + + defp copy_previous(_space, _bucket, _generation, nil, _changed, _report_full, _overview_full), + do: :ok + + defp copy_previous(space, bucket, generation, previous, changed, report_full, overview_full) do + params = [space, bucket, generation, previous, changed] + + with {:ok, _} <- ch(Query.copy_sessions(), params, 90_000), + :ok <- copy_overview(params, overview_full), + do: copy_reports(params, report_full) + end + + defp copy_overview(_params, 1), do: :ok + + defp copy_overview(params, 0) do + case ch(Query.copy_overview(), params, 90_000) do + {:ok, _} -> :ok + error -> error + end + end + + defp copy_reports(_params, 1), do: :ok + + defp copy_reports(params, 0) do + case ch(Query.copy_reports(), params, 90_000) do + {:ok, _} -> :ok + error -> error + end + end + + defp discard_generation(space, generation) do + # This UUID was never published, so no reader can depend on its partitions. + delete_retired(space, [generation]) + :ok + end + + @doc "Hide retired analytics immediately; obsolete generations are collected separately." + def invalidate(space_hash) do + with :ok <- validate_hash(space_hash), do: invalidate_transaction(space_hash) + end + + defp invalidate_transaction(space_hash) do + case Repo.transaction(fn -> invalidate_locked(space_hash) end, timeout: 180_000) do + {:ok, {:error, _} = error} -> error + result -> result + end + end + + defp invalidate_locked(space) do + lock!(space) + Repo.query!("DELETE FROM analytics_session_buckets WHERE space_hash = $1", [space]) + + with {:ok, _} <- + ch( + "ALTER TABLE analytics_sessions DELETE WHERE space_hash = {$0:String} SETTINGS mutations_sync = 1", + [space], + 60_000 + ), + {:ok, _} <- + ch( + "ALTER TABLE analytics_session_years DELETE WHERE space_hash = {$0:String} SETTINGS mutations_sync = 1", + [space], + 60_000 + ), + {:ok, _} <- + ch( + "ALTER TABLE analytics_session_reports DELETE WHERE space_hash = {$0:String} SETTINGS mutations_sync = 1", + [space], + 60_000 + ) do + Repo.query!("DELETE FROM analytics_session_retirements WHERE space_hash = $1", [space]) + :ok + else + _ -> {:error, :analytics_unavailable} + end + end + + @doc "Delete retired generations after the reader grace period, in batches per space." + def cleanup do + with {:ok, %{rows: rows}} <- + Repo.query(""" + SELECT space_hash FROM analytics_session_buckets + GROUP BY space_hash ORDER BY min(last_cleanup_at) ASC NULLS FIRST + """), + {:ok, %{rows: stored}} <- ch(Query.stored_spaces(), [], 60_000) do + ((stored -- rows) ++ rows) + |> Enum.take(setting(:cleanup_space_limit, 32)) + |> Enum.reduce_while(:ok, &cleanup_one/2) + end + end + + defp cleanup_one([space], :ok) do + case Repo.transaction(fn -> cleanup_locked(space) end, timeout: 180_000) do + {:ok, :ok} -> {:cont, :ok} + _ -> {:halt, {:error, :cleanup_failed}} + end + end + + defp cleanup_locked(space) do + lock!(space) + discover_retired!(space) + + generations = + Repo.query!( + """ + SELECT r.generation::text FROM analytics_session_retirements r + WHERE r.space_hash = $1 AND r.retired_at < now() - make_interval(secs => $2) + AND NOT EXISTS (SELECT 1 FROM analytics_session_buckets b + WHERE b.space_hash = r.space_hash AND b.generation = r.generation) + ORDER BY r.retired_at LIMIT $3 + """, + [ + space, + max(120, setting(:reader_grace_seconds, 300)), + setting(:cleanup_generation_limit, 64) + ] + ).rows + |> List.flatten() + + case delete_retired(space, generations) do + :ok -> + Repo.query!( + "DELETE FROM analytics_session_retirements WHERE space_hash = $1 AND generation = ANY($2::text[]::uuid[])", + [space, generations] + ) + + Repo.query!( + "UPDATE analytics_session_buckets SET last_cleanup_at = now() WHERE space_hash = $1", + [space] + ) + + :ok + + _ -> + Repo.rollback(:cleanup_failed) + end + end + + defp retire_previous!(space, bucket) do + Repo.query!( + """ + INSERT INTO analytics_session_retirements (space_hash, generation, retired_at) + SELECT space_hash, generation, now() FROM analytics_session_buckets + WHERE space_hash = $1 AND bucket = $2 + ON CONFLICT (space_hash, generation) DO UPDATE SET retired_at = EXCLUDED.retired_at + """, + [space, bucket] + ) + end + + defp discover_retired!(space) do + # Recover unpublished inserts left by a process crash. Start their grace + # period at discovery; creation time cannot prove when a reader last used them. + case ch(Query.stored_generations(), [space], 60_000) do + {:ok, %{rows: rows}} -> + Repo.query!( + """ + INSERT INTO analytics_session_retirements (space_hash, generation, retired_at) + SELECT $1, g::uuid, now() FROM unnest($2::text[]) AS g + WHERE NOT EXISTS (SELECT 1 FROM analytics_session_buckets b + WHERE b.space_hash = $1 AND b.generation = g::uuid) + ON CONFLICT DO NOTHING + """, + [space, List.flatten(rows)] + ) + + _ -> + Repo.rollback(:cleanup_failed) + end + end + + defp delete_retired(space, generations) do + Enum.reduce_while(generations, :ok, fn generation, :ok -> + with {:ok, _} <- ch(Query.cleanup(), [space, generation], 60_000), + {:ok, _} <- ch(Query.cleanup_overview(), [space, generation], 60_000), + {:ok, _} <- ch(Query.cleanup_reports(), [space, generation], 60_000) do + {:cont, :ok} + else + _ -> {:halt, {:error, :cleanup_failed}} + end + end) + end + + defp setting(key, default) do + Application.get_env(:mave_core, :analytics_sessions, []) |> Keyword.get(key, default) + end + + defp unchanged?([[count, fingerprint, version, refresh_after]], {count, fingerprint}), + do: version == Query.version() and not refresh_due?(refresh_after) + + defp unchanged?(_, _), do: false + + defp refresh_due?(nil), do: false + defp refresh_due?(at), do: NaiveDateTime.compare(naive(at), NaiveDateTime.utc_now()) != :gt + defp naive(nil), do: nil + + defp naive(%DateTime{} = at), + do: + at + |> DateTime.to_unix(:microsecond) + |> DateTime.from_unix!(:microsecond) + |> DateTime.to_naive() + + defp naive(%NaiveDateTime{} = at), do: at + + defp read_snapshot(space) do + # Read pointers and time boundaries in the same PostgreSQL snapshot. A + # concurrent publication cannot switch the fast-path decision underneath + # a reader that is still using the previous immutable generation. + case Repo.query( + "SELECT array_agg(generation::text ORDER BY bucket), max(latest_event_at), min(refresh_after), count(*) FROM analytics_session_buckets WHERE space_hash=$1 AND format_version=$2", + [space, Query.version()] + ) do + {:ok, %{rows: [[generations, latest, due, 16]]}} -> + {:ok, {generations, latest, due}} + + {:ok, _} -> + source_readiness(space) + + _ -> + {:error, :analytics_unavailable} + end + end + + defp overview_query(now, latest, due) 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()} + end + + defp lock!(space_hash) do + Repo.query!("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", [ + "analytics:" <> space_hash + ]) + end + + defp params(space, generations, now) do + date = DateTime.to_date(now) + + [ + space, + generations, + DateTime.new!(Date.new!(date.year, 1, 1), ~T[00:00:00]), + DateTime.new!(Date.new!(date.year, date.month, 1), ~T[00:00:00]), + DateTime.new!(date, ~T[00:00:00]), + now + ] + end + + defp map_overview(rows, space_hash) do + Enum.reduce(rows, empty(space_hash), fn + ["total", _, _, year, today, month, wt, wm, wy], data -> + %{ + data + | views: %{today: today, month: month, year: year}, + watch_time: %{today: wt, month: wm, year: wy} + } + + ["month", month, _, views | _], data -> + %{ + data + | views_by_month: + List.replace_at(data.views_by_month, month - 1, %{month: month, views: views}) + } + + ["device", _, dimension, views | _], data -> + %{data | devices: data.devices ++ [%{device: dimension || "Unknown", views: views}]} + + ["browser", _, dimension, views | _], data -> + %{data | browsers: data.browsers ++ [%{browser: dimension || "Unknown", views: views}]} + end) + end + + defp empty(space) do + %{ + id: space, + views: %{today: 0, month: 0, year: 0}, + watch_time: %{today: 0, month: 0, year: 0}, + unique_watch_time: %{today: 0, month: 0, year: 0}, + views_by_month: Enum.map(1..12, &%{month: &1, views: 0}), + devices: [], + browsers: [] + } + end + + defp validate_hash(hash) when is_binary(hash) do + if Regex.match?(~r/\A[A-Za-z0-9]{5}\z/, hash), do: :ok, else: {:error, :invalid_space_hash} + end + + defp validate_hash(_), do: {:error, :invalid_space_hash} + + defp validate_optional_hash(nil), do: :ok + defp validate_optional_hash(hash), do: validate_hash(hash) + + defp ch(sql, params, timeout \\ 15_000) do + ClickHouseRepo.query(sql, params, timeout: timeout) + end +end diff --git a/lib/mave_core/analytics/sessions/config.ex b/lib/mave_core/analytics/sessions/config.ex new file mode 100644 index 0000000..102a12d --- /dev/null +++ b/lib/mave_core/analytics/sessions/config.ex @@ -0,0 +1,70 @@ +defmodule MaveCore.Analytics.Sessions.Config do + @moduledoc false + + def from_env!(env) do + booleans = [ + {"MAVE_ANALYTICS_SESSIONS", :enabled}, + {"MAVE_ANALYTICS_PROCESSOR_ENABLED", :processor_enabled} + ] + + integers = [ + {"MAVE_ANALYTICS_BUILD_MEMORY_BYTES", :build_memory_bytes}, + {"MAVE_ANALYTICS_READ_MEMORY_BYTES", :read_memory_bytes}, + {"MAVE_ANALYTICS_QUERY_THREADS", :query_threads}, + {"MAVE_ANALYTICS_REFRESH_BUCKET_LIMIT", :refresh_bucket_limit}, + {"MAVE_ANALYTICS_CLEANUP_SPACE_LIMIT", :cleanup_space_limit}, + {"MAVE_ANALYTICS_CLEANUP_GENERATION_LIMIT", :cleanup_generation_limit}, + {"MAVE_ANALYTICS_READER_GRACE_SECONDS", :reader_grace_seconds} + ] + + Enum.flat_map(booleans, fn {name, key} -> + case Map.get(env, name) do + nil -> [] + value when value in ["true", "1"] -> [{key, true}] + value when value in ["false", "0"] -> [{key, false}] + _ -> raise ArgumentError, "#{name} must be true, false, 1 or 0" + end + end) ++ + Enum.flat_map(integers, fn {name, key} -> + case Map.get(env, name) do + nil -> [] + value -> [{key, integer_setting!(name, value)}] + end + end) ++ + space_list(env, "MAVE_ANALYTICS_READ_SPACES", :read_spaces) ++ + space_list(env, "MAVE_ANALYTICS_PROCESS_SPACES", :process_spaces) + end + + defp integer_setting!("MAVE_ANALYTICS_READER_GRACE_SECONDS" = name, value) do + seconds = positive_integer!(name, value) + if seconds < 120, do: raise(ArgumentError, "#{name} must be at least 120 seconds") + seconds + end + + defp integer_setting!(name, value), do: positive_integer!(name, value) + + defp positive_integer!(name, value) do + case Integer.parse(value) do + {number, ""} when number > 0 -> number + _ -> raise ArgumentError, "#{name} must be a positive integer" + end + end + + defp space_list(env, name, key) do + case Map.get(env, name) do + nil -> + [] + + "*" -> + [{key, :all}] + + value -> + spaces = value |> String.split(",", trim: true) |> Enum.map(&String.trim/1) + + unless Enum.all?(spaces, &Regex.match?(~r/\A[A-Za-z0-9]{5}\z/, &1)), + do: raise(ArgumentError, "#{name} must contain space hashes") + + [{key, Enum.uniq(spaces)}] + end + end +end diff --git a/lib/mave_core/analytics/sessions/query.ex b/lib/mave_core/analytics/sessions/query.ex new file mode 100644 index 0000000..fff5b17 --- /dev/null +++ b/lib/mave_core/analytics/sessions/query.ex @@ -0,0 +1,519 @@ +defmodule MaveCore.Analytics.Sessions.Query do + @moduledoc false + + # Keep this expression identical to the inventory projection. A revision + # describes all event fields, including changes that preserve the row count. + @fingerprint "cityHash64(timestamp, name, session_id, space_hash, embed_hash, video_time, duration, source_url, component, browser, browser_version, os, os_version, device, device_brand)" + @buckets 16 + + def buckets, do: @buckets + def version, do: 5 + + def inventory do + """ + SELECT space_hash, modulo(cityHash64(session_id), #{@buckets}) AS bucket, + count() AS source_rows, toString(sumWithOverflow(#{@fingerprint})) AS fingerprint + FROM events + GROUP BY space_hash, bucket + ORDER BY space_hash, bucket + """ + end + + def changed_sessions do + """ + WITH source AS ( + SELECT session_id, count() AS source_rows, + sumWithOverflow(#{@fingerprint}) AS fingerprint + FROM events + 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 + 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 + 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 + """ + end + + def copy_sessions do + """ + INSERT INTO analytics_sessions + SELECT space_hash, bucket, {$2:UUID}, session_id, source_rows, fingerprint, events, segments, created_at + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$3:UUID} + WHERE session_id NOT IN {$4:Array(UUID)} + #{copy_settings()} + """ + end + + def copy_overview do + """ + INSERT INTO analytics_session_years + WITH #{compact_days("days")} AS daily + SELECT space_hash, bucket, {$2:UUID}, session_id, year, first_at, source_device, source_browser, + watched, arrayMap(d -> (toDate(d.1), d.2), arrayZip(daily.1, daily.2)), created_at, unique_intervals + FROM analytics_session_years + PREWHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$3:UUID} + WHERE session_id NOT IN {$4:Array(UUID)} + #{copy_settings()} + """ + end + + def copy_reports do + """ + INSERT INTO analytics_session_reports + SELECT space_hash, bucket, {$2:UUID}, session_id, complete, days + FROM analytics_session_reports + PREWHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$3:UUID} + WHERE session_id NOT IN {$4:Array(UUID)} + #{copy_settings()} + """ + end + + def build_reports do + """ + INSERT INTO analytics_session_reports + WITH input AS ( + SELECT session_id, + NOT arrayExists(s -> s.3 > s.2 AND dateDiff('day', toDate(s.2, 'UTC'), toDate(s.3, 'UTC')) > 732, segments) AS complete, + arrayMap(s -> (s.1,s.2,s.3,s.2,s.3,s.4,s.5,s.6), + arrayFilter(s -> s.3 > s.2 AND dateDiff('day', toDate(s.2, 'UTC'), toDate(s.3, 'UTC')) <= 732, segments)) AS clipped + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$2:UUID} + AND ({$4:UInt8} = 1 OR session_id IN {$3:Array(UUID)}) + ), expanded AS ( + SELECT session_id, arrayJoin(#{bucket_segments(:day)}) AS p FROM input + ), daily AS ( + SELECT session_id, p.1 AS embed_hash, toDate(p.2, 'UTC') AS day, + sum(dateDiff('microsecond',p.2,p.3)) AS watched, + maxOrNullIf(least(1.0,greatest(0.0,(p.6+(p.7-p.6)*dateDiff('microsecond',p.4,p.3)/dateDiff('microsecond',p.4,p.5))/p.8)), + isFinite(p.8) AND p.8>0 AND isFinite(p.6) AND isFinite(p.7)) AS completion + FROM expanded GROUP BY session_id,embed_hash,day + ), reports AS ( + SELECT session_id, groupArray((embed_hash,day,watched,completion)) AS days + FROM daily GROUP BY session_id + ) + SELECT {$0:String},{$1:UInt8},{$2:UUID},input.session_id,toUInt8(input.complete),reports.days + FROM input LEFT JOIN reports ON input.session_id=reports.session_id + SETTINGS async_insert = 0, join_use_nulls = 0, max_threads = #{limit(:query_threads, 2)}, + max_memory_usage = #{limit(:build_memory_bytes, 1_073_741_824)}, + max_bytes_before_external_group_by = #{div(limit(:build_memory_bytes, 1_073_741_824), 4)}, max_execution_time = 60 + """ + end + + def report_coverage 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 + key = + %{ + day: "toString(p.2)", + week: "toString(toMonday(p.2))", + month: "toString(toStartOfMonth(p.2))" + }[granularity] + + """ + 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")} + """ + end + + def build do + """ + /* mave_perf:sessions.build */ + INSERT INTO analytics_sessions + WITH ordered AS ( + SELECT *, row_number() OVER ( + PARTITION BY embed_hash, session_id ORDER BY timestamp + ) AS event_order + FROM events + PREWHERE space_hash = {$0:String} + AND modulo(cityHash64(session_id), #{@buckets}) = {$1:UInt8} + AND ({$4:UInt8} = 1 OR session_id IN {$3:Array(UUID)}) + ), videos AS ( + SELECT session_id, embed_hash, count() AS source_rows, + sumWithOverflow(#{@fingerprint}) AS fingerprint, + arraySort(e -> e.1, groupArrayIf( + (event_order, timestamp, name, toFloat64(video_time), toFloat64(duration), + device, browser, os, source_url), name IN ('play', 'pause'))) AS ev + FROM ordered GROUP BY session_id, embed_hash + ), prepared AS ( + SELECT session_id, source_rows, fingerprint, + arrayMap(e -> (embed_hash, e.2, e.3, e.4, e.5, e.6, e.7, e.8, e.9), ev) AS events, + arrayMap(p -> (embed_hash, p.1.2, p.2.2, p.1.4, p.2.4, p.1.5), + arrayFilter(p -> p.1.3 = 'play' AND p.2.3 = 'pause', + arrayZip(arrayPopBack(ev), arrayPopFront(ev)))) AS segments + FROM videos + ) + SELECT {$0:String}, {$1:UInt8}, {$2:UUID}, session_id, + sum(source_rows), sumWithOverflow(fingerprint), + groupArrayArray(events), groupArrayArray(segments), now64(6) + FROM prepared GROUP BY session_id + SETTINGS async_insert = 0, max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:build_memory_bytes, 1_073_741_824)}, + max_bytes_before_external_group_by = #{div(limit(:build_memory_bytes, 1_073_741_824), 4)}, max_execution_time = 60 + """ + end + + 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 + SELECT sum(source_rows), toString(sumWithOverflow(fingerprint)), + minOrNullIf(arrayMin(future_events), notEmpty(future_events)), + maxOrNullIf(arrayMax(past_events), notEmpty(past_events)) + FROM analytics_sessions + WHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$2:UUID} + """ + end + + def build_overview do + """ + /* mave_perf:sessions.build_overview */ + INSERT INTO analytics_session_years + WITH years AS ( + SELECT *, arrayJoin(arrayDistinct(arrayMap(e -> toYear(e.2, 'UTC'), + arrayFilter(e -> e.2 < created_at, events)))) AS year + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND bucket = {$1:UInt8} AND generation = {$2:UUID} + AND ({$4:UInt8} = 1 OR session_id IN {$3:Array(UUID)}) + ), prepared AS ( + SELECT *, arraySort(e -> e.2, + arrayFilter(e -> toYear(e.2, 'UTC') = year AND e.2 < created_at, events)) AS ev, + arrayFilter(s -> toYear(s.2, 'UTC') = year AND toYear(s.3, 'UTC') = year AND s.3 < created_at, segments) AS seg + FROM years + ), daily_totals AS ( + SELECT *, #{compact_days("arrayMap(s -> (toDate(s.2, 'UTC'), greatest(toInt64(0), dateDiff('millisecond', s.2, s.3))), seg)")} AS daily + FROM prepared + ) + SELECT space_hash, bucket, generation, session_id, year, ev[1].2, ev[1].6, ev[1].7, + arraySum(arrayMap(s -> greatest(toInt64(0), dateDiff('millisecond', s.2, s.3)), seg)), + arrayMap(d -> (toDate(d.1), d.2), arrayZip(daily.1, daily.2)), created_at, + arrayDistinct(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 -> dateDiff('millisecond', s.2, s.3) < 3600000 + AND isFinite(s.4) AND isFinite(s.5) AND s.4 >= 0 AND s.5 > s.4, seg))) + FROM daily_totals + SETTINGS async_insert = 0, max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:read_memory_bytes, 536_870_912)}, max_execution_time = 60 + """ + end + + def overview_at do + """ + /* mave_perf:sessions.overview */ + WITH metrics AS ( + 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, + arraySort(e -> e.2, ev)[1] AS first, + toMonth(first.2) AS month, + nullIf(first.6, '') AS device, + nullIf(first.7, '') AS browser, + arraySum(arrayMap(s -> greatest(toInt64(0), dateDiff('millisecond', s.2, s.3)), seg)) AS watch_year, + arraySum(arrayMap(s -> if(s.2 >= {$3:DateTime64(6)}, greatest(toInt64(0), dateDiff('millisecond', s.2, s.3)), 0), seg)) AS watch_month, + 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)} + 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 + """ + /* 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') + ) + 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 unique_overview do + """ + WITH intervals AS ( + 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') + ), lengths AS ( + SELECT arrayReduce('intervalLengthSum', arrayMap(i -> i.2, iv), arrayMap(i -> i.3, iv)) AS year, + arrayReduce('intervalLengthSum', + arrayMap(i -> i.2, arrayFilter(i -> i.1 < toDate({$3:DateTime64(6)}, 'UTC'), iv)), + arrayMap(i -> i.3, arrayFilter(i -> i.1 < toDate({$3:DateTime64(6)}, 'UTC'), iv))) AS before_month, + arrayReduce('intervalLengthSum', + arrayMap(i -> i.2, arrayFilter(i -> i.1 < toDate({$4:DateTime64(6)}, 'UTC'), iv)), + arrayMap(i -> i.3, arrayFilter(i -> i.1 < toDate({$4:DateTime64(6)}, 'UTC'), iv))) AS before_today + FROM intervals + ) + SELECT sum(year - before_today) * 1000, sum(year - before_month) * 1000, sum(year) * 1000 + FROM lengths + SETTINGS max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:read_memory_bytes, 536_870_912)}, max_execution_time = 15 + """ + end + + # Union half-open integer intervals instead of expanding each watched second. + # Subtract the pre-period union to retain the legacy earliest-watch semantics. + def unique_watch_time do + """ + /* mave_perf:sessions.unique_watch_time */ + WITH prepared AS ( + SELECT 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 seg + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + ), intervals AS ( + SELECT arrayMap(s -> (s.2, 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))))), seg) AS iv + FROM prepared + ), lengths AS ( + SELECT arrayReduce('intervalLengthSum', arrayMap(i -> i.2, iv), arrayMap(i -> i.3, iv)) AS year, + arrayReduce('intervalLengthSum', + arrayMap(i -> i.2, arrayFilter(i -> i.1 < {$3:DateTime64(6)}, iv)), + arrayMap(i -> i.3, arrayFilter(i -> i.1 < {$3:DateTime64(6)}, iv))) AS before_month, + arrayReduce('intervalLengthSum', + arrayMap(i -> i.2, arrayFilter(i -> i.1 < {$4:DateTime64(6)}, iv)), + arrayMap(i -> i.3, arrayFilter(i -> i.1 < {$4:DateTime64(6)}, iv))) AS before_today + FROM intervals + ) + SELECT sum(year - before_today) * 1000, sum(year - before_month) * 1000, sum(year) * 1000 + FROM lengths + SETTINGS max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:read_memory_bytes, 536_870_912)}, max_execution_time = 15 + """ + end + + def stored_spaces do + """ + SELECT space_hash FROM analytics_sessions GROUP BY space_hash + UNION DISTINCT + SELECT space_hash FROM analytics_session_years GROUP BY space_hash + UNION DISTINCT + SELECT space_hash FROM analytics_session_reports GROUP BY space_hash + """ + end + + def stored_generations do + """ + SELECT toString(generation) FROM analytics_sessions + WHERE space_hash = {$0:String} GROUP BY generation + UNION DISTINCT + SELECT toString(generation) FROM analytics_session_years + WHERE space_hash = {$0:String} GROUP BY generation + UNION DISTINCT + SELECT toString(generation) FROM analytics_session_reports + WHERE space_hash = {$0:String} GROUP BY generation + """ + end + + def cleanup do + "ALTER TABLE analytics_sessions DROP PARTITION ({$0:String}, {$1:UUID})" + end + + def cleanup_reports do + "ALTER TABLE analytics_session_reports DROP PARTITION ({$0:String}, {$1:UUID})" + end + + def cleanup_overview do + "ALTER TABLE analytics_session_years DROP PARTITION ({$0:String}, {$1:UUID})" + end + + def monthly_watch_time do + """ + /* mave_perf:sessions.monthly_watch_time */ + SELECT sum(arraySum(arrayMap(s -> greatest(toInt64(0), dateDiff('millisecond', s.2, s.3)), + arrayFilter(s -> s.2 >= {$2:DateTime64(6)} AND s.3 < {$3:DateTime64(6)}, segments)))) + FROM analytics_sessions + PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} + SETTINGS max_threads = #{limit(:query_threads, 2)}, max_memory_usage = #{limit(:read_memory_bytes, 268_435_456)}, max_execution_time = 5 + """ + end + + def range(granularity, mode) + when granularity in [:day, :week, :month] and mode in [:series, :videos] do + bucket = + %{ + day: "toDate(p.2, 'UTC')", + week: "toMonday(toDate(p.2, 'UTC'))", + month: "toStartOfMonth(toDate(p.2, 'UTC'))" + }[granularity] + + key = if mode == :series, do: "toString(#{bucket})", else: "p.1" + rows = if mode == :series, do: bucket_segments(granularity), else: "clipped" + + completion_at = + if mode == :series, + do: + "if(p.7 < p.6, least(p.3, toDateTime64(addDays(toDate(p.2, 'UTC'), 1), 6, 'UTC')), p.3)", + else: "p.3" + + """ + /* mave_perf:sessions.range */ + WITH filtered AS ( + 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) + ) + #{range_aggregation(mode, "analytics_sessions")} + """ + end + + defp range_aggregation(mode, table) do + keys = + if mode == :series, + do: "arrayConcat([''], arrayDistinct(arrayMap(p -> p.1, pieces)))", + else: "arrayDistinct(arrayMap(p -> p.1, pieces))" + + matches = if mode == :series, do: "key = '' OR p.1 = key", else: "p.1 = key" + single_bucket_keys = if mode == :series, do: 2, else: 1 + + source = + if mode == :series do + "(SELECT metrics FROM session_metrics UNION ALL SELECT [('', toInt64(0), CAST(NULL, 'Nullable(Float64)'))] AS metrics FROM (SELECT 1 FROM #{table} PREWHERE space_hash = {$0:String} AND generation IN {$1:Array(UUID)} LIMIT 1))" + else + "session_metrics" + end + + """ + , grouped AS ( + SELECT pieces, #{keys} AS keys, + arraySum(arrayMap(p -> p.2, pieces)) AS total_watched, + arrayReduce('maxOrNull', arrayMap(p -> p.3, pieces)) AS total_completion + FROM pieces + ), session_metrics AS ( + SELECT if(length(keys) <= #{single_bucket_keys}, + arrayMap(key -> (key, total_watched, total_completion), keys), + arrayMap((key, selected) -> (key, + arraySum(arrayMap(p -> p.2, selected)), + arrayReduce('maxOrNull', arrayMap(p -> p.3, selected))), + keys, arrayMap(key -> arrayFilter(p -> #{matches}, pieces), keys))) AS metrics + FROM grouped + ) + SELECT metric.1, countIf(metric.2 >= 1000000), sum(metric.2) / 1000000.0, + avgOrNullIf(metric.3, metric.2 >= 1000000 AND isNotNull(metric.3)) + FROM #{source} ARRAY JOIN metrics AS metric + GROUP BY metric.1 + SETTINGS max_threads = #{limit(:query_threads, 2)}, + max_memory_usage = #{limit(:read_memory_bytes, 536_870_912)}, max_execution_time = 15, + short_circuit_function_evaluation = 'force_enable' + """ + end + + defp bucket_segments(granularity) do + {unit, start, finish, add} = + case granularity do + :day -> + {"day", "toDate(s.2, 'UTC')", "toDate(s.3 - INTERVAL 1 MICROSECOND, 'UTC')", "addDays"} + + :week -> + {"week", "toMonday(toDate(s.2, 'UTC'))", + "toMonday(toDate(s.3 - INTERVAL 1 MICROSECOND, 'UTC'))", "addWeeks"} + + :month -> + {"month", "toStartOfMonth(toDate(s.2, 'UTC'))", + "toStartOfMonth(toDate(s.3 - INTERVAL 1 MICROSECOND, 'UTC'))", "addMonths"} + end + + """ + arrayConcat(arrayFilter(s -> #{start} = #{finish}, clipped), + arrayFlatten(arrayMap(s -> arrayMap(offset -> + (s.1, greatest(s.2, toDateTime64(#{add}(#{start}, offset), 6, 'UTC')), + least(s.3, toDateTime64(#{add}(#{start}, offset + 1), 6, 'UTC')), + s.4, s.5, s.6, s.7, s.8), + range(toUInt32(dateDiff('#{unit}', #{start}, #{finish})) + 1)), + arrayFilter(s -> #{start} != #{finish}, clipped)))) + """ + 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 + """ + SETTINGS async_insert = 0, max_threads = 1, max_insert_threads = 1, + max_block_size = 512, preferred_block_size_bytes = 2097152, + min_insert_block_size_rows = 0, min_insert_block_size_bytes = 33554432, + max_memory_usage = #{min(limit(:build_memory_bytes, 1_073_741_824), 536_870_912)}, + max_execution_time = 60 + """ + end + + # Preserve the original start-day attribution while combining repeated + # contributions. Dashboard reads need one total per day, not per segment. + defp compact_days(days) do + "arrayReduce('sumMap', arrayMap(d -> [toUInt16(d.1)], #{days}), arrayMap(d -> [d.2], #{days}))" + end + + defp limit(key, default) do + case Keyword.get(Application.get_env(:mave_core, :analytics_sessions, []) || [], key, default) do + value when is_integer(value) and value >= 4 -> value + value when key == :query_threads and is_integer(value) and value > 0 -> value + _ -> default + end + end +end diff --git a/lib/mave_core/workers/analytics_sessions_cleanup_worker.ex b/lib/mave_core/workers/analytics_sessions_cleanup_worker.ex new file mode 100644 index 0000000..ae00b3e --- /dev/null +++ b/lib/mave_core/workers/analytics_sessions_cleanup_worker.ex @@ -0,0 +1,16 @@ +defmodule MaveCore.Workers.AnalyticsSessionsCleanupWorker do + @moduledoc "Collects superseded session generations outside interactive requests." + use Oban.Worker, + queue: :analytics, + max_attempts: 3, + unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]] + + alias MaveCore.Analytics.Sessions + + @impl true + def perform(_job) do + if Sessions.enabled?() or Sessions.processing_enabled?(), + do: Sessions.cleanup(), + else: :ok + end +end diff --git a/lib/mave_core/workers/analytics_sessions_worker.ex b/lib/mave_core/workers/analytics_sessions_worker.ex new file mode 100644 index 0000000..9a29d74 --- /dev/null +++ b/lib/mave_core/workers/analytics_sessions_worker.ex @@ -0,0 +1,21 @@ +defmodule MaveCore.Workers.AnalyticsSessionsWorker do + @moduledoc "Refreshes bounded session buckets using the durable event inventory." + use Oban.Worker, + queue: :analytics, + max_attempts: 5, + unique: [period: :infinity, states: [:available, :scheduled, :executing, :retryable]] + + alias MaveCore.Analytics.Sessions + + @impl true + def perform(_job) do + config = Application.get_env(:mave_core, :analytics_sessions, []) + spaces = Keyword.get(config, :process_spaces, :all) + + if Sessions.processing_enabled?() and spaces != [] do + with {:ok, _} <- Sessions.refresh_changed(spaces: spaces), do: :ok + else + :ok + end + end +end diff --git a/lib/mave_core_web/controllers/api/data/spaces_controller.ex b/lib/mave_core_web/controllers/api/data/spaces_controller.ex index 2f87a2b..0283b5f 100644 --- a/lib/mave_core_web/controllers/api/data/spaces_controller.ex +++ b/lib/mave_core_web/controllers/api/data/spaces_controller.ex @@ -26,6 +26,11 @@ defmodule MaveCoreWeb.Api.Data.SpacesController do {:ok, data} -> json(conn, %{data: data}) + {:error, reason} when reason in [:preparing, :analytics_unavailable] -> + conn + |> put_resp_header("retry-after", "60") + |> send_resp(:service_unavailable, "Analytics are temporarily unavailable") + _ -> send_resp(conn, :bad_request, "Invalid space_hash") end diff --git a/lib/mave_core_web/live/dashboard/data/index.ex b/lib/mave_core_web/live/dashboard/data/index.ex index c64fced..3620302 100644 --- a/lib/mave_core_web/live/dashboard/data/index.ex +++ b/lib/mave_core_web/live/dashboard/data/index.ex @@ -12,6 +12,7 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do def mount(_params, _session, socket) do {:ok, socket + |> assign(:analytics_preparing, false) |> assign(:page_title, gettext("Data")) |> assign(:settings_open, false) |> assign(:show_all_devices, false) @@ -31,7 +32,7 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do |> assign(:data_access_notice, nil) |> assign(:data_access_help, nil) |> assign_async_defaults() - |> start_async(:fetch_stats, fn -> Analytics.space_data(space_hash) end) + |> start_async(:fetch_stats, fn -> Analytics.space_overview(space_hash) end) {:error, reason} -> socket @@ -44,6 +45,7 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do end def handle_async(:fetch_stats, {:ok, {:ok, result}}, socket) do + socket = assign(socket, :analytics_preparing, false) views_by_month = Enum.map(result.views_by_month, &Map.put(&1, :count, &1.views)) max_views = Enum.max_by(views_by_month, & &1.count, fn -> %{count: 0} end).count max_month = max_views + max_views / 3 @@ -67,11 +69,27 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do |> assign(:show_all_browsers, false)} end + def handle_async(:fetch_stats, {:ok, {:error, :preparing}}, socket) do + Process.send_after(self(), {:refresh_analytics, socket.assigns.current_space.hash}, 5_000) + {:noreply, socket |> assign_failed() |> assign(:analytics_preparing, true)} + end + def handle_async(:fetch_stats, {:ok, {:error, _reason}}, socket), do: {:noreply, assign_failed(socket)} def handle_async(:fetch_stats, {:exit, _reason}, socket), do: {:noreply, assign_failed(socket)} + def handle_info({:refresh_analytics, space_hash}, socket) do + if socket.assigns.current_space.hash == space_hash and socket.assigns.analytics_preparing do + {:noreply, + start_async(socket, :fetch_stats, fn -> Analytics.space_overview(space_hash) end)} + else + {:noreply, socket} + end + end + + def handle_info(_message, socket), do: {:noreply, socket} + def handle_event("toggle_show_all_devices", _, socket) do {:noreply, assign(socket, :show_all_devices, !socket.assigns.show_all_devices)} end @@ -99,6 +117,14 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do
+

+ {gettext("Analytics are being prepared. Please check back shortly.")} +

<.subtitle label={gettext("Numbers")} icon="hero-hashtag" />
@@ -229,13 +255,13 @@ defmodule MaveCoreWeb.Dashboard.Data.Index do """ end - attr :title, :string, required: true - attr :assign, :any, required: true - attr :rows, :list, required: true - attr :toggle, :string, required: true - attr :expanded, :boolean, required: true - attr :key_name, :atom, required: true - attr :empty_icon, :string, required: true + attr(:title, :string, required: true) + attr(:assign, :any, required: true) + attr(:rows, :list, required: true) + attr(:toggle, :string, required: true) + attr(:expanded, :boolean, required: true) + attr(:key_name, :atom, required: true) + attr(:empty_icon, :string, required: true) defp source_panel(assigns) do ~H""" diff --git a/lib/mix/tasks/analytics.sessions.backfill.ex b/lib/mix/tasks/analytics.sessions.backfill.ex new file mode 100644 index 0000000..61f2c0e --- /dev/null +++ b/lib/mix/tasks/analytics.sessions.backfill.ex @@ -0,0 +1,73 @@ +defmodule Mix.Tasks.Analytics.Sessions.Backfill do + use Mix.Task + @moduledoc "Resumable maintenance command for prepared analytics." + alias MaveCore.Analytics.Sessions + alias MaveCore.ClickHouseRepo + + @shortdoc "Build or resume prepared analytics sessions in bounded buckets" + + @impl true + def run(args) do + {opts, rest, invalid} = + OptionParser.parse(args, + strict: [ + rounds: :integer, + limit: :integer, + materialize_inventory: :boolean, + materialize_session_revisions: :boolean, + space: :string + ] + ) + + rounds = Keyword.get(opts, :rounds, 100) + limit = Keyword.get(opts, :limit, 16) + space = Keyword.get(opts, :space) + + validate_options!(rest, invalid, rounds, limit, space) + + Mix.Task.run("app.start") + + if Keyword.get(opts, :materialize_inventory, false) do + ClickHouseRepo.query!( + "ALTER TABLE events MATERIALIZE PROJECTION analytics_session_inventory SETTINGS mutations_sync = 1", + [], + timeout: :infinity + ) + end + + if Keyword.get(opts, :materialize_session_revisions, false) do + ClickHouseRepo.query!( + "ALTER TABLE events MATERIALIZE PROJECTION analytics_session_revisions SETTINGS mutations_sync = 1", + [], + timeout: :infinity + ) + end + + Enum.reduce_while(1..rounds, :ok, fn round, :ok -> + case Sessions.refresh_changed(limit: limit, space_hash: space) do + {:ok, 0} -> + Mix.shell().info("Session inventory is up to date.") + {:halt, :ok} + + {:ok, count} -> + Mix.shell().info("Round #{round}: published #{count} session buckets.") + {:cont, :ok} + + {:error, reason} -> + Mix.raise("Session backfill stopped: #{inspect(reason)}") + end + end) + end + + defp validate_options!(rest, invalid, rounds, limit, space) do + unless rest == [] and invalid == [] and rounds in 1..1000 and limit in 1..64 and + valid_space?(space) do + Mix.raise( + "Expected --rounds 1..1000, --limit 1..64, --space HASH, or --materialize-inventory / --materialize-session-revisions" + ) + end + end + + defp valid_space?(nil), do: true + defp valid_space?(space), do: Regex.match?(~r/\A[A-Za-z0-9]{5}\z/, space) +end diff --git a/mix.exs b/mix.exs index 9643749..ca9af35 100644 --- a/mix.exs +++ b/mix.exs @@ -49,6 +49,8 @@ defmodule MaveCore.MixProject do defp deps do [ {:phoenix, "~> 1.8.3"}, + # Keep downstream applications on the patched token/error boundary. + {:plug_crypto, ">= 2.2.1 and < 3.0.0"}, {:phoenix_ecto, "~> 4.5"}, {:ecto_sql, "~> 3.13"}, {:postgrex, "~> 0.22.3"}, diff --git a/mix.lock b/mix.lock index 2ceaec7..1d9dacf 100644 --- a/mix.lock +++ b/mix.lock @@ -67,7 +67,7 @@ "phoenix_storybook": {:hex, :phoenix_storybook, "1.3.0", "fb5224398fc63ca4d64d0261da1e4edf4d4cdc9e8767d93460bd97974e0972ed", [:mix], [{:jason, "~> 1.3", [hex: :jason, repo: "hexpm", optional: true]}, {:makeup_eex, "~> 2.0.2", [hex: :makeup_eex, repo: "hexpm", optional: false]}, {:makeup_html, "~> 0.2.0", [hex: :makeup_html, repo: "hexpm", optional: false]}, {:mdex, "~> 0.12", [hex: :mdex, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.8.1", [hex: :phoenix, repo: "hexpm", optional: false]}, {:phoenix_live_view, "~> 1.1.21 or ~> 1.2.0", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}], "hexpm", "3c1a864c1d20fdf7fd140e734cabde1d0da1f04612858bb0fa4f017fca942c6f"}, "phoenix_template": {:hex, :phoenix_template, "1.0.4", "e2092c132f3b5e5b2d49c96695342eb36d0ed514c5b252a77048d5969330d639", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "2c0c81f0e5c6753faf5cca2f229c9709919aba34fab866d3bc05060c9c444206"}, "plug": {:hex, :plug, "1.20.3", "56c480c633ec2ce10140e236e15233bf576e1d323887d7c96711bd02ab5160db", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "be266aee1b8536ef6409d58cf39a3121319f0ec47cfa1b24024485aa0e76ad76"}, - "plug_crypto": {:hex, :plug_crypto, "2.2.0", "144014737daaf485407f5ed77daeaad74d651b216a28c87543f8cc7043f8efc8", [:mix], [], "hexpm", "83a95744ab1c75876542b6fab135fcc176280e0f301a111c1f757fddcec95d2c"}, + "plug_crypto": {:hex, :plug_crypto, "2.2.1", "bd1d0358a2e2f5ffb818217e64c394f9d24ed191d862b7fa1bcce264ace2b8ea", [:mix], [], "hexpm", "4417ccbd8f85e627321cd07d2ab9c4b621f1fd883b9905af241f9b4b780b94b6"}, "postgrex": {:hex, :postgrex, "0.22.4", "d271f595dfd25230b6398354e19d17bb5e2d20130fd2d9bdca7e15f125d43552", [:mix], [{:db_connection, "~> 2.9", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "4aae45a2d60e35b04eea2602440be152fae332901f1fc7a60fc7cb7f0f9a9c5a"}, "quic": {:hex, :quic, "1.8.0", "7b074fc6f8a13b4a81e4e11254bf061db61c5c07b4366d568a79a38c5e8bb5f3", [:rebar3], [], "hexpm", "6b1f8f08a45412ac503bc6771efb8edc696bb14079bc48d2ca960b92933da8a4"}, "ranch": {:hex, :ranch, "2.2.1", "fc2bb0e800efbeab8ff82eb325450b9c08653b14c3f69e8b09bbea304973105d", [:make, :rebar3], [], "hexpm", "55f05cce20ec2da1d90de5d5981afb93dbfc01325fc7e933189aa5c62f037c24"}, diff --git a/priv/clickhouse_repo/migrations/20261023100000_create_analytics_sessions.exs b/priv/clickhouse_repo/migrations/20261023100000_create_analytics_sessions.exs new file mode 100644 index 0000000..a4cf0e9 --- /dev/null +++ b/priv/clickhouse_repo/migrations/20261023100000_create_analytics_sessions.exs @@ -0,0 +1,49 @@ +defmodule MaveCore.ClickHouseRepo.Migrations.CreateAnalyticsSessions do + use Ecto.Migration + + def up do + # Generations are written once and retired as complete partitions; merging + # their parts spends memory/IO without improving update correctness. + execute(""" + CREATE TABLE analytics_sessions ( + space_hash String, bucket UInt8, generation UUID, session_id UUID, + source_rows UInt64, fingerprint UInt64, + events Array(Tuple(String, DateTime64(6), String, Float64, Float64, String, String, String, String)), + segments Array(Tuple(String, DateTime64(6), DateTime64(6), Float64, Float64, Float64)), + created_at DateTime64(6) + ) ENGINE = MergeTree + PARTITION BY (space_hash, generation) + ORDER BY (space_hash, bucket, generation, session_id) + SETTINGS max_bytes_to_merge_at_max_space_in_pool = 0 + """) + + execute(""" + CREATE TABLE analytics_session_years ( + space_hash String, bucket UInt8, generation UUID, session_id UUID, year UInt16, + first_at DateTime64(6), source_device String, source_browser String, + watched Int64, days Array(Tuple(Date, Int64)), created_at DateTime64(6), + unique_intervals Array(Tuple(Date, UInt64, UInt64)) + ) ENGINE = MergeTree PARTITION BY (space_hash, generation) ORDER BY (space_hash, bucket, generation, year, session_id) + SETTINGS max_bytes_to_merge_at_max_space_in_pool = 0 + """) + + # Existing parts may be materialized separately; correctness does not depend + # on the projection being present in every part. Never backfill during deploy. + execute(""" + ALTER TABLE events ADD PROJECTION analytics_session_inventory ( + SELECT space_hash, modulo(cityHash64(session_id), 16) AS bucket, + count() AS source_rows, + sumWithOverflow(cityHash64(timestamp, name, session_id, space_hash, embed_hash, + video_time, duration, source_url, component, browser, browser_version, + os, os_version, device, device_brand)) AS fingerprint + GROUP BY space_hash, bucket + ) + """) + end + + def down do + execute("ALTER TABLE events DROP PROJECTION IF EXISTS analytics_session_inventory") + execute("DROP TABLE IF EXISTS analytics_sessions") + execute("DROP TABLE IF EXISTS analytics_session_years") + end +end diff --git a/priv/clickhouse_repo/migrations/20261023120000_add_analytics_session_revision_projection.exs b/priv/clickhouse_repo/migrations/20261023120000_add_analytics_session_revision_projection.exs new file mode 100644 index 0000000..3d6fd8d --- /dev/null +++ b/priv/clickhouse_repo/migrations/20261023120000_add_analytics_session_revision_projection.exs @@ -0,0 +1,21 @@ +defmodule MaveCore.ClickHouseRepo.Migrations.AddAnalyticsSessionRevisionProjection do + use Ecto.Migration + + def up do + # An aggregate inventory does not reorder raw event reads or their ties. + execute(""" + ALTER TABLE events ADD PROJECTION analytics_session_revisions ( + SELECT space_hash, modulo(cityHash64(session_id), 16) AS bucket, session_id, + count() AS source_rows, + sumWithOverflow(cityHash64(timestamp, name, session_id, space_hash, embed_hash, + video_time, duration, source_url, component, browser, browser_version, + os, os_version, device, device_brand)) AS fingerprint + GROUP BY space_hash, bucket, session_id + ) + """) + end + + def down do + execute("ALTER TABLE events DROP PROJECTION IF EXISTS analytics_session_revisions") + end +end diff --git a/priv/clickhouse_repo/migrations/20261023130000_create_analytics_session_reports.exs b/priv/clickhouse_repo/migrations/20261023130000_create_analytics_session_reports.exs new file mode 100644 index 0000000..aa85e6d --- /dev/null +++ b/priv/clickhouse_repo/migrations/20261023130000_create_analytics_session_reports.exs @@ -0,0 +1,17 @@ +defmodule MaveCore.ClickHouseRepo.Migrations.CreateAnalyticsSessionReports do + use Ecto.Migration + + def up do + execute(""" + CREATE TABLE analytics_session_reports ( + space_hash String, bucket UInt8, generation UUID, session_id UUID, + complete UInt8, days Array(Tuple(String, Date, Int64, Nullable(Float64))) + ) ENGINE = MergeTree + PARTITION BY (space_hash, generation) + ORDER BY (space_hash, bucket, generation, session_id) + SETTINGS max_bytes_to_merge_at_max_space_in_pool = 0 + """) + end + + def down, do: execute("DROP TABLE IF EXISTS analytics_session_reports") +end diff --git a/priv/repo/migrations/20261023100000_create_analytics_session_buckets.exs b/priv/repo/migrations/20261023100000_create_analytics_session_buckets.exs new file mode 100644 index 0000000..7f8b585 --- /dev/null +++ b/priv/repo/migrations/20261023100000_create_analytics_session_buckets.exs @@ -0,0 +1,18 @@ +defmodule MaveCore.Repo.Migrations.CreateAnalyticsSessionBuckets do + use Ecto.Migration + + def change do + create table(:analytics_session_buckets, primary_key: false) do + add(:space_hash, :text, primary_key: true) + add(:bucket, :integer, primary_key: true) + add(:generation, :uuid, null: false) + add(:source_rows, :bigint, null: false) + add(:fingerprint, :text, null: false) + add(:format_version, :integer, null: false) + add(:refresh_after, :utc_datetime_usec) + add(:latest_event_at, :utc_datetime_usec) + add(:last_cleanup_at, :utc_datetime_usec) + add(:updated_at, :utc_datetime_usec, null: false) + end + end +end diff --git a/priv/repo/migrations/20261023110000_create_analytics_session_retirements.exs b/priv/repo/migrations/20261023110000_create_analytics_session_retirements.exs new file mode 100644 index 0000000..30a4185 --- /dev/null +++ b/priv/repo/migrations/20261023110000_create_analytics_session_retirements.exs @@ -0,0 +1,13 @@ +defmodule MaveCore.Repo.Migrations.CreateAnalyticsSessionRetirements do + use Ecto.Migration + + def change do + create table(:analytics_session_retirements, primary_key: false) do + add(:space_hash, :text, primary_key: true) + add(:generation, :uuid, primary_key: true) + add(:retired_at, :utc_datetime_usec, null: false) + end + + create(index(:analytics_session_retirements, [:retired_at])) + end +end diff --git a/test/mave_core/analytics/sessions/config_test.exs b/test/mave_core/analytics/sessions/config_test.exs new file mode 100644 index 0000000..6a2fb6b --- /dev/null +++ b/test/mave_core/analytics/sessions/config_test.exs @@ -0,0 +1,50 @@ +defmodule MaveCore.Analytics.Sessions.ConfigTest do + use ExUnit.Case, async: true + + alias MaveCore.Analytics.Sessions.Config + + test "unset rollout settings leave application defaults intact" do + assert Config.from_env!(%{}) == [] + end + + test "processing can be enabled independently of a scoped reader" do + settings = + Config.from_env!(%{ + "MAVE_ANALYTICS_SESSIONS" => "false", + "MAVE_ANALYTICS_PROCESSOR_ENABLED" => "true", + "MAVE_ANALYTICS_READ_SPACES" => "AAAAA, BBBBB,AAAAA", + "MAVE_ANALYTICS_BUILD_MEMORY_BYTES" => "2147483648", + "MAVE_ANALYTICS_REFRESH_BUCKET_LIMIT" => "4", + "MAVE_ANALYTICS_CLEANUP_SPACE_LIMIT" => "32" + }) + + assert settings[:enabled] == false + assert settings[:processor_enabled] == true + assert settings[:read_spaces] == ["AAAAA", "BBBBB"] + assert settings[:build_memory_bytes] == 2_147_483_648 + assert settings[:refresh_bucket_limit] == 4 + assert settings[:cleanup_space_limit] == 32 + assert Config.from_env!(%{"MAVE_ANALYTICS_READ_SPACES" => ""}) == [read_spaces: []] + assert Config.from_env!(%{"MAVE_ANALYTICS_READ_SPACES" => "*"}) == [read_spaces: :all] + + assert Config.from_env!(%{"MAVE_ANALYTICS_PROCESS_SPACES" => "AAAAA,AAAAA"}) == + [process_spaces: ["AAAAA"]] + + assert Config.from_env!(%{"MAVE_ANALYTICS_PROCESS_SPACES" => "*"}) == [process_spaces: :all] + assert Config.from_env!(%{"MAVE_ANALYTICS_PROCESS_SPACES" => ""}) == [process_spaces: []] + end + + test "invalid rollout settings fail startup instead of enabling an unintended scope" do + for env <- [ + %{"MAVE_ANALYTICS_SESSIONS" => "yes"}, + %{"MAVE_ANALYTICS_PROCESSOR_ENABLED" => "typo"}, + %{"MAVE_ANALYTICS_QUERY_THREADS" => "0"}, + %{"MAVE_ANALYTICS_READ_MEMORY_BYTES" => "-1"}, + %{"MAVE_ANALYTICS_READER_GRACE_SECONDS" => "119"}, + %{"MAVE_ANALYTICS_READ_SPACES" => "invalid"}, + %{"MAVE_ANALYTICS_PROCESS_SPACES" => "invalid"} + ] do + assert_raise ArgumentError, fn -> Config.from_env!(env) end + end + end +end diff --git a/test/mave_core/analytics/sessions_test.exs b/test/mave_core/analytics/sessions_test.exs new file mode 100644 index 0000000..53eac1b --- /dev/null +++ b/test/mave_core/analytics/sessions_test.exs @@ -0,0 +1,598 @@ +defmodule MaveCore.Analytics.SessionsTest do + use MaveCore.DataCase, async: false + + alias MaveCore.{Analytics, ClickHouseRepo, Repo} + alias MaveCore.Analytics.Sessions + alias MaveCore.Workers.AnalyticsSessionsWorker + + setup do + Process.put(:analytics_time, DateTime.utc_now() |> DateTime.truncate(:second)) + old = Application.get_env(:mave_core, :analytics_sessions) + Application.put_env(:mave_core, :analytics_sessions, enabled: false) + ClickHouseRepo.query!("TRUNCATE TABLE events") + ClickHouseRepo.query!("TRUNCATE TABLE analytics_sessions") + ClickHouseRepo.query!("TRUNCATE TABLE analytics_session_years") + ClickHouseRepo.query!("TRUNCATE TABLE analytics_session_reports") + on_exit(fn -> Application.put_env(:mave_core, :analytics_sessions, old || []) end) + :ok + end + + test "prepares the legacy overview and unique-watch result without changing its values" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(1, "AAAAA", "first", 10, 15, 2, 7) + pair(2, "AAAAA", "other", 20, 22, 0, 2) + pair(3, "BBBBB", "other", 20, 29, 0, 9) + + assert {:error, :preparing} = Sessions.data("AAAAA") + assert {:ok, expected} = Analytics.space_data("AAAAA") + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, actual} = Sessions.data("AAAAA", include_unique: true) + assert actual == expected + assert {:ok, 0} = Sessions.refresh_changed(limit: 32) + end + + test "bounded rounds eventually prepare all spaces without starving later buckets" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(2, "BBBBB", "second", 20, 25, 0, 5) + for _ <- 1..8, do: assert({:ok, 4} = Sessions.refresh_changed()) + assert {:ok, _} = Sessions.generations("AAAAA") + assert {:ok, _} = Sessions.generations("BBBBB") + assert {:ok, 0} = Sessions.refresh_changed() + end + + test "a scoped backfill leaves other spaces unprepared" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(2, "BBBBB", "second", 20, 25, 0, 5) + assert {:ok, 16} = Sessions.refresh_changed(limit: 32, space_hash: "AAAAA") + assert {:ok, _} = Sessions.data("AAAAA") + assert {:error, :preparing} = Sessions.data("BBBBB") + + assert {:error, :invalid_space_hash} = + Sessions.refresh_changed(limit: 32, space_hash: "invalid") + end + + test "background preparation is independent of reader activation and reader scope" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(2, "BBBBB", "second", 20, 25, 0, 5) + + Application.put_env(:mave_core, :analytics_sessions, + enabled: false, + processor_enabled: true, + refresh_bucket_limit: 32 + ) + + refute Sessions.enabled?("AAAAA") + assert Sessions.processing_enabled?() + assert :ok = AnalyticsSessionsWorker.perform(%{}) + assert {:ok, _} = Sessions.generations("AAAAA") + + Application.put_env(:mave_core, :analytics_sessions, + enabled: true, + processor_enabled: false, + read_spaces: ["AAAAA"] + ) + + assert Sessions.enabled?("AAAAA") + refute Sessions.enabled?("BBBBB") + refute Sessions.processing_enabled?() + Repo.query!("DELETE FROM analytics_session_buckets WHERE space_hash='BBBBB'") + assert {:ok, %{views: %{year: 1}}} = Analytics.space_data("BBBBB") + assert {:ok, %{views: %{year: 1}}} = Analytics.space_data("AAAAA") + end + + test "the background worker only prepares its configured spaces" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(2, "BBBBB", "second", 20, 25, 0, 5) + + Application.put_env(:mave_core, :analytics_sessions, + enabled: false, + processor_enabled: true, + process_spaces: ["AAAAA"], + refresh_bucket_limit: 32 + ) + + assert :ok = AnalyticsSessionsWorker.perform(%{}) + assert {:ok, _} = Sessions.generations("AAAAA") + assert {:error, :preparing} = Sessions.generations("BBBBB") + assert {:ok, 0} = Sessions.refresh_changed(spaces: []) + assert {:error, :invalid_space_hash} = Sessions.refresh_changed(spaces: ["invalid"]) + assert {:ok, 16} = Sessions.refresh_changed(space_hash: "BBBBB", limit: 32) + end + + test "a pause arriving first and a later historical play refresh the affected bucket" do + event(1, "AAAAA", "video", "pause", 10, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, before} = Sessions.data("AAAAA") + assert before.views.year == 0 + + # Arrival order differs from the event timestamp order. + event(1, "AAAAA", "video", "play", 15, 0) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + assert {:ok, after_late} = Sessions.data("AAAAA") + assert after_late.views.year == 1 + assert after_late.watch_time.year == 5_000 + end + + test "large per-video event arrays remain flattenable into a session" do + # Nested groupArray serializes each entire video array as one element, which + # exceeds ClickHouse's 16 MiB element limit even when total memory fits. + ClickHouseRepo.query!(""" + INSERT INTO events (timestamp, name, session_id, space_hash, embed_hash, + video_time, duration, source_url) + SELECT now64(6) - INTERVAL 2048 SECOND + toIntervalSecond(number), + if(number % 2 = 0, 'play', 'pause'), + toUUID('00000000-0000-0000-0000-000000000001'), 'AAAAA', 'large', + toFloat32(number % 2), 60, repeat('x', 32768) + FROM numbers(576) + """) + + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, data} = Sessions.data("AAAAA") + assert data.views.year == 1 + assert data.watch_time.year == 288_000 + + assert ClickHouseRepo.query!( + "SELECT source_rows, length(events), length(segments) FROM analytics_sessions" + ).rows == [[576, 576, 288]] + end + + test "timestamp ties preserve the raw window's playback pairing" do + ClickHouseRepo.query!(""" + INSERT INTO events (timestamp, name, session_id, space_hash, embed_hash, + video_time, duration) + SELECT now64(6) - INTERVAL 2048 SECOND + toIntervalSecond(intDiv(number, 4)), + ['play', 'pause', 'play', 'pause'][1 + number % 4], + toUUID('00000000-0000-0000-0000-000000000001'), 'AAAAA', 'tied', + toFloat32(number % 30), 60 + FROM numbers(1024) + """) + + assert {:ok, expected} = Analytics.space_data("AAAAA") + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, actual} = Sessions.data("AAAAA", include_unique: true) + assert actual == expected + end + + test "an incremental refresh preserves unchanged session rows exactly" do + [[other_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", "first", 20, 25, 0, 5) + pair(other_id, "AAAAA", "second", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + {:ok, before_generations} = Sessions.generations("AAAAA") + + sql = + "SELECT events, segments, created_at FROM analytics_sessions WHERE space_hash='AAAAA' AND session_id=toUUID('00000000-0000-0000-0000-000000000001') AND generation IN {$0:Array(UUID)}" + + before = ClickHouseRepo.query!(sql, [before_generations]).rows + pair(other_id, "AAAAA", "added", 10, 15, 0, 5) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + {:ok, after_generations} = Sessions.generations("AAAAA") + assert before == ClickHouseRepo.query!(sql, [after_generations]).rows + assert {:ok, %{views: %{year: 2}, watch_time: %{year: 15_000}}} = Sessions.data("AAAAA") + end + + test "a failed rebuild preserves the published generation and can be retried" do + pair(1, "AAAAA", "video", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, generations} = Sessions.generations("AAAAA") + ClickHouseRepo.query!("RENAME TABLE analytics_session_years TO temporarily_unavailable_years") + + try do + assert {:error, :analytics_unavailable} = + Sessions.refresh_bucket("AAAAA", 1, {999, "forced"}) + + assert {:ok, ^generations} = Sessions.generations("AAAAA") + after + ClickHouseRepo.query!( + "RENAME TABLE temporarily_unavailable_years TO analytics_session_years" + ) + end + + assert {:ok, 0} = Sessions.refresh_changed(limit: 32) + end + + test "failed physical invalidation hides pointers and reports an error for retry" do + pair(1, "AAAAA", "video", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + ClickHouseRepo.query!("RENAME TABLE analytics_session_years TO temporarily_unavailable_years") + + try do + assert {:error, :analytics_unavailable} = Sessions.invalidate("AAAAA") + assert {:error, :preparing} = Sessions.generations("AAAAA") + after + ClickHouseRepo.query!( + "RENAME TABLE temporarily_unavailable_years TO analytics_session_years" + ) + end + end + + test "source deletion invalidates the inventory and cannot retain old totals" do + pair(1, "AAAAA", "video", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + ClickHouseRepo.query!( + "ALTER TABLE events DELETE WHERE space_hash = 'AAAAA' SETTINGS mutations_sync = 1" + ) + + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + assert {:ok, data} = Sessions.data("AAAAA") + assert data.views.year == 0 + end + + test "invalid hashes and the shared storage space never expose other tenants" do + assert {:error, :invalid_space_hash} = Sessions.data("AAAAA' OR 1=1") + assert {:ok, empty} = Sessions.data("trial") + assert empty.views.year == 0 + assert Repo.query!("SELECT count(*) FROM analytics_session_buckets").rows == [[0]] + from = DateTime.add(DateTime.utc_now(), -86_400) + assert {:ok, %{}} = Sessions.range_data("trial", from, DateTime.utc_now()) + end + + test "range views remain unique across midnight and support video export batches" do + midnight = DateTime.new!(Date.utc_today(), ~T[00:00:00]) + Process.put(:analytics_time, DateTime.add(midnight, 120)) + pair(1, "AAAAA", "video", 118, 122, 0, 4) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + from = DateTime.add(midnight, -86_400) + to = DateTime.add(midnight, 86_400) + assert {:ok, report} = Sessions.range_data("AAAAA", from, to) + assert report[""].views == 1 + assert report[""].watch_seconds == 4.0 + assert report[Date.to_iso8601(Date.utc_today())].views == 1 + assert report[Date.to_iso8601(Date.add(Date.utc_today(), -1))].views == 1 + + assert {:ok, videos} = + Sessions.range_data("AAAAA", from, to, mode: :videos, embed_hashes: ["video"]) + + assert videos["video"].views == 1 + + assert {:ok, %{}} = + Sessions.range_data("AAAAA", from, to, mode: :videos, embed_hashes: ["absent"]) + end + + test "range view thresholds combine a shared session but stay separate per video" do + pair(1, "AAAAA", "first", 20.4, 21, 0, 0.6) + pair(1, "AAAAA", "second", 10.4, 11, 0, 0.6) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + to = Process.get(:analytics_time) + from = DateTime.add(to, -30) + assert {:ok, report} = Sessions.range_data("AAAAA", from, to) + assert report[""].views == 1 + assert report[""].watch_seconds == 1.2 + assert {:ok, videos} = Sessions.range_data("AAAAA", from, to, mode: :videos) + assert videos["first"].views == 0 + assert videos["second"].views == 0 + end + + test "same-count source mutations are detected by the inventory fingerprint" do + pair(1, "AAAAA", "video", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + ClickHouseRepo.query!(""" + ALTER TABLE events UPDATE device = 'mobile' + WHERE space_hash = 'AAAAA' SETTINGS mutations_sync = 1 + """) + + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + assert {:ok, data} = Sessions.data("AAAAA") + assert data.watch_time.year == 5000 + assert data.devices == [%{device: "mobile", views: 1}] + end + + test "empty periods keep the zero series total and no video rows" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + from = DateTime.add(DateTime.utc_now(), 3600) + to = DateTime.add(from, 3600) + + assert {:ok, %{"" => %{views: 0, watch_seconds: zero, completion_rate: nil}}} = + Sessions.range_data("AAAAA", from, to) + + assert zero == 0.0 + assert {:ok, %{}} = Sessions.range_data("AAAAA", from, to, mode: :videos) + end + + test "coarse ranges preserve daily completion semantics across boundaries" do + id = "00000000-0000-0000-0000-000000000001" + + for {at, name, position} <- [ + {~U[2026-01-30 12:00:00Z], "play", 90}, + {~U[2026-02-02 12:00:00Z], "pause", 0} + ] do + ClickHouseRepo.query!( + "INSERT INTO events (timestamp,name,session_id,space_hash,embed_hash,video_time,duration) VALUES ({$0:DateTime64(6)},{$1:String},{$2:UUID},'AAAAA','long',{$3:Float32},100)", + [at, name, id, position] + ) + end + + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + from = ~U[2026-01-30 12:00:00Z] + to = ~U[2026-02-02 12:00:00Z] + assert {:ok, months} = Sessions.range_data("AAAAA", from, to, granularity: :month) + assert months[""].watch_seconds == 259_200.0 + assert months["2026-01-01"].watch_seconds == 129_600.0 + assert months["2026-02-01"].watch_seconds == 129_600.0 + assert_in_delta months["2026-01-01"].completion_rate, 0.75, 0.00001 + assert_in_delta months["2026-02-01"].completion_rate, 0.15, 0.00001 + assert {:ok, weeks} = Sessions.range_data("AAAAA", from, to, granularity: :week) + assert weeks["2026-01-26"].watch_seconds == 216_000.0 + assert weeks["2026-02-02"].watch_seconds == 43_200.0 + assert_in_delta weeks["2026-01-26"].completion_rate, 0.75, 0.00001 + assert weeks["2026-02-02"].completion_rate == 0.0 + assert {:ok, videos} = Sessions.range_data("AAAAA", from, to, mode: :videos) + assert videos["long"].watch_seconds == 259_200.0 + assert videos["long"].completion_rate == 0.0 + + assert {:ok, ^months} = + Sessions.range_data("AAAAA", ~U[2026-01-30 00:00:00Z], ~U[2026-02-03 00:00:00Z], + granularity: :month + ) + + assert {:ok, ^weeks} = + Sessions.range_data("AAAAA", ~U[2026-01-30 00:00:00Z], ~U[2026-02-03 00:00:00Z], + granularity: :week + ) + end + + test "unique intervals deduplicate within a day while retaining earliest-day attribution" do + for ago <- [20, 86_420, 172_820], embed <- ["first", "second"] do + pair(1, "AAAAA", embed, ago, ago + 5, 0, 5) + end + + assert {:ok, expected} = Analytics.space_data("AAAAA") + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, actual} = Sessions.data("AAAAA", include_unique: true) + assert actual == expected + assert actual.unique_watch_time.year == 5000 + + assert [[3]] = + ClickHouseRepo.query!( + "SELECT sum(length(unique_intervals)) FROM analytics_session_years" + ).rows + end + + test "older summaries gain unique intervals without rebuilding raw session rows" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + before = + ClickHouseRepo.query!("SELECT events,segments,created_at FROM analytics_sessions").rows + + ClickHouseRepo.query!( + "ALTER TABLE analytics_session_years UPDATE unique_intervals=[] WHERE 1 SETTINGS mutations_sync=1" + ) + + Repo.query!("UPDATE analytics_session_buckets SET format_version=4") + assert {:ok, 16} = Sessions.refresh_changed(limit: 32) + {:ok, active} = Sessions.generations("AAAAA") + + assert before == + ClickHouseRepo.query!( + "SELECT events,segments,created_at FROM analytics_sessions WHERE generation IN {$0:Array(UUID)}", + [active] + ).rows + + assert {:ok, %{unique_watch_time: %{year: 5000}}} = + Sessions.data("AAAAA", include_unique: true) + end + + test "historical reads retain precise unique-watch cutoffs within a day" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + pair(1, "AAAAA", "second", 5, 10, 5, 10) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + assert {:ok, %{unique_watch_time: %{year: 5000}}} = + Sessions.data("AAAAA", + include_unique: true, + now: DateTime.add(Process.get(:analytics_time), -15) + ) + + assert {:ok, %{unique_watch_time: %{year: 10_000}}} = + Sessions.data("AAAAA", include_unique: true) + end + + test "overview combines same-day contributions without changing session thresholds" do + pair(1, "AAAAA", "first", 20, 20.6, 0, 0.6) + pair(1, "AAAAA", "second", 10, 10.6, 0, 0.6) + assert {:ok, expected} = Analytics.space_data("AAAAA") + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert {:ok, actual} = Sessions.data("AAAAA", include_unique: true) + assert actual == expected + assert actual.views.today == 1 + + assert [[1, 1200]] = + ClickHouseRepo.query!( + "SELECT length(days), arraySum(arrayMap(d -> d.2, days)) FROM analytics_session_years" + ).rows + end + + test "copied older overview rows combine repeated day entries" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + ClickHouseRepo.query!(""" + ALTER TABLE analytics_session_years UPDATE + days=[(toDate(first_at),toInt64(2000)),(toDate(first_at),toInt64(3000))] + WHERE space_hash='AAAAA' SETTINGS mutations_sync=1 + """) + + assert {:ok, before} = Sessions.data("AAAAA") + Repo.query!("UPDATE analytics_session_buckets SET format_version=3") + assert {:ok, 16} = Sessions.refresh_changed(limit: 32) + assert {:ok, after_copy} = Sessions.data("AAAAA") + assert after_copy == before + {:ok, active} = Sessions.generations("AAAAA") + + assert [[1, 5000]] = + ClickHouseRepo.query!( + "SELECT length(days), arraySum(arrayMap(d -> d.2, days)) FROM analytics_session_years WHERE generation IN {$0:Array(UUID)}", + [active] + ).rows + end + + test "existing session snapshots gain compact reports during a format upgrade" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + Repo.query!("UPDATE analytics_session_buckets SET format_version=3") + ClickHouseRepo.query!("TRUNCATE TABLE analytics_session_reports") + assert {:ok, 16} = Sessions.refresh_changed(limit: 32) + assert {:ok, %{views: %{year: 1}, watch_time: %{year: 5000}}} = Sessions.data("AAAAA") + assert [[1]] = ClickHouseRepo.query!("SELECT count() FROM analytics_session_reports").rows + end + + test "very long segments retain exact range fallback without unbounded report expansion" do + for {at, name, position} <- [ + {~U[2023-01-01 00:00:00Z], "play", 0}, + {~U[2026-01-01 00:00:00Z], "pause", 1} + ] do + ClickHouseRepo.query!( + "INSERT INTO events (timestamp,name,session_id,space_hash,embed_hash,video_time,duration) VALUES ({$0:DateTime64(6)},{$1:String},toUUID('00000000-0000-0000-0000-000000000001'),'AAAAA','long',{$2:Float32},10)", + [at, name, position] + ) + end + + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + assert [[0]] = ClickHouseRepo.query!("SELECT complete FROM analytics_session_reports").rows + + assert {:ok, report} = + Sessions.range_data("AAAAA", ~U[2025-01-01 00:00:00Z], ~U[2025-01-08 00:00:00Z]) + + assert report[""].views == 1 + assert report[""].watch_seconds == 604_800.0 + end + + test "partial future days use clipped segments instead of complete day reports" do + target = DateTime.new!(Date.add(Date.utc_today(), 2), ~T[12:00:00]) + Process.put(:analytics_time, target) + pair(1, "AAAAA", "future", -5, 5, 0, 10) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + from = DateTime.new!(DateTime.to_date(target), ~T[00:00:00]) + assert {:ok, report} = Sessions.range_data("AAAAA", from, target) + assert report[""].watch_seconds == 5.0 + end + + test "future-dated pauses become visible without another insert" do + pair(1, "AAAAA", "normal", 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) + assert {:ok, before} = Sessions.data("AAAAA") + assert before.watch_time.year == 5000 + Process.sleep(16_000) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + assert {:ok, later} = Sessions.data("AAAAA") + assert later.watch_time.year == 10_000 + 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) + + ClickHouseRepo.query!( + "ALTER TABLE analytics_sessions UPDATE created_at = now() - INTERVAL 2 HOUR WHERE 1 SETTINGS mutations_sync = 1" + ) + + ClickHouseRepo.query!( + "ALTER TABLE analytics_session_years UPDATE created_at = now() - INTERVAL 2 HOUR WHERE 1 SETTINGS mutations_sync = 1" + ) + + pair(1, "AAAAA", "second", 10, 15, 0, 5) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + # The old snapshot was just retired, even though it was built two hours ago. + assert :ok = Sessions.cleanup() + assert ClickHouseRepo.query!("SELECT count() FROM analytics_sessions").rows == [[2]] + + Repo.query!( + "UPDATE analytics_session_retirements SET retired_at = now() - interval '2 hours'" + ) + + assert :ok = Sessions.cleanup() + assert ClickHouseRepo.query!("SELECT count() FROM analytics_sessions").rows == [[1]] + assert ClickHouseRepo.query!("SELECT count() FROM analytics_session_years").rows == [[1]] + assert {:ok, data} = Sessions.data("AAAAA") + assert data.watch_time.year == 10_000 + end + + test "cleanup makes bounded progress through a retirement backlog" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + pair(1, "AAAAA", "second", 10, 15, 0, 5) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + pair(1, "AAAAA", "third", 5, 10, 0, 5) + assert {:ok, 1} = Sessions.refresh_changed(limit: 32) + Repo.query!("UPDATE analytics_session_retirements SET retired_at=now()-interval '2 hours'") + Application.put_env(:mave_core, :analytics_sessions, cleanup_generation_limit: 1) + assert :ok = Sessions.cleanup() + assert [[2]] = ClickHouseRepo.query!("SELECT count() FROM analytics_sessions").rows + assert :ok = Sessions.cleanup() + assert [[1]] = ClickHouseRepo.query!("SELECT count() FROM analytics_sessions").rows + assert {:ok, %{views: %{year: 1}}} = Sessions.data("AAAAA") + end + + test "cleanup discovers an unpublished generation and preserves active readers" do + pair(1, "AAAAA", "first", 20, 25, 0, 5) + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + {:ok, active} = Sessions.generations("AAAAA") + # Simulate a crash after the ClickHouse insert but before first publication. + ClickHouseRepo.query!( + "INSERT INTO analytics_sessions SELECT * REPLACE('CCCCC' AS space_hash) FROM analytics_sessions" + ) + + assert :ok = Sessions.cleanup() + + assert [[1]] = + Repo.query!( + "SELECT count(*) FROM analytics_session_retirements WHERE space_hash='CCCCC'" + ).rows + + Repo.query!("UPDATE analytics_session_retirements SET retired_at=now()-interval '2 hours'") + # A stale retirement record must never delete a currently published version. + Repo.query!( + "INSERT INTO analytics_session_retirements SELECT space_hash,generation,now()-interval '2 hours' FROM analytics_session_buckets ON CONFLICT DO NOTHING" + ) + + assert :ok = Sessions.cleanup() + + assert [[0]] = + ClickHouseRepo.query!( + "SELECT count() FROM analytics_sessions WHERE space_hash='CCCCC'" + ).rows + + assert {:ok, ^active} = Sessions.generations("AAAAA") + assert {:ok, %{views: %{year: 1}}} = Sessions.data("AAAAA") + end + + defp pair(id, space, embed, pause_ago, play_ago, start, finish) do + event(id, space, embed, "play", play_ago, start) + event(id, space, embed, "pause", pause_ago, finish) + end + + defp event(id, space, embed, name, seconds_ago, position) do + uuid = "00000000-0000-0000-0000-" <> String.pad_leading(to_string(id), 12, "0") + + ClickHouseRepo.query!( + """ + INSERT INTO events (timestamp, name, session_id, space_hash, embed_hash, + video_time, duration, device, browser, os, source_url) + VALUES ({$0:DateTime64(6)}, {$1:String}, {$2:UUID}, + {$3:String}, {$4:String}, {$5:Float32}, 60, 'desktop', 'Chrome', 'Linux', 'https://example.test') + """, + [ + DateTime.add(Process.get(:analytics_time), -round(seconds_ago * 1000), :millisecond), + name, + uuid, + space, + embed, + position + ] + ) + end +end diff --git a/test/mave_core/playback/session_test.exs b/test/mave_core/playback/session_test.exs index db4cc69..81bd880 100644 --- a/test/mave_core/playback/session_test.exs +++ b/test/mave_core/playback/session_test.exs @@ -44,4 +44,12 @@ defmodule MaveCore.Playback.SessionTest do expired = Phoenix.Token.sign(Endpoint, "media-playback-v1", %{claims | expires_at: now - 1}) assert {:error, :unauthorized} == Playback.authorize(expired, embed) end + + test "malformed request token types are rejected without a crypto exception" do + embed = %Embed{id: Ecto.UUID.generate(), space_id: Ecto.UUID.generate()} + + for token <- [["token"], %{"token" => "value"}, 42, true] do + assert {:error, :unauthorized} == Playback.authorize(token, embed) + end + end end diff --git a/test/mave_core_web/controllers/api/data/spaces_controller_test.exs b/test/mave_core_web/controllers/api/data/spaces_controller_test.exs index 1649458..7a567a1 100644 --- a/test/mave_core_web/controllers/api/data/spaces_controller_test.exs +++ b/test/mave_core_web/controllers/api/data/spaces_controller_test.exs @@ -3,6 +3,7 @@ defmodule MaveCoreWeb.Api.Data.SpacesControllerTest do alias MaveCore.Accounts alias MaveCore.Analytics + alias MaveCore.Analytics.Sessions alias MaveCore.Spaces defmodule DataBlockedUsageLimits do @@ -37,6 +38,41 @@ defmodule MaveCoreWeb.Api.Data.SpacesControllerTest do end describe "GET /api/v1/spaces/:space_hash/data" do + test "prepared analytics reports a pending backfill as unavailable", %{ + conn: conn, + space: space, + auth: auth + } do + previous = Application.get_env(:mave_core, :analytics_sessions) + Application.put_env(:mave_core, :analytics_sessions, enabled: true) + on_exit(fn -> Application.put_env(:mave_core, :analytics_sessions, previous || []) end) + + MaveCore.ClickHouseRepo.query!( + """ + INSERT INTO events (timestamp, name, session_id, space_hash, embed_hash) + VALUES (now64(6), 'play', '00000000-0000-0000-0000-000000005555', {$0:String}, 'AAAAAAAAAA') + """, + [space.hash] + ) + + response = + conn + |> put_req_header("authorization", auth) + |> get(~p"/api/v1/spaces/#{space.hash}/data") + + assert response.status == 503 + assert get_resp_header(response, "retry-after") == ["60"] + + assert {:ok, _} = Sessions.refresh_changed(limit: 32) + + response = + conn + |> put_req_header("authorization", auth) + |> get(~p"/api/v1/spaces/#{space.hash}/data") + + assert %{"data" => %{"views" => %{"year" => 0}}} = json_response(response, 200) + end + test "space_monthly_watch_time returns only the current month watch time", %{ db_name: db_name, space: space diff --git a/test/mave_core_web/controllers/api/playback_controller_test.exs b/test/mave_core_web/controllers/api/playback_controller_test.exs index 85ad5e3..a142776 100644 --- a/test/mave_core_web/controllers/api/playback_controller_test.exs +++ b/test/mave_core_web/controllers/api/playback_controller_test.exs @@ -94,6 +94,16 @@ defmodule MaveCoreWeb.Api.PlaybackControllerTest do assert %{"visibility" => "public"} = json_response(conn, 200) end + test "array and map playback tokens return unauthorized without a crypto error", context do + {:ok, embed} = Embeds.create_video_embed(context.space, %{visibility: :private}) + path = "/api/v1/playback/media/#{context.space.hash}#{embed.hash}/manifest.json" + + for token <- [["invalid"], %{"nested" => "invalid"}] do + conn = Phoenix.ConnTest.get(build_conn(), path, %{token: token}) + assert response(conn, 401) == "Playback authorization required" + end + end + test "failed or unavailable private protection and invalid visibility leave no video behind", context do Process.put(:playback_storage_result, {:error, :storage_unavailable}) diff --git a/test/mave_core_web/live/dashboard/data/index_test.exs b/test/mave_core_web/live/dashboard/data/index_test.exs index 5999d7c..c70eedb 100644 --- a/test/mave_core_web/live/dashboard/data/index_test.exs +++ b/test/mave_core_web/live/dashboard/data/index_test.exs @@ -53,6 +53,7 @@ defmodule MaveCoreWeb.Live.Dashboard.Data.IndexTest do html = %{ data_access_notice: nil, + analytics_preparing: false, data_access_help: nil, views_today: async_ok(12_345), views_this_month: async_ok(67_890), @@ -86,6 +87,7 @@ defmodule MaveCoreWeb.Live.Dashboard.Data.IndexTest do html = %{ data_access_notice: nil, + analytics_preparing: false, data_access_help: nil, views_today: loading, views_this_month: loading,