Merge branch 'from/upstream-develop/tusooa/assoc-object-id' into 'develop'

Add function to calculate associated object id

Closes #2307

See merge request pleroma/pleroma!3692

lib/mix/tasks/pleroma/database.ex

@@ -154,9 +154,8 @@ def run(["ensure_expiration"]) do
     |> join(:inner, [a], o in Object,
       on:
         fragment(
-          "(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
+          "(?->>'id') = associated_object_id((?))",
           o.data,
-          a.data,
           a.data
         )
     )

lib/pleroma/activity.ex

@@ -53,7 +53,7 @@ defmodule Pleroma.Activity do
     #
     # ```
     # |> join(:inner, [activity], o in Object,
-    #      on: fragment("(?->>'id') = COALESCE((?)->'object'->> 'id', (?)->>'object')",
+    #      on: fragment("(?->>'id') = associated_object_id((?))",
     #        o.data, activity.data, activity.data))
     # |> preload([activity, object], [object: object])
     # ```
@@ -69,9 +69,8 @@ def with_joined_object(query, join_type \\ :inner) do
     join(query, join_type, [activity], o in Object,
       on:
         fragment(
-          "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
+          "(?->>'id') = associated_object_id(?)",
           o.data,
-          activity.data,
           activity.data
         ),
       as: :object

lib/pleroma/activity/queries.ex

@@ -52,8 +52,7 @@ def by_object_id(query, object_ids) when is_list(object_ids) do
       activity in query,
       where:
         fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ANY(?)",
+          "associated_object_id((?)) = ANY(?)",
-          activity.data,
           activity.data,
           ^object_ids
         )
