lib/workers/clickhouse_clean_sites.ex

defmodule Plausible.Workers.ClickhouseCleanSites do
  @moduledoc """
  Cleans deleted site data from ClickHouse asynchronously.

  We batch up data deletions from ClickHouse as deleting a single site is
  just as expensive as deleting many.
  """

  use Plausible.Repo
  use Plausible.ClickhouseRepo
  use Plausible.IngestRepo
  use Oban.Worker, queue: :clickhouse_clean_sites

  import Ecto.Query

  require Logger

  @tables_to_clear [
    "events_v2",
    "sessions_v2",
    "ingest_counters",
    "imported_browsers",
    "imported_devices",
    "imported_entry_pages",
    "imported_exit_pages",
    "imported_locations",
    "imported_operating_systems",
    "imported_pages",
    "imported_custom_events",
    "imported_sources",
    "imported_visitors"
  ]

  @settings if Mix.env() in [:test, :ce_test, :e2e_test], do: [mutations_sync: 2], else: []

  def perform(_job) do
    deleted_sites = get_deleted_sites_with_clickhouse_data()

    if not Enum.empty?(deleted_sites) do
      Logger.notice(
        "Clearing ClickHouse data for the following #{length(deleted_sites)} sites which have been deleted: #{inspect(deleted_sites)}"
      )

      for table <- @tables_to_clear do
        IngestRepo.query!(
          "ALTER TABLE {$0:Identifier} DELETE WHERE site_id IN {$1:Array(UInt64)}",
          [table, deleted_sites],
          settings: @settings
        )
      end
    end

    :ok
  end

  def get_deleted_sites_with_clickhouse_data() do
    pg_sites =
      from(s in Plausible.Site.regular(), select: s.id)
      |> Plausible.Repo.all()
      |> MapSet.new()

    {:ok, ch} =
      Plausible.ClickhouseRepo.get_config_without_ch_query_execution_timeout()
      |> Ch.start_link()

    %Ch.Result{columns: ["site_id"], rows: rows} =
      DBConnection.run(
        ch,
        fn conn ->
          Ch.query!(conn, "FROM events_v2 SELECT site_id GROUP BY site_id", [],
            timeout: :infinity
          )
        end,
        timeout: :infinity
      )

    ch_sites = rows |> MapSet.new(fn [site_id] -> site_id end)

    MapSet.difference(ch_sites, pg_sites) |> MapSet.to_list()
  end
end