From f3e061c9645571997a01b1091d0e8a3f68c6bb21 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Sat, 6 Aug 2022 03:24:31 +0200 Subject: Object: remove context_id field 30 to 70% of the objects in the object table are simple JSON objects containing a single field, 'id', being the context's ID. The reason for the creation of an object per context seems to be an old relic from the StatusNet era, and has only been used nowadays as an helper for threads in Pleroma-FE via the `pleroma.conversation_id` field in status views. An object per context was created, and its numerical ID (table column) was used and stored as 'context_id' in the object and activity along with the full 'context' URI/string. This commit removes this field and stops creation of objects for each context, which will also allow incoming activities to use activity IDs as contexts, something which was not possible before, or would have been very broken under most circumstances. The `pleroma.conversation_id` field has been reimplemented in a way to maintain backwards-compatibility by calculating a CRC32 of the full context URI/string in the object, instead of relying on the row ID for the created context object. --- lib/pleroma/object.ex | 4 --- .../article_note_page_validator.ex | 2 +- .../activity_pub/object_validators/common_fixes.ex | 4 +-- .../object_validators/event_validator.ex | 2 +- .../object_validators/question_validator.ex | 2 +- lib/pleroma/web/activity_pub/utils.ex | 23 ++--------------- lib/pleroma/web/common_api/utils.ex | 29 ---------------------- lib/pleroma/web/mastodon_api/views/status_view.ex | 2 +- 8 files changed, 7 insertions(+), 61 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/object.ex b/lib/pleroma/object.ex index fe264b5e0..c214a79c5 100644 --- a/lib/pleroma/object.ex +++ b/lib/pleroma/object.ex @@ -208,10 +208,6 @@ defmodule Pleroma.Object do end end - def context_mapping(context) do - Object.change(%Object{}, %{data: %{"id" => context}}) - end - def make_tombstone(%Object{data: %{"id" => id, "type" => type}}, deleted \\ DateTime.utc_now()) do %ObjectTombstone{ id: id, diff --git a/lib/pleroma/web/activity_pub/object_validators/article_note_page_validator.ex b/lib/pleroma/web/activity_pub/object_validators/article_note_page_validator.ex index 57c8d1dc0..c5fb94034 100644 --- a/lib/pleroma/web/activity_pub/object_validators/article_note_page_validator.ex +++ b/lib/pleroma/web/activity_pub/object_validators/article_note_page_validator.ex @@ -94,7 +94,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.ArticleNotePageValidator do defp validate_data(data_cng) do data_cng |> validate_inclusion(:type, ["Article", "Note", "Page"]) - |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id]) + |> validate_required([:id, :actor, :attributedTo, :type, :context]) |> CommonValidations.validate_any_presence([:cc, :to]) |> CommonValidations.validate_fields_match([:actor, :attributedTo]) |> CommonValidations.validate_actor_presence() diff --git a/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex b/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex index 4f8c083eb..c7e292bec 100644 --- a/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex +++ b/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex @@ -22,14 +22,12 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonFixes do end def fix_object_defaults(data) do - %{data: %{"id" => context}, id: context_id} = - Utils.create_context(data["context"] || data["conversation"]) + context = Utils.maybe_create_context(data["context"] || data["conversation"]) %User{follower_address: follower_collection} = User.get_cached_by_ap_id(data["attributedTo"]) data |> Map.put("context", context) - |> Map.put("context_id", context_id) |> cast_and_filter_recipients("to", follower_collection) |> cast_and_filter_recipients("cc", follower_collection) |> cast_and_filter_recipients("bto", follower_collection) diff --git a/lib/pleroma/web/activity_pub/object_validators/event_validator.ex b/lib/pleroma/web/activity_pub/object_validators/event_validator.ex index 0e99f2037..ab204f69a 100644 --- a/lib/pleroma/web/activity_pub/object_validators/event_validator.ex +++ b/lib/pleroma/web/activity_pub/object_validators/event_validator.ex @@ -62,7 +62,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.EventValidator do defp validate_data(data_cng) do data_cng |> validate_inclusion(:type, ["Event"]) - |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id]) + |> validate_required([:id, :actor, :attributedTo, :type, :context]) |> CommonValidations.validate_any_presence([:cc, :to]) |> CommonValidations.validate_fields_match([:actor, :attributedTo]) |> CommonValidations.validate_actor_presence() diff --git a/lib/pleroma/web/activity_pub/object_validators/question_validator.ex b/lib/pleroma/web/activity_pub/object_validators/question_validator.ex index 9412be4bc..ce3305142 100644 --- a/lib/pleroma/web/activity_pub/object_validators/question_validator.ex +++ b/lib/pleroma/web/activity_pub/object_validators/question_validator.ex @@ -80,7 +80,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.QuestionValidator do defp validate_data(data_cng) do data_cng |> validate_inclusion(:type, ["Question"]) - |> validate_required([:id, :actor, :attributedTo, :type, :context, :context_id]) + |> validate_required([:id, :actor, :attributedTo, :type, :context]) |> CommonValidations.validate_any_presence([:cc, :to]) |> CommonValidations.validate_fields_match([:actor, :attributedTo]) |> CommonValidations.validate_actor_presence() diff --git a/lib/pleroma/web/activity_pub/utils.ex b/lib/pleroma/web/activity_pub/utils.ex index 9cde7805c..d3b7d804f 100644 --- a/lib/pleroma/web/activity_pub/utils.ex +++ b/lib/pleroma/web/activity_pub/utils.ex @@ -154,22 +154,7 @@ defmodule Pleroma.Web.ActivityPub.Utils do Notification.get_notified_from_activity(%Activity{data: object}, false) end - def create_context(context) do - context = context || generate_id("contexts") - - # Ecto has problems accessing the constraint inside the jsonb, - # so we explicitly check for the existed object before insert - object = Object.get_cached_by_ap_id(context) - - with true <- is_nil(object), - changeset <- Object.context_mapping(context), - {:ok, inserted_object} <- Repo.insert(changeset) do - inserted_object - else - _ -> - object - end - end + def maybe_create_context(context), do: context || generate_id("contexts") @doc """ Enqueues an activity for federation if it's local @@ -201,18 +186,16 @@ defmodule Pleroma.Web.ActivityPub.Utils do |> Map.put_new("id", "pleroma:fakeid") |> Map.put_new_lazy("published", &make_date/0) |> Map.put_new("context", "pleroma:fakecontext") - |> Map.put_new("context_id", -1) |> lazy_put_object_defaults(true) end def lazy_put_activity_defaults(map, _fake?) do - %{data: %{"id" => context}, id: context_id} = create_context(map["context"]) + context = maybe_create_context(map["context"]) map |> Map.put_new_lazy("id", &generate_activity_id/0) |> Map.put_new_lazy("published", &make_date/0) |> Map.put_new("context", context) - |> Map.put_new("context_id", context_id) |> lazy_put_object_defaults(false) end @@ -226,7 +209,6 @@ defmodule Pleroma.Web.ActivityPub.Utils do |> Map.put_new("id", "pleroma:fake_object_id") |> Map.put_new_lazy("published", &make_date/0) |> Map.put_new("context", activity["context"]) - |> Map.put_new("context_id", activity["context_id"]) |> Map.put_new("fake", true) %{activity | "object" => object} @@ -239,7 +221,6 @@ defmodule Pleroma.Web.ActivityPub.Utils do |> Map.put_new_lazy("id", &generate_object_id/0) |> Map.put_new_lazy("published", &make_date/0) |> Map.put_new("context", activity["context"]) - |> Map.put_new("context_id", activity["context_id"]) %{activity | "object" => object} end diff --git a/lib/pleroma/web/common_api/utils.ex b/lib/pleroma/web/common_api/utils.ex index ce850b038..052bd7770 100644 --- a/lib/pleroma/web/common_api/utils.ex +++ b/lib/pleroma/web/common_api/utils.ex @@ -449,35 +449,6 @@ defmodule Pleroma.Web.CommonAPI.Utils do def get_report_statuses(_, _), do: {:ok, nil} - # DEPRECATED mostly, context objects are now created at insertion time. - def context_to_conversation_id(context) do - with %Object{id: id} <- Object.get_cached_by_ap_id(context) do - id - else - _e -> - changeset = Object.context_mapping(context) - - case Repo.insert(changeset) do - {:ok, %{id: id}} -> - id - - # This should be solved by an upsert, but it seems ecto - # has problems accessing the constraint inside the jsonb. - {:error, _} -> - Object.get_cached_by_ap_id(context).id - end - end - end - - def conversation_id_to_context(id) do - with %Object{data: %{"id" => context}} <- Repo.get(Object, id) do - context - else - _e -> - {:error, dgettext("errors", "No such conversation")} - end - end - def validate_character_limit("" = _full_payload, [] = _attachments) do {:error, dgettext("errors", "Cannot post an empty status without attachments")} end diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex index 1ebfd6740..31a0c420f 100644 --- a/lib/pleroma/web/mastodon_api/views/status_view.ex +++ b/lib/pleroma/web/mastodon_api/views/status_view.ex @@ -61,7 +61,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do do: context_id defp get_context_id(%{data: %{"context" => context}}) when is_binary(context), - do: Utils.context_to_conversation_id(context) + do: :erlang.crc32(context) defp get_context_id(_), do: nil -- cgit v1.2.3 From 7f71e3d0fe347920834be8c8e28d9c7f5b169e9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Sat, 6 Aug 2022 03:32:58 +0200 Subject: CommonFields: remove context_id --- lib/pleroma/web/activity_pub/object_validators/common_fields.ex | 2 -- lib/pleroma/web/mastodon_api/views/status_view.ex | 3 --- 2 files changed, 5 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/object_validators/common_fields.ex b/lib/pleroma/web/activity_pub/object_validators/common_fields.ex index 8e768ffbf..095bd0da2 100644 --- a/lib/pleroma/web/activity_pub/object_validators/common_fields.ex +++ b/lib/pleroma/web/activity_pub/object_validators/common_fields.ex @@ -51,8 +51,6 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonFields do field(:summary, :string) field(:context, :string) - # short identifier for PleromaFE to group statuses by context - field(:context_id, :integer) field(:sensitive, :boolean, default: false) field(:replies_count, :integer, default: 0) diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex index 31a0c420f..3ffe55c5d 100644 --- a/lib/pleroma/web/mastodon_api/views/status_view.ex +++ b/lib/pleroma/web/mastodon_api/views/status_view.ex @@ -57,9 +57,6 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do end) end - defp get_context_id(%{data: %{"context_id" => context_id}}) when not is_nil(context_id), - do: context_id - defp get_context_id(%{data: %{"context" => context}}) when is_binary(context), do: :erlang.crc32(context) -- cgit v1.2.3 From a9111bcaf2ba2371a1021dc171dab50615a4c040 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Sun, 7 Aug 2022 20:37:17 +0200 Subject: StatusView: clear MSB on calculated conversation_id This field seems to be a left-over from the StatusNet era. If your application uses `pleroma.conversation_id`: this field is deprecated. It is currently stubbed instead by doing a CRC32 of the context, and clearing the MSB to avoid overflow exceptions with signed integers on the different clients using this field (Java/Kotlin code, mostly; see Husky and probably other mobile clients.) This should be removed in a future version of Pleroma. Pleroma-FE currently depends on this field, as well. --- lib/pleroma/web/mastodon_api/views/status_view.ex | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) (limited to 'lib') diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex index 3ffe55c5d..5cb524f56 100644 --- a/lib/pleroma/web/mastodon_api/views/status_view.ex +++ b/lib/pleroma/web/mastodon_api/views/status_view.ex @@ -57,8 +57,19 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do end) end - defp get_context_id(%{data: %{"context" => context}}) when is_binary(context), - do: :erlang.crc32(context) + # DEPRECATED This field seems to be a left-over from the StatusNet era. + # If your application uses `pleroma.conversation_id`: this field is deprecated. + # It is currently stubbed instead by doing a CRC32 of the context, and + # clearing the MSB to avoid overflow exceptions with signed integers on the + # different clients using this field (Java/Kotlin code, mostly; see Husky.) + # This should be removed in a future version of Pleroma. Pleroma-FE currently + # depends on this field, as well. + defp get_context_id(%{data: %{"context" => context}}) when is_binary(context) do + use Bitwise + + :erlang.crc32(context) + |> band(bnot(0x8000_0000)) + end defp get_context_id(_), do: nil -- cgit v1.2.3 From def0f5dc2e76b7c4ac22b393abf7f5de5e197659 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Sun, 7 Aug 2022 20:39:35 +0200 Subject: StatusView: implement pleroma.context field This field replaces the now deprecated conversation_id field, and now exposes the ActivityPub object `context` directly via the MastoAPI instead of relying on StatusNet-era data concepts. --- lib/pleroma/web/api_spec/schemas/status.ex | 9 ++++++++- lib/pleroma/web/mastodon_api/views/status_view.ex | 1 + 2 files changed, 9 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/api_spec/schemas/status.ex b/lib/pleroma/web/api_spec/schemas/status.ex index 6e6e30315..8c19a9d9f 100644 --- a/lib/pleroma/web/api_spec/schemas/status.ex +++ b/lib/pleroma/web/api_spec/schemas/status.ex @@ -142,9 +142,15 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Status do description: "A map consisting of alternate representations of the `content` property with the key being it's mimetype. Currently the only alternate representation supported is `text/plain`" }, + context: %Schema{ + type: :string, + description: "The thread identifier the status is associated with" + }, conversation_id: %Schema{ type: :integer, - description: "The ID of the AP context the status is associated with (if any)" + deprecated: true, + description: + "The ID of the AP context the status is associated with (if any); deprecated, please use `context` instead" }, direct_conversation_id: %Schema{ type: :integer, @@ -319,6 +325,7 @@ defmodule Pleroma.Web.ApiSpec.Schemas.Status do "pinned" => false, "pleroma" => %{ "content" => %{"text/plain" => "foobar"}, + "context" => "http://localhost:4001/objects/8b4c0c80-6a37-4d2a-b1b9-05a19e3875aa", "conversation_id" => 345_972, "direct_conversation_id" => nil, "emoji_reactions" => [], diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex index 5cb524f56..a4d6cd807 100644 --- a/lib/pleroma/web/mastodon_api/views/status_view.ex +++ b/lib/pleroma/web/mastodon_api/views/status_view.ex @@ -375,6 +375,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do pleroma: %{ local: activity.local, conversation_id: get_context_id(activity), + context: object.data["context"], in_reply_to_account_acct: reply_to_user && reply_to_user.nickname, content: %{"text/plain" => content_plaintext}, spoiler_text: %{"text/plain" => summary}, -- cgit v1.2.3 From 3b6784b1de8454ab8c009ac688f6c62039117742 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Tue, 2 Aug 2022 17:30:36 +0200 Subject: CreateGenericValidator: fix reply context fixing Incoming Pleroma replies to a Misskey thread were rejected due to a broken context fix, which caused them to not be visible until a non-Pleroma user interacted with the replies. This fix properly sets the post-fix object context to its parent Create activity as well, if it was changed. --- .../web/activity_pub/object_validators/create_generic_validator.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/object_validators/create_generic_validator.ex b/lib/pleroma/web/activity_pub/object_validators/create_generic_validator.ex index c9a621cb1..2395abfd4 100644 --- a/lib/pleroma/web/activity_pub/object_validators/create_generic_validator.ex +++ b/lib/pleroma/web/activity_pub/object_validators/create_generic_validator.ex @@ -75,7 +75,7 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CreateGenericValidator do data |> CommonFixes.fix_actor() - |> Map.put_new("context", object["context"]) + |> Map.put("context", object["context"]) |> fix_addressing(object) end -- cgit v1.2.3 From bb02ee99f58e378e33162211f41fe5979d5da8ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Wed, 10 Aug 2022 04:21:28 +0200 Subject: CommonFixes: more predictable context generation `context` fields for objects and activities can now be generated based on the object/activity `inReplyTo` field or its ActivityPub ID, as a fallback method in cases where `context` fields are missing for incoming activities and objects. --- lib/pleroma/web/activity_pub/object_validators/common_fixes.ex | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) (limited to 'lib') diff --git a/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex b/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex index c7e292bec..add46d561 100644 --- a/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex +++ b/lib/pleroma/web/activity_pub/object_validators/common_fixes.ex @@ -22,7 +22,10 @@ defmodule Pleroma.Web.ActivityPub.ObjectValidators.CommonFixes do end def fix_object_defaults(data) do - context = Utils.maybe_create_context(data["context"] || data["conversation"]) + context = + Utils.maybe_create_context( + data["context"] || data["conversation"] || data["inReplyTo"] || data["id"] + ) %User{follower_address: follower_collection} = User.get_cached_by_ap_id(data["attributedTo"]) -- cgit v1.2.3 From 88c1c76d3eca3412d1e02008f1b8d96fe8fe0b96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9l=C3=A8ne?= Date: Mon, 15 Aug 2022 01:15:23 +0200 Subject: Migrations: delete contexts with BaseMigrator Due to the lengthiness of this task, the migration has been adapted into a BaseMigrator migration, running in the background instead. --- lib/pleroma/application.ex | 3 +- lib/pleroma/data_migration.ex | 1 + .../migrators/context_objects_deletion_migrator.ex | 139 +++++++++++++++++++++ 3 files changed, 142 insertions(+), 1 deletion(-) create mode 100644 lib/pleroma/migrators/context_objects_deletion_migrator.ex (limited to 'lib') diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index d808bc732..c546713ca 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -238,7 +238,8 @@ defmodule Pleroma.Application do defp background_migrators do [ - Pleroma.Migrators.HashtagsTableMigrator + Pleroma.Migrators.HashtagsTableMigrator, + Pleroma.Migrators.ContextObjectsDeletionMigrator ] end diff --git a/lib/pleroma/data_migration.ex b/lib/pleroma/data_migration.ex index 59d891d8d..8451678fc 100644 --- a/lib/pleroma/data_migration.ex +++ b/lib/pleroma/data_migration.ex @@ -42,4 +42,5 @@ defmodule Pleroma.DataMigration do end def populate_hashtags_table, do: get_by_name("populate_hashtags_table") + def delete_context_objects, do: get_by_name("delete_context_objects") end diff --git a/lib/pleroma/migrators/context_objects_deletion_migrator.ex b/lib/pleroma/migrators/context_objects_deletion_migrator.ex new file mode 100644 index 000000000..fb224795a --- /dev/null +++ b/lib/pleroma/migrators/context_objects_deletion_migrator.ex @@ -0,0 +1,139 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Migrators.ContextObjectsDeletionMigrator do + defmodule State do + use Pleroma.Migrators.Support.BaseMigratorState + + @impl Pleroma.Migrators.Support.BaseMigratorState + defdelegate data_migration(), to: Pleroma.DataMigration, as: :delete_context_objects + end + + use Pleroma.Migrators.Support.BaseMigrator + + alias Pleroma.Migrators.Support.BaseMigrator + alias Pleroma.Object + + @doc "This migration removes objects created exclusively for contexts, containing only an `id` field." + + @impl BaseMigrator + def feature_config_path, do: [:features, :delete_context_objects] + + @impl BaseMigrator + def fault_rate_allowance, do: Config.get([:delete_context_objects, :fault_rate_allowance], 0) + + @impl BaseMigrator + def perform do + data_migration_id = data_migration_id() + max_processed_id = get_stat(:max_processed_id, 0) + + Logger.info("Deleting context objects from `objects` (from oid: #{max_processed_id})...") + + query() + |> where([object], object.id > ^max_processed_id) + |> Repo.chunk_stream(100, :batches, timeout: :infinity) + |> Stream.each(fn objects -> + object_ids = Enum.map(objects, & &1.id) + + results = Enum.map(object_ids, &delete_context_object(&1)) + + failed_ids = + results + |> Enum.filter(&(elem(&1, 0) == :error)) + |> Enum.map(&elem(&1, 1)) + + chunk_affected_count = + results + |> Enum.filter(&(elem(&1, 0) == :ok)) + |> length() + + for failed_id <- failed_ids do + _ = + Repo.query( + "INSERT INTO data_migration_failed_ids(data_migration_id, record_id) " <> + "VALUES ($1, $2) ON CONFLICT DO NOTHING;", + [data_migration_id, failed_id] + ) + end + + _ = + Repo.query( + "DELETE FROM data_migration_failed_ids " <> + "WHERE data_migration_id = $1 AND record_id = ANY($2)", + [data_migration_id, object_ids -- failed_ids] + ) + + max_object_id = Enum.at(object_ids, -1) + + put_stat(:max_processed_id, max_object_id) + increment_stat(:iteration_processed_count, length(object_ids)) + increment_stat(:processed_count, length(object_ids)) + increment_stat(:failed_count, length(failed_ids)) + increment_stat(:affected_count, chunk_affected_count) + put_stat(:records_per_second, records_per_second()) + persist_state() + + # A quick and dirty approach to controlling the load this background migration imposes + sleep_interval = Config.get([:delete_context_objects, :sleep_interval_ms], 0) + Process.sleep(sleep_interval) + end) + |> Stream.run() + end + + @impl BaseMigrator + def query do + # Context objects have no activity type, and only one field, `id`. + # Only those context objects are without types. + from( + object in Object, + where: fragment("(?)->'type' IS NULL", object.data), + select: %{ + id: object.id + } + ) + end + + @spec delete_context_object(integer()) :: {:ok | :error, integer()} + defp delete_context_object(id) do + result = + %Object{id: id} + |> Repo.delete() + |> elem(0) + + {result, id} + end + + @impl BaseMigrator + def retry_failed do + data_migration_id = data_migration_id() + + failed_objects_query() + |> Repo.chunk_stream(100, :one) + |> Stream.each(fn object -> + with {res, _} when res != :error <- delete_context_object(object.id) do + _ = + Repo.query( + "DELETE FROM data_migration_failed_ids " <> + "WHERE data_migration_id = $1 AND record_id = $2", + [data_migration_id, object.id] + ) + end + end) + |> Stream.run() + + put_stat(:failed_count, failures_count()) + persist_state() + + force_continue() + end + + defp failed_objects_query do + from(o in Object) + |> join(:inner, [o], dmf in fragment("SELECT * FROM data_migration_failed_ids"), + on: dmf.record_id == o.id + ) + |> where([_o, dmf], dmf.data_migration_id == ^data_migration_id()) + |> order_by([o], asc: o.id) + end +end -- cgit v1.2.3