git.squeep.com Git - akkoma/blob - lib/pleroma/migrators/hashtags_table_migrator.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Migrators.HashtagsTableMigrator do
   6   use GenServer
   7
   8   require Logger
   9
  10   import Ecto.Query
  11
  12   alias __MODULE__.State
  13   alias Pleroma.Config
  14   alias Pleroma.Hashtag
  15   alias Pleroma.Object
  16   alias Pleroma.Repo
  17
  18   defdelegate data_migration(), to: Pleroma.DataMigration, as: :populate_hashtags_table
  19   defdelegate data_migration_id(), to: State
  20
  21   defdelegate state(), to: State
  22   defdelegate persist_state(), to: State, as: :persist_to_db
  23   defdelegate get_stat(key, value \\ nil), to: State, as: :get_data_key
  24   defdelegate put_stat(key, value), to: State, as: :put_data_key
  25   defdelegate increment_stat(key, increment), to: State, as: :increment_data_key
  26
  27   @feature_config_path [:database, :improved_hashtag_timeline]
  28   @reg_name {:global, __MODULE__}
  29
  30   def whereis, do: GenServer.whereis(@reg_name)
  31
  32   def feature_state, do: Config.get(@feature_config_path)
  33
  34   def start_link(_) do
  35     case whereis() do
  36       nil ->
  37         GenServer.start_link(__MODULE__, nil, name: @reg_name)
  38
  39       pid ->
  40         {:ok, pid}
  41     end
  42   end
  43
  44   @impl true
  45   def init(_) do
  46     {:ok, nil, {:continue, :init_state}}
  47   end
  48
  49   @impl true
  50   def handle_continue(:init_state, _state) do
  51     {:ok, _} = State.start_link(nil)
  52
  53     data_migration = data_migration()
  54     manual_migrations = Config.get([:instance, :manual_data_migrations], [])
  55
  56     cond do
  57       Config.get(:env) == :test ->
  58         update_status(:noop)
  59
  60       is_nil(data_migration) ->
  61         message = "Data migration does not exist."
  62         update_status(:failed, message)
  63         Logger.error("#{__MODULE__}: #{message}")
  64
  65       data_migration.state == :manual or data_migration.name in manual_migrations ->
  66         message = "Data migration is in manual execution or manual fix mode."
  67         update_status(:manual, message)
  68         Logger.warn("#{__MODULE__}: #{message}")
  69
  70       data_migration.state == :complete ->
  71         on_complete(data_migration)
  72
  73       true ->
  74         send(self(), :migrate_hashtags)
  75     end
  76
  77     {:noreply, nil}
  78   end
  79
  80   @impl true
  81   def handle_info(:migrate_hashtags, state) do
  82     State.reinit()
  83
  84     update_status(:running)
  85     put_stat(:started_at, NaiveDateTime.utc_now())
  86
  87     data_migration_id = data_migration_id()
  88     max_processed_id = get_stat(:max_processed_id, 0)
  89
  90     Logger.info("Transferring embedded hashtags to `hashtags` (from oid: #{max_processed_id})...")
  91
  92     query()
  93     |> where([object], object.id > ^max_processed_id)
  94     |> Repo.chunk_stream(100, :batches, timeout: :infinity)
  95     |> Stream.each(fn objects ->
  96       object_ids = Enum.map(objects, & &1.id)
  97
  98       results = Enum.map(objects, &transfer_object_hashtags(&1))
  99
 100       failed_ids =
 101         results
 102         |> Enum.filter(&(elem(&1, 0) == :error))
 103         |> Enum.map(&elem(&1, 1))
 104
 105       # Count of objects with hashtags (`{:noop, id}` is returned for objects having other AS2 tags)
 106       chunk_affected_count =
 107         results
 108         |> Enum.filter(&(elem(&1, 0) == :ok))
 109         |> length()
 110
 111       for failed_id <- failed_ids do
 112         _ =
 113           Repo.query(
 114             "INSERT INTO data_migration_failed_ids(data_migration_id, record_id) " <>
 115               "VALUES ($1, $2) ON CONFLICT DO NOTHING;",
 116             [data_migration_id, failed_id]
 117           )
 118       end
 119
 120       _ =
 121         Repo.query(
 122           "DELETE FROM data_migration_failed_ids " <>
 123             "WHERE data_migration_id = $1 AND record_id = ANY($2)",
 124           [data_migration_id, object_ids -- failed_ids]
 125         )
 126
 127       max_object_id = Enum.at(object_ids, -1)
 128
 129       put_stat(:max_processed_id, max_object_id)
 130       increment_stat(:processed_count, length(object_ids))
 131       increment_stat(:failed_count, length(failed_ids))
 132       increment_stat(:affected_count, chunk_affected_count)
 133       put_stat(:records_per_second, records_per_second())
 134       persist_state()
 135
 136       # A quick and dirty approach to controlling the load this background migration imposes
 137       sleep_interval = Config.get([:populate_hashtags_table, :sleep_interval_ms], 0)
 138       Process.sleep(sleep_interval)
 139     end)
 140     |> Stream.run()
 141
 142     fault_rate = fault_rate()
 143     put_stat(:fault_rate, fault_rate)
 144     fault_rate_allowance = Config.get([:populate_hashtags_table, :fault_rate_allowance], 0)
 145
 146     cond do
 147       fault_rate == 0 ->
 148         set_complete()
 149
 150       is_float(fault_rate) and fault_rate <= fault_rate_allowance ->
 151         message = """
 152         Done with fault rate of #{fault_rate} which doesn't exceed #{fault_rate_allowance}.
 153         Putting data migration to manual fix mode. Check `retry_failed/0`.
 154         """
 155
 156         Logger.warn("#{__MODULE__}: #{message}")
 157         update_status(:manual, message)
 158         on_complete(data_migration())
 159
 160       true ->
 161         message = "Too many failures. Check data_migration_failed_ids records / `retry_failed/0`."
 162         Logger.error("#{__MODULE__}: #{message}")
 163         update_status(:failed, message)
 164     end
 165
 166     persist_state()
 167     {:noreply, state}
 168   end
 169
 170   def fault_rate do
 171     with failures_count when is_integer(failures_count) <- failures_count() do
 172       failures_count / Enum.max([get_stat(:affected_count, 0), 1])
 173     else
 174       _ -> :error
 175     end
 176   end
 177
 178   defp records_per_second do
 179     get_stat(:processed_count, 0) / Enum.max([running_time(), 1])
 180   end
 181
 182   defp running_time do
 183     NaiveDateTime.diff(NaiveDateTime.utc_now(), get_stat(:started_at, NaiveDateTime.utc_now()))
 184   end
 185
 186   @hashtags_objects_cleanup_query """
 187   DELETE FROM hashtags_objects WHERE object_id IN
 188     (SELECT DISTINCT objects.id FROM objects
 189       JOIN hashtags_objects ON hashtags_objects.object_id = objects.id LEFT JOIN activities
 190         ON COALESCE(activities.data->'object'->>'id', activities.data->>'object') =
 191           (objects.data->>'id')
 192         AND activities.data->>'type' = 'Create'
 193       WHERE activities.id IS NULL);
 194   """
 195
 196   @hashtags_cleanup_query """
 197   DELETE FROM hashtags WHERE id IN
 198     (SELECT hashtags.id FROM hashtags
 199       LEFT OUTER JOIN hashtags_objects
 200         ON hashtags_objects.hashtag_id = hashtags.id
 201       WHERE hashtags_objects.hashtag_id IS NULL);
 202   """
 203
 204   @doc """
 205   Deletes `hashtags_objects` for legacy objects not asoociated with Create activity.
 206   Also deletes unreferenced `hashtags` records (might occur after deletion of `hashtags_objects`).
 207   """
 208   def delete_non_create_activities_hashtags do
 209     {:ok, %{num_rows: hashtags_objects_count}} =
 210       Repo.query(@hashtags_objects_cleanup_query, [], timeout: :infinity)
 211
 212     {:ok, %{num_rows: hashtags_count}} =
 213       Repo.query(@hashtags_cleanup_query, [], timeout: :infinity)
 214
 215     {:ok, hashtags_objects_count, hashtags_count}
 216   end
 217
 218   defp query do
 219     # Note: most objects have Mention-type AS2 tags and no hashtags (but we can't filter them out)
 220     # Note: not checking activity type, expecting remove_non_create_objects_hashtags/_ to clean up
 221     from(
 222       object in Object,
 223       where:
 224         fragment("(?)->'tag' IS NOT NULL AND (?)->'tag' != '[]'::jsonb", object.data, object.data),
 225       select: %{
 226         id: object.id,
 227         tag: fragment("(?)->'tag'", object.data)
 228       }
 229     )
 230     |> join(:left, [o], hashtags_objects in fragment("SELECT object_id FROM hashtags_objects"),
 231       on: hashtags_objects.object_id == o.id
 232     )
 233     |> where([_o, hashtags_objects], is_nil(hashtags_objects.object_id))
 234   end
 235
 236   @spec transfer_object_hashtags(Map.t()) :: {:noop | :ok | :error, integer()}
 237   defp transfer_object_hashtags(object) do
 238     embedded_tags = if Map.has_key?(object, :tag), do: object.tag, else: object.data["tag"]
 239     hashtags = Object.object_data_hashtags(%{"tag" => embedded_tags})
 240
 241     if Enum.any?(hashtags) do
 242       transfer_object_hashtags(object, hashtags)
 243     else
 244       {:noop, object.id}
 245     end
 246   end
 247
 248   defp transfer_object_hashtags(object, hashtags) do
 249     Repo.transaction(fn ->
 250       with {:ok, hashtag_records} <- Hashtag.get_or_create_by_names(hashtags) do
 251         maps = Enum.map(hashtag_records, &%{hashtag_id: &1.id, object_id: object.id})
 252         base_error = "ERROR when inserting hashtags_objects for object with id #{object.id}"
 253
 254         try do
 255           with {rows_count, _} when is_integer(rows_count) <-
 256                  Repo.insert_all("hashtags_objects", maps, on_conflict: :nothing) do
 257             object.id
 258           else
 259             e ->
 260               Logger.error("#{base_error}: #{inspect(e)}")
 261               Repo.rollback(object.id)
 262           end
 263         rescue
 264           e ->
 265             Logger.error("#{base_error}: #{inspect(e)}")
 266             Repo.rollback(object.id)
 267         end
 268       else
 269         e ->
 270           error = "ERROR: could not create hashtags for object #{object.id}: #{inspect(e)}"
 271           Logger.error(error)
 272           Repo.rollback(object.id)
 273       end
 274     end)
 275   end
 276
 277   @doc "Approximate count for current iteration (including processed records count)"
 278   def count(force \\ false, timeout \\ :infinity) do
 279     stored_count = get_stat(:count)
 280
 281     if stored_count && !force do
 282       stored_count
 283     else
 284       processed_count = get_stat(:processed_count, 0)
 285       max_processed_id = get_stat(:max_processed_id, 0)
 286       query = where(query(), [object], object.id > ^max_processed_id)
 287
 288       count = Repo.aggregate(query, :count, :id, timeout: timeout) + processed_count
 289       put_stat(:count, count)
 290       persist_state()
 291
 292       count
 293     end
 294   end
 295
 296   defp on_complete(data_migration) do
 297     cond do
 298       data_migration.feature_lock ->
 299         :noop
 300
 301       not is_nil(feature_state()) ->
 302         :noop
 303
 304       true ->
 305         Config.put(@feature_config_path, true)
 306         :ok
 307     end
 308   end
 309
 310   def failed_objects_query do
 311     from(o in Object)
 312     |> join(:inner, [o], dmf in fragment("SELECT * FROM data_migration_failed_ids"),
 313       on: dmf.record_id == o.id
 314     )
 315     |> where([_o, dmf], dmf.data_migration_id == ^data_migration_id())
 316     |> order_by([o], asc: o.id)
 317   end
 318
 319   def failures_count do
 320     with {:ok, %{rows: [[count]]}} <-
 321            Repo.query(
 322              "SELECT COUNT(record_id) FROM data_migration_failed_ids WHERE data_migration_id = $1;",
 323              [data_migration_id()]
 324            ) do
 325       count
 326     end
 327   end
 328
 329   def retry_failed do
 330     data_migration_id = data_migration_id()
 331
 332     failed_objects_query()
 333     |> Repo.chunk_stream(100, :one)
 334     |> Stream.each(fn object ->
 335       with {res, _} when res != :error <- transfer_object_hashtags(object) do
 336         _ =
 337           Repo.query(
 338             "DELETE FROM data_migration_failed_ids " <>
 339               "WHERE data_migration_id = $1 AND record_id = $2",
 340             [data_migration_id, object.id]
 341           )
 342       end
 343     end)
 344     |> Stream.run()
 345
 346     put_stat(:failed_count, failures_count())
 347     persist_state()
 348
 349     force_continue()
 350   end
 351
 352   def force_continue do
 353     send(whereis(), :migrate_hashtags)
 354   end
 355
 356   def force_restart do
 357     :ok = State.reset()
 358     force_continue()
 359   end
 360
 361   def set_complete do
 362     update_status(:complete)
 363     persist_state()
 364     on_complete(data_migration())
 365   end
 366
 367   defp update_status(status, message \\ nil) do
 368     put_stat(:state, status)
 369     put_stat(:message, message)
 370   end
 371 end