File size: 2,117 Bytes
c27e67a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | 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
|