@@ -64,8 +63,7 @@ def by_object_id(query, object_id) when is_binary(object_id) do
     from(activity in query,
       where:
         fragment(
-          "coalesce((?)->'object'->>'id', (?)->>'object') = ?",
+          "associated_object_id((?)) = ?",
-          activity.data,
           activity.data,
           ^object_id
         )

lib/pleroma/migrators/hashtags_table_migrator.ex

@@ -183,7 +183,7 @@ def delete_non_create_activities_hashtags do
     DELETE FROM hashtags_objects WHERE object_id IN
       (SELECT DISTINCT objects.id FROM objects
         JOIN hashtags_objects ON hashtags_objects.object_id = objects.id LEFT JOIN activities
-          ON COALESCE(activities.data->'object'->>'id', activities.data->>'object') =
+          ON associated_object_id(activities) =
             (objects.data->>'id')
           AND activities.data->>'type' = 'Create'
         WHERE activities.id IS NULL);

lib/pleroma/notification.ex

@@ -117,9 +117,8 @@ def for_user_query(user, opts \\ %{}) do
     |> join(:left, [n, a], object in Object,
       on:
         fragment(
-          "(?->>'id') = COALESCE(?->'object'->>'id', ?->>'object')",
+          "(?->>'id') = associated_object_id(?)",
           object.data,
-          a.data,
           a.data
         )
     )
@@ -193,13 +192,11 @@ defp exclude_visibility(query, %{exclude_visibilities: visibility})
       |> join(:left, [n, a], mutated_activity in Pleroma.Activity,
         on:
           fragment(
-            "COALESCE((?->'object')->>'id', ?->>'object')",
+            "associated_object_id(?)",
-            a.data,
             a.data
           ) ==
             fragment(
-              "COALESCE((?->'object')->>'id', ?->>'object')",
+              "associated_object_id(?)",
-              mutated_activity.data,
               mutated_activity.data
             ) and
             fragment("(?->>'type' = 'Like' or ?->>'type' = 'Announce')", a.data, a.data) and

lib/pleroma/object.ex

@@ -40,8 +40,7 @@ def with_joined_activity(query, activity_type \\ "Create", join_type \\ :inner)
     join(query, join_type, [{object, object_position}], a in Activity,
       on:
         fragment(
-          "COALESCE(?->'object'->>'id', ?->>'object') = (? ->> 'id') AND (?->>'type' = ?) ",
+          "associated_object_id(?) = (? ->> 'id') AND (?->>'type' = ?) ",
-          a.data,
           a.data,
           object.data,
           a.data,

lib/pleroma/web/activity_pub/activity_pub.ex

@@ -1159,8 +1159,7 @@ defp restrict_pinned(query, %{pinned: true, pinned_object_ids: ids}) do
       [activity, object: o] in query,
       where:
         fragment(
-          "(?)->>'type' = 'Create' and coalesce((?)->'object'->>'id', (?)->>'object') = any (?)",
+          "(?)->>'type' = 'Create' and associated_object_id((?)) = any (?)",
-          activity.data,
           activity.data,
           activity.data,
           ^ids

priv/repo/migrations/20220711182322_add_associated_object_id_function.exs

@@ -0,0 +1,37 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Repo.Migrations.AddAssociatedObjectIdFunction do
+  use Ecto.Migration
+
+  def up do
+    statement = """
+    CREATE OR REPLACE FUNCTION associated_object_id(data jsonb) RETURNS varchar AS $$
+    DECLARE
+      object_data jsonb;
+    BEGIN
+      IF jsonb_typeof(data->'object') = 'array' THEN
+        object_data := data->'object'->0;
+      ELSE
+        object_data := data->'object';
+      END IF;
+
+      IF jsonb_typeof(object_data->'id') = 'string' THEN
+        RETURN object_data->>'id';
+      ELSIF jsonb_typeof(object_data) = 'string' THEN
+        RETURN object_data#>>'{}';
+      ELSE
+        RETURN NULL;
+      END IF;
+    END;
+    $$ LANGUAGE plpgsql IMMUTABLE;
+    """
+
+    execute(statement)
+  end
+
+  def down do
+    execute("DROP FUNCTION IF EXISTS associated_object_id(data jsonb)")
+  end
+end

priv/repo/migrations/20220711192750_switch_to_associated_object_id_index.exs

@@ -0,0 +1,37 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Repo.Migrations.SwitchToAssociatedObjectIdIndex do
+  use Ecto.Migration
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def up do
+    drop_if_exists(
+      index(:activities, ["(coalesce(data->'object'->>'id', data->>'object'))"],
+        name: :activities_create_objects_index
+      )
+    )
+
+    create(
+      index(:activities, ["associated_object_id(data)"],
+        name: :activities_create_objects_index,
+        concurrently: true
+      )
+    )
+  end
+
+  def down do
+    drop_if_exists(
+      index(:activities, ["associated_object_id(data)"], name: :activities_create_objects_index)
+    )
+
+    create(
+      index(:activities, ["(coalesce(data->'object'->>'id', data->>'object'))"],
+        name: :activities_create_objects_index,
+        concurrently: true
+      )
+    )
+  end
+end

priv/repo/migrations/20220821004840_change_thread_visibility_to_use_new_object_id_index.exs

@@ -0,0 +1,156 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Repo.Migrations.ChangeThreadVisibilityToUseNewObjectIdIndex do
+  use Ecto.Migration
+
+  def up do
+    execute(update_thread_visibility())
+  end
+
+  def down do
+    execute(restore_thread_visibility())
+  end
+
+  def update_thread_visibility do
+    """
+    CREATE OR REPLACE FUNCTION thread_visibility(actor varchar, activity_id varchar, local_public varchar default '') RETURNS boolean AS $$
+    DECLARE
+      public varchar := 'https://www.w3.org/ns/activitystreams#Public';
+      child objects%ROWTYPE;
+      activity activities%ROWTYPE;
+      author_fa varchar;
+      valid_recipients varchar[];
+      actor_user_following varchar[];
+    BEGIN
+      --- Fetch actor following
+      SELECT array_agg(following.follower_address) INTO actor_user_following FROM following_relationships
+      JOIN users ON users.id = following_relationships.follower_id
+      JOIN users AS following ON following.id = following_relationships.following_id
+      WHERE users.ap_id = actor;
+
+      --- Fetch our initial activity.
+      SELECT * INTO activity FROM activities WHERE activities.data->>'id' = activity_id;
+
+      LOOP
+        --- Ensure that we have an activity before continuing.
+        --- If we don't, the thread is not satisfiable.
+        IF activity IS NULL THEN
+          RETURN false;
+        END IF;
+
+        --- We only care about Create activities.
+        IF activity.data->>'type' != 'Create' THEN
+          RETURN true;
+        END IF;
+
+        --- Normalize the child object into child.
+        SELECT * INTO child FROM objects
+        INNER JOIN activities ON associated_object_id(activities.data) = objects.data->>'id'
+        WHERE associated_object_id(activity.data) = objects.data->>'id';
+
+        --- Fetch the author's AS2 following collection.
+        SELECT COALESCE(users.follower_address, '') INTO author_fa FROM users WHERE users.ap_id = activity.actor;
+
+        --- Prepare valid recipients array.
+        valid_recipients := ARRAY[actor, public];
+        --- If we specified local public, add it.
+        IF local_public <> '' THEN
+          valid_recipients := valid_recipients || local_public;
+        END IF;
+        IF ARRAY[author_fa] && actor_user_following THEN
+          valid_recipients := valid_recipients || author_fa;
+        END IF;
+
+        --- Check visibility.
+        IF NOT valid_recipients && activity.recipients THEN
+          --- activity not visible, break out of the loop
+          RETURN false;
+        END IF;
+
+        --- If there's a parent, load it and do this all over again.
+        IF (child.data->'inReplyTo' IS NOT NULL) AND (child.data->'inReplyTo' != 'null'::jsonb) THEN
+          SELECT * INTO activity FROM activities
+          INNER JOIN objects ON associated_object_id(activities.data) = objects.data->>'id'
+          WHERE child.data->>'inReplyTo' = objects.data->>'id';
+        ELSE
+          RETURN true;
+        END IF;
+      END LOOP;
+    END;
+    $$ LANGUAGE plpgsql IMMUTABLE;
+    """
+  end
+
+  # priv/repo/migrations/20220509180452_change_thread_visibility_to_be_local_only_aware.exs
+  def restore_thread_visibility do
+    """
+    CREATE OR REPLACE FUNCTION thread_visibility(actor varchar, activity_id varchar, local_public varchar default '') RETURNS boolean AS $$
+    DECLARE
+      public varchar := 'https://www.w3.org/ns/activitystreams#Public';
+      child objects%ROWTYPE;
+      activity activities%ROWTYPE;
+      author_fa varchar;
+      valid_recipients varchar[];
+      actor_user_following varchar[];
+    BEGIN
+      --- Fetch actor following
+      SELECT array_agg(following.follower_address) INTO actor_user_following FROM following_relationships
+      JOIN users ON users.id = following_relationships.follower_id
+      JOIN users AS following ON following.id = following_relationships.following_id
+      WHERE users.ap_id = actor;
+
+      --- Fetch our initial activity.
+      SELECT * INTO activity FROM activities WHERE activities.data->>'id' = activity_id;
+
+      LOOP
+        --- Ensure that we have an activity before continuing.
+        --- If we don't, the thread is not satisfiable.
+        IF activity IS NULL THEN
+          RETURN false;
+        END IF;
+
+        --- We only care about Create activities.
+        IF activity.data->>'type' != 'Create' THEN
+          RETURN true;
+        END IF;
+
+        --- Normalize the child object into child.
+        SELECT * INTO child FROM objects
+        INNER JOIN activities ON COALESCE(activities.data->'object'->>'id', activities.data->>'object') = objects.data->>'id'
+        WHERE COALESCE(activity.data->'object'->>'id', activity.data->>'object') = objects.data->>'id';
+
+        --- Fetch the author's AS2 following collection.
+        SELECT COALESCE(users.follower_address, '') INTO author_fa FROM users WHERE users.ap_id = activity.actor;
+
+        --- Prepare valid recipients array.
+        valid_recipients := ARRAY[actor, public];
+        --- If we specified local public, add it.
+        IF local_public <> '' THEN
+          valid_recipients := valid_recipients || local_public;
+        END IF;
+        IF ARRAY[author_fa] && actor_user_following THEN
+          valid_recipients := valid_recipients || author_fa;
+        END IF;
+
+        --- Check visibility.
+        IF NOT valid_recipients && activity.recipients THEN
+          --- activity not visible, break out of the loop
+          RETURN false;
+        END IF;
+
+        --- If there's a parent, load it and do this all over again.
+        IF (child.data->'inReplyTo' IS NOT NULL) AND (child.data->'inReplyTo' != 'null'::jsonb) THEN
+          SELECT * INTO activity FROM activities
+          INNER JOIN objects ON COALESCE(activities.data->'object'->>'id', activities.data->>'object') = objects.data->>'id'
+          WHERE child.data->>'inReplyTo' = objects.data->>'id';
+        ELSE
+          RETURN true;
+        END IF;
+      END LOOP;
+    END;
+    $$ LANGUAGE plpgsql IMMUTABLE;
+    """
+  end
+end

test/pleroma/activity_test.exs

@@ -278,4 +278,78 @@ test "add_by_params_query/3" do
 
     assert Repo.aggregate(Activity, :count, :id) == 2
   end
+
+  describe "associated_object_id() sql function" do
+    test "with json object" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{"object": {"id":"foobar"}}'::jsonb);
+          """
+        )
+
+      assert object_id == "foobar"
+    end
+
+    test "with string object" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{"object": "foobar"}'::jsonb);
+          """
+        )
+
+      assert object_id == "foobar"
+    end
+
+    test "with array object" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{"object": ["foobar", {}]}'::jsonb);
+          """
+        )
+
+      assert object_id == "foobar"
+    end
+
+    test "invalid" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{"object": {}}'::jsonb);
+          """
+        )
+
+      assert is_nil(object_id)
+    end
+
+    test "invalid object id" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{"object": {"id": 123}}'::jsonb);
+          """
+        )
+
+      assert is_nil(object_id)
+    end
+
+    test "no object field" do
+      %{rows: [[object_id]]} =
+        Ecto.Adapters.SQL.query!(
+          Pleroma.Repo,
+          """
+          select associated_object_id('{}'::jsonb);
+          """
+        )
+
+      assert is_nil(object_id)
+    end
+  end
 end