defmodule Plausible.Exports do @moduledoc """ Contains functions to export data for events and sessions as Zip archives. """ use Plausible use Plausible.Stats.SQL.Fragments import Ecto.Query @doc "Schedules CSV export job to S3 storage" @spec schedule_s3_export(pos_integer, String.t()) :: {:ok, Oban.Job.t()} | {:error, :no_data} def schedule_s3_export(site_id, email_to) do with :ok <- ensure_has_data(site_id) do args = %{ "storage" => "s3", "site_id" => site_id, "email_to" => email_to, "s3_bucket" => Plausible.S3.exports_bucket(), "s3_path" => s3_export_key(site_id) } {:ok, Oban.insert!(Plausible.Workers.ExportAnalytics.new(args))} end end @doc "Schedules CSV export job to local storage" @spec schedule_local_export(pos_integer, String.t()) :: {:ok, Oban.Job.t()} | {:error, :no_data} def schedule_local_export(site_id, email_to) do with :ok <- ensure_has_data(site_id) do args = %{ "storage" => "local", "site_id" => site_id, "email_to" => email_to, "local_path" => local_export_file(site_id) } {:ok, Oban.insert!(Plausible.Workers.ExportAnalytics.new(args))} end end @spec ensure_has_data(pos_integer) :: :ok | {:error, :no_data} defp ensure_has_data(site_id) do # SELECT true FROM "events_v2" AS e0 WHERE (e0."site_id" = ^site_id) LIMIT 1 has_data? = Plausible.ClickhouseRepo.exists?(from "events_v2", where: [site_id: ^site_id]) if has_data?, do: :ok, else: {:error, :no_data} end @doc "Gets last CSV export job for a site" @spec get_last_export_job(pos_integer) :: Oban.Job.t() | nil def get_last_export_job(site_id) do Plausible.Repo.one( from e in Plausible.Workers.ExportAnalytics.base_query(site_id), order_by: [desc: :id], limit: 1 ) end @doc "Subscribes to CSV export job notifications" def oban_listen, do: Oban.Notifier.listen(__MODULE__) @doc false def oban_notify(site_id), do: Oban.Notifier.notify(__MODULE__, %{"site_id" => site_id}) @doc """ Renders export archive filename. Examples: iex> archive_filename("plausible.io", _created_on = ~D[2024-12-31]) "plausible_io_20241231.zip" """ def archive_filename(domain, %Date{} = created_on) do String.replace(domain, ".", "_") <> "_" <> Calendar.strftime(created_on, "%Y%m%d") <> ".zip" end @doc ~S""" Safely renders content disposition for an arbitrary export filename. Examples: iex> content_disposition("plausible_io_20241231.zip") "attachment; filename=\"plausible_io_20241231.zip\"" iex> content_disposition("📊.zip") "attachment; filename=\"plausible-export.zip\"; filename*=utf-8''%F0%9F%93%8A.zip" """ def content_disposition(filename) do encoded_filename = URI.encode(filename) if encoded_filename == filename do ~s[attachment; filename="#{filename}"] else ~s[attachment; filename="plausible-export.zip"; filename*=utf-8''#{encoded_filename}] end end @type export :: %{ path: Path.t(), name: String.t(), expires_at: DateTime.t() | nil, size: pos_integer } @doc "Gets local export for a site" @spec get_local_export(pos_integer, String.t(), String.t()) :: export | nil def get_local_export(site_id, domain, timezone) do path = local_export_file(site_id) if File.exists?(path) do %File.Stat{size: size, mtime: mtime} = File.stat!(path, time: :posix) created_at = DateTime.from_unix!(mtime) created_on_in_site_tz = Plausible.Timezones.to_date_in_timezone(created_at, timezone) name = archive_filename(domain, created_on_in_site_tz) %{path: path, name: name, expires_at: nil, size: size} end end @doc "Deletes local export for a site" @spec delete_local_export(pos_integer) :: :ok def delete_local_export(site_id) do file = local_export_file(site_id) if File.exists?(file) do File.rm!(file) end :ok end @spec local_export_file(pos_integer) :: Path.t() defp local_export_file(site_id) do data_dir = Application.get_env(:plausible, :data_dir) Path.join([data_dir || System.tmp_dir!(), "plausible-exports", Integer.to_string(site_id)]) end @doc "Gets S3 export for a site. Raises if object storage is unavailable." @spec get_s3_export!(pos_integer, non_neg_integer) :: export | nil def get_s3_export!(site_id, retries \\ 0) do path = s3_export_key(site_id) bucket = Plausible.S3.exports_bucket() head_object_op = ExAws.S3.head_object(bucket, path) case ExAws.request(head_object_op, retries: retries) do {:ok, %{status_code: 200, headers: headers}} -> "attachment; filename=" <> filename = :proplists.get_value("content-disposition", headers) name = String.trim(filename, "\"") size = :proplists.get_value("content-length", headers, nil) expires_at = if x_amz_expiration = :proplists.get_value("x-amz-expiration", headers, nil) do ["expiry-date=", expiry_date, ", rule-id=", _rule_id] = String.split(x_amz_expiration, "\"", trim: true) Plausible.Times.parse!(expiry_date, "{RFC1123}") end %{ path: path, name: name, expires_at: expires_at, size: String.to_integer(size) } {:error, {:http_error, 404, _response}} -> nil {:error, %Finch.TransportError{} = e} -> raise e end end @doc "Deletes S3 export for a site. Raises if object storage is unavailable." @spec delete_s3_export!(pos_integer) :: :ok def delete_s3_export!(site_id) do if export = get_s3_export!(site_id) do exports_bucket = Plausible.S3.exports_bucket() delete_op = ExAws.S3.delete_object(exports_bucket, export.path) ExAws.request!(delete_op) end :ok end defp s3_export_key(site_id), do: Integer.to_string(site_id) @doc "Returns the date range for the site's events data in site's timezone or `nil` if there is no data" @spec date_range(pos_integer, String.t()) :: Date.Range.t() | nil def date_range(site_id, timezone) do [%Date{} = start_date, %Date{} = end_date] = Plausible.ClickhouseRepo.one( from e in "events_v2", where: [site_id: ^site_id], select: [ fragment("toDate(min(?),?)", e.timestamp, ^timezone), fragment("toDate(max(?),?)", e.timestamp, ^timezone) ] ) unless end_date == ~D[1970-01-01] do Date.range(start_date, end_date) end end @doc """ Builds Ecto queries to export data from `events_v2` and `sessions_v2` tables into the format of `imported_*` tables for a website. """ @spec export_queries(Plausible.Site.t(), extname: String.t(), date_range: Date.Range.t(), timezone: String.t() ) :: %{String.t() => Ecto.Query.t()} def export_queries(site, opts \\ []) do extname = opts[:extname] || ".csv" date_range = opts[:date_range] timezone = opts[:timezone] || "UTC" suffix = if date_range do first_date = Calendar.strftime(date_range.first, "%Y%m%d") last_date = Calendar.strftime(date_range.last, "%Y%m%d") "_#{first_date}_#{last_date}" <> extname else extname end filename = fn name -> name <> suffix end %{ filename.("imported_visitors") => export_visitors_q(site, timezone, date_range), filename.("imported_sources") => export_sources_q(site, timezone, date_range), filename.("imported_pages") => export_pages_q(site, timezone, date_range), filename.("imported_entry_pages") => export_entry_pages_q(site, timezone, date_range), filename.("imported_exit_pages") => export_exit_pages_q(site, timezone, date_range), filename.("imported_custom_events") => export_custom_events_q(site, timezone, date_range), filename.("imported_locations") => export_locations_q(site, timezone, date_range), filename.("imported_devices") => export_devices_q(site, timezone, date_range), filename.("imported_browsers") => export_browsers_q(site, timezone, date_range), filename.("imported_operating_systems") => export_operating_systems_q(site, timezone, date_range), filename.("imported_custom_props") => export_custom_props_q(site, timezone, date_range) } end on_ee do defp sampled(table) do Plausible.Stats.Sampling.add_query_hint(from(table)) end else defp sampled(table) do table end end defp export_filter(site_id, date_range) do filter = dynamic([t], t.site_id == ^site_id) if date_range do dynamic( ^filter and selected_as(:date) >= ^date_range.first and selected_as(:date) <= ^date_range.last ) else filter end end defmacrop date(timestamp, timezone) do quote do selected_as( fragment("toDate(?,?)", unquote(timestamp), unquote(timezone)), :date ) end end defmacrop visit_duration(t) do quote do selected_as( scale_sample(fragment("greatest(sum(?*?),0)", unquote(t).sign, unquote(t).duration)), :visit_duration ) end end defmacrop visitors(t) do quote do selected_as( scale_sample(fragment("uniq(?)", unquote(t).user_id)), :visitors ) end end defmacrop visits(t) do quote do selected_as( scale_sample(fragment("greatest(sum(?),0)", unquote(t).sign)), :visits ) end end defmacrop bounces(t) do quote do selected_as( scale_sample( fragment( "greatest(sum(?*?),0)", unquote(t).sign, unquote(t).is_bounce ) ), :bounces ) end end defmacrop pageviews(t) do quote do selected_as( scale_sample( fragment( "greatest(sum(?*?),0)", unquote(t).sign, unquote(t).pageviews ) ), :pageviews ) end end defp export_visitors_q(site, timezone, date_range) do visitors_sessions_q = from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: selected_as(:date), select: %{ date: date(s.timestamp, ^timezone), bounces: bounces(s), visits: visits(s), visit_duration: visit_duration(s), visitors: visitors(s) } visitors_events_q = from e in sampled("events_v2"), where: ^export_filter(site.id, date_range), group_by: selected_as(:date), select: %{ date: date(e.timestamp, ^timezone), pageviews: selected_as( scale_sample(fragment("countIf(?='pageview')", e.name)), :pageviews ) } visitors_q = "e" |> with_cte("e", as: ^visitors_events_q) |> with_cte("s", as: ^visitors_sessions_q) from e in visitors_q, full_join: s in "s", on: e.date == s.date, order_by: selected_as(:date), select: [ selected_as(fragment("greatest(?,?)", s.date, e.date), :date), s.visitors, e.pageviews, s.bounces, s.visits, s.visit_duration ] end defp export_sources_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [ selected_as(:date), selected_as(:source), s.referrer, s.utm_source, s.utm_medium, s.utm_campaign, s.utm_content, s.utm_term ], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), selected_as(s.referrer_source, :source), s.referrer, s.utm_source, s.utm_medium, s.utm_campaign, s.utm_content, s.utm_term, pageviews(s), visitors(s), visits(s), visit_duration(s), bounces(s) ] end defp export_pages_q(site, timezone, date_range) do base_q = from(e in sampled("events_v2"), where: ^export_filter(site.id, date_range), where: [name: "pageview"], group_by: [selected_as(:date), selected_as(:page)], order_by: selected_as(:date) ) max_scroll_depth_per_session_q = from(e in "events_v2", where: ^export_filter(site.id, date_range), where: e.name == "engagement" and e.scroll_depth <= 100, select: %{ date: date(e.timestamp, ^timezone), page: selected_as(e.pathname, :page), session_id: e.session_id, max_scroll_depth: max(e.scroll_depth) }, group_by: [e.session_id, selected_as(:date), selected_as(:page)] ) scroll_depth_q = from(p in subquery(max_scroll_depth_per_session_q), select: %{ date: p.date, page: p.page, total_scroll_depth: fragment("sum(?)", p.max_scroll_depth), total_scroll_depth_visits: fragment("uniq(?)", p.session_id) }, group_by: [:date, :page] ) from(e in base_q, left_join: s in subquery(scroll_depth_q), on: s.date == selected_as(:date) and s.page == selected_as(:page), select: %{ date: date(e.timestamp, ^timezone), hostname: selected_as(fragment("any(?)", e.hostname), :hostname), page: selected_as(e.pathname, :page), visits: selected_as( scale_sample(fragment("uniq(?)", e.session_id)), :visits ), visitors: visitors(e), pageviews: selected_as(scale_sample(fragment("count()")), :pageviews), total_scroll_depth: selected_as(fragment("any(?)", s.total_scroll_depth), :total_scroll_depth), total_scroll_depth_visits: selected_as(fragment("any(?)", s.total_scroll_depth_visits), :total_scroll_depth_visits) } ) |> add_time_on_page_columns(site.id, timezone, date_range) end defp add_time_on_page_columns(q, site_id, timezone, date_range) do site = Plausible.Repo.get(Plausible.Site, site_id) if Plausible.Stats.TimeOnPage.new_time_on_page_visible?(site) do cutoff = Plausible.Stats.TimeOnPage.legacy_time_on_page_cutoff(site) engagements_q = from(e in sampled("events_v2"), where: ^export_filter(site_id, date_range), where: e.name == "engagement", group_by: [selected_as(:date), selected_as(:page)], order_by: selected_as(:date), select: %{ date: date(e.timestamp, ^timezone), page: selected_as(e.pathname, :page), total_time_on_page: fragment( "toUInt64(round(sumIf(?, ? >= ?) / 1000))", e.engagement_time, e.timestamp, ^cutoff ), total_time_on_page_visits: fragment("uniqIf(?, ? >= ?)", e.session_id, e.timestamp, ^cutoff) } ) q |> join(:left, [], s in subquery(engagements_q), on: s.date == selected_as(:date) and s.page == selected_as(:page) ) |> select_merge_as([..., s], %{ total_time_on_page: fragment("any(?)", s.total_time_on_page), total_time_on_page_visits: fragment("any(?)", s.total_time_on_page_visits) }) else q end end defp export_entry_pages_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [selected_as(:date), s.entry_page], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), s.entry_page, visitors(s), selected_as( scale_sample(fragment("greatest(sum(?),0)", s.sign)), :entrances ), visit_duration(s), bounces(s), pageviews(s) ] end defp export_exit_pages_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [selected_as(:date), s.exit_page], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), s.exit_page, visitors(s), visit_duration(s), selected_as( scale_sample(fragment("greatest(sum(?),0)", s.sign)), :exits ), bounces(s), pageviews(s) ] end defp export_custom_events_q(site, timezone, date_range) do from e in sampled("events_v2"), where: ^export_filter(site.id, date_range), where: e.name != "pageview", group_by: [ selected_as(:date), e.name, selected_as(:link_url), selected_as(:path) ], order_by: selected_as(:date), select: [ date(e.timestamp, ^timezone), e.name, selected_as( fragment( "if(? in ?, ?, '')", e.name, ^Plausible.Event.SystemEvents.events_with_url_prop(), get_by_key(e, :meta, "url") ), :link_url ), selected_as( fragment( "if(? in ?, ?, '')", e.name, ^Plausible.Event.SystemEvents.events_with_path_prop(), get_by_key(e, :meta, "path") ), :path ), visitors(e), selected_as(scale_sample(fragment("count()")), :events) ] end defp export_custom_props_q(site, timezone, date_range) do query = from e in sampled("events_v2"), join: pv in fragment("arrayZip(`meta.key`, `meta.value`)"), on: true, hints: "ARRAY", where: ^export_filter(site.id, date_range), group_by: [ selected_as(:date), selected_as(:property), selected_as(:value) ], order_by: selected_as(:date), select: [ date(e.timestamp, ^timezone), selected_as(fragment("tupleElement(?, 1)", pv), :property), selected_as(fragment("tupleElement(?, 2)", pv), :value), visitors(e), selected_as(scale_sample(fragment("count()")), :events) ] if Plausible.Billing.Feature.Props.enabled?(site) do query else where(query, [], selected_as(:property) in ^Plausible.Props.internal_keys()) end end defp export_locations_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), where: s.country_code != "\0\0" and s.country_code != "ZZ", group_by: [selected_as(:date), s.country_code, s.subdivision1_code, s.city_geoname_id], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), selected_as(s.country_code, :country), selected_as(s.subdivision1_code, :region), selected_as(s.city_geoname_id, :city), visitors(s), visits(s), visit_duration(s), bounces(s), pageviews(s) ] end defp export_devices_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [selected_as(:date), s.screen_size], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), selected_as(s.screen_size, :device), visitors(s), visits(s), visit_duration(s), bounces(s), pageviews(s) ] end defp export_browsers_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [selected_as(:date), s.browser, s.browser_version], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), s.browser, s.browser_version, visitors(s), visits(s), visit_duration(s), bounces(s), pageviews(s) ] end defp export_operating_systems_q(site, timezone, date_range) do from s in sampled("sessions_v2"), where: ^export_filter(site.id, date_range), group_by: [selected_as(:date), s.operating_system, s.operating_system_version], order_by: selected_as(:date), select: [ date(s.timestamp, ^timezone), s.operating_system, s.operating_system_version, visitors(s), visits(s), visit_duration(s), bounces(s), pageviews(s) ] end @doc """ Creates a streamable Zip archive from the provided (named) Ecto queries. Example usage: {:ok, pool} = Ch.start_link(pool_size: 1) DBConnection.run(pool, fn conn -> conn |> stream_archive(export_queries(site), format: "CSVWithNames") |> Stream.into(File.stream!("export.zip")) |> Stream.run() end) """ @spec stream_archive(DBConnection.t(), %{String.t() => Ecto.Query.t()}, [Ch.query_option()]) :: Enumerable.t() def stream_archive(conn, named_queries, opts \\ []) do entries = Enum.map(named_queries, fn {name, query} -> {sql, params} = Plausible.ClickhouseRepo.to_sql(:all, query) datastream = conn |> Ch.stream(sql, params, opts) |> Stream.map(fn %Ch.Result{data: data} -> data end) Zstream.entry(name, datastream, coder: Zstream.Coder.Stored) end) Zstream.zip(entries) end end