X-Git-Url: https://git.squeep.com/?a=blobdiff_plain;f=lib%2Fpleroma%2Fobject.ex;h=740d687a3505fc03ea0cc2161aa44bba4a5f8872;hb=32d1e048178a94017c9d8cca79b28272fa6da9f4;hp=dabb495364277fdd0299a5e185db8947f26fd8bc;hpb=5d34e5a51e2387df55be1f53c37b1a8fd6879dde;p=akkoma diff --git a/lib/pleroma/object.ex b/lib/pleroma/object.ex index dabb49536..20aba4c15 100644 --- a/lib/pleroma/object.ex +++ b/lib/pleroma/object.ex @@ -5,44 +5,48 @@ defmodule Pleroma.Object do use Ecto.Schema - alias Pleroma.Repo - alias Pleroma.Object - alias Pleroma.User alias Pleroma.Activity + alias Pleroma.Object + alias Pleroma.Object.Fetcher alias Pleroma.ObjectTombstone + alias Pleroma.Repo + alias Pleroma.User import Ecto.Query import Ecto.Changeset + require Logger + + @type t() :: %__MODULE__{} + + @derive {Jason.Encoder, only: [:data]} + schema "objects" do field(:data, :map) timestamps() end - def insert_or_get(cng) do - {_, data} = fetch_field(cng, :data) - id = data["id"] || data[:id] - key = "object:#{id}" + def with_joined_activity(query, activity_type \\ "Create", join_type \\ :inner) do + object_position = Map.get(query.aliases, :object, 0) - fetcher = fn _ -> - with nil <- get_by_ap_id(id), - {:ok, object} <- Repo.insert(cng) do - {:commit, object} - else - %Object{} = object -> {:commit, object} - e -> {:ignore, e} - end - end - - with {state, object} when state in [:commit, :ok] <- Cachex.fetch(:object_cache, key, fetcher) do - {:ok, object} - end + join(query, join_type, [{object, object_position}], a in Activity, + on: + fragment( + "COALESCE(?->'object'->>'id', ?->>'object') = (? ->> 'id') AND (?->>'type' = ?) ", + a.data, + a.data, + object.data, + a.data, + ^activity_type + ), + as: :object_activity + ) end def create(data) do Object.change(%Object{}, %{data: data}) - |> insert_or_get() + |> Repo.insert() end def change(struct, params \\ %{}) do @@ -52,15 +56,87 @@ defmodule Pleroma.Object do |> unique_constraint(:ap_id, name: :objects_unique_apid_index) end + def get_by_id(nil), do: nil + def get_by_id(id), do: Repo.get(Object, id) + + def get_by_id_and_maybe_refetch(id, opts \\ []) do + %{updated_at: updated_at} = object = get_by_id(id) + + if opts[:interval] && + NaiveDateTime.diff(NaiveDateTime.utc_now(), updated_at) > opts[:interval] do + case Fetcher.refetch_object(object) do + {:ok, %Object{} = object} -> + object + + e -> + Logger.error("Couldn't refresh #{object.data["id"]}:\n#{inspect(e)}") + object + end + else + object + end + end + def get_by_ap_id(nil), do: nil def get_by_ap_id(ap_id) do Repo.one(from(object in Object, where: fragment("(?)->>'id' = ?", object.data, ^ap_id))) end - def normalize(%{"id" => ap_id}), do: normalize(ap_id) - def normalize(ap_id) when is_binary(ap_id), do: get_cached_by_ap_id(ap_id) - def normalize(_), do: nil + @doc """ + Get a single attachment by it's name and href + """ + @spec get_attachment_by_name_and_href(String.t(), String.t()) :: Object.t() | nil + def get_attachment_by_name_and_href(name, href) do + query = + from(o in Object, + where: fragment("(?)->>'name' = ?", o.data, ^name), + where: fragment("(?)->>'href' = ?", o.data, ^href) + ) + + Repo.one(query) + end + + defp warn_on_no_object_preloaded(ap_id) do + "Object.normalize() called without preloaded object (#{inspect(ap_id)}). Consider preloading the object" + |> Logger.debug() + + Logger.debug("Backtrace: #{inspect(Process.info(:erlang.self(), :current_stacktrace))}") + end + + def normalize(_, fetch_remote \\ true, options \\ []) + + # If we pass an Activity to Object.normalize(), we can try to use the preloaded object. + # Use this whenever possible, especially when walking graphs in an O(N) loop! + def normalize(%Object{} = object, _, _), do: object + def normalize(%Activity{object: %Object{} = object}, _, _), do: object + + # A hack for fake activities + def normalize(%Activity{data: %{"object" => %{"fake" => true} = data}}, _, _) do + %Object{id: "pleroma:fake_object_id", data: data} + end + + # No preloaded object + def normalize(%Activity{data: %{"object" => %{"id" => ap_id}}}, fetch_remote, _) do + warn_on_no_object_preloaded(ap_id) + normalize(ap_id, fetch_remote) + end + + # No preloaded object + def normalize(%Activity{data: %{"object" => ap_id}}, fetch_remote, _) do + warn_on_no_object_preloaded(ap_id) + normalize(ap_id, fetch_remote) + end + + # Old way, try fetching the object through cache. + def normalize(%{"id" => ap_id}, fetch_remote, _), do: normalize(ap_id, fetch_remote) + def normalize(ap_id, false, _) when is_binary(ap_id), do: get_cached_by_ap_id(ap_id) + + def normalize(ap_id, true, options) when is_binary(ap_id) do + Fetcher.fetch_object_from_id!(ap_id, options) + end + + def normalize(_, _, _), do: nil # Owned objects can only be mutated by their owner def authorize_mutation(%Object{data: %{"actor" => actor}}, %User{ap_id: ap_id}), @@ -69,18 +145,18 @@ defmodule Pleroma.Object do # Legacy objects can be mutated by anybody def authorize_mutation(%Object{}, %User{}), do: true + @spec get_cached_by_ap_id(String.t()) :: Object.t() | nil def get_cached_by_ap_id(ap_id) do key = "object:#{ap_id}" - Cachex.fetch!(:object_cache, key, fn _ -> - object = get_by_ap_id(ap_id) - - if object do - {:commit, object} - else - {:ignore, object} - end - end) + with {:ok, nil} <- Cachex.get(:object_cache, key), + object when not is_nil(object) <- get_by_ap_id(ap_id), + {:ok, true} <- Cachex.put(:object_cache, key, object) do + object + else + {:ok, object} -> object + nil -> nil + end end def context_mapping(context) do @@ -106,8 +182,24 @@ defmodule Pleroma.Object do def delete(%Object{data: %{"id" => id}} = object) do with {:ok, _obj} = swap_object_with_tombstone(object), - Repo.delete_all(Activity.by_object_ap_id(id)), - {:ok, true} <- Cachex.del(:object_cache, "object:#{id}") do + deleted_activity = Activity.delete_all_by_object_ap_id(id), + {:ok, true} <- Cachex.del(:object_cache, "object:#{id}"), + {:ok, _} <- Cachex.del(:web_resp_cache, URI.parse(id).path) do + with true <- Pleroma.Config.get([:instance, :cleanup_attachments]) do + {:ok, _} = + Pleroma.Workers.AttachmentsCleanupWorker.enqueue("cleanup_attachments", %{ + "object" => object + }) + end + + {:ok, object, deleted_activity} + end + end + + def prune(%Object{data: %{"id" => id}} = object) do + with {:ok, object} <- Repo.delete(object), + {:ok, true} <- Cachex.del(:object_cache, "object:#{id}"), + {:ok, _} <- Cachex.del(:web_resp_cache, URI.parse(id).path) do {:ok, object} end end @@ -120,8 +212,115 @@ defmodule Pleroma.Object do def update_and_set_cache(changeset) do with {:ok, object} <- Repo.update(changeset) do set_cache(object) + end + end + + def increase_replies_count(ap_id) do + Object + |> where([o], fragment("?->>'id' = ?::text", o.data, ^to_string(ap_id))) + |> update([o], + set: [ + data: + fragment( + """ + safe_jsonb_set(?, '{repliesCount}', + (coalesce((?->>'repliesCount')::int, 0) + 1)::varchar::jsonb, true) + """, + o.data, + o.data + ) + ] + ) + |> Repo.update_all([]) + |> case do + {1, [object]} -> set_cache(object) + _ -> {:error, "Not found"} + end + end + + def decrease_replies_count(ap_id) do + Object + |> where([o], fragment("?->>'id' = ?::text", o.data, ^to_string(ap_id))) + |> update([o], + set: [ + data: + fragment( + """ + safe_jsonb_set(?, '{repliesCount}', + (greatest(0, (?->>'repliesCount')::int - 1))::varchar::jsonb, true) + """, + o.data, + o.data + ) + ] + ) + |> Repo.update_all([]) + |> case do + {1, [object]} -> set_cache(object) + _ -> {:error, "Not found"} + end + end + + def increase_vote_count(ap_id, name) do + with %Object{} = object <- Object.normalize(ap_id), + "Question" <- object.data["type"] do + multiple = Map.has_key?(object.data, "anyOf") + + options = + (object.data["anyOf"] || object.data["oneOf"] || []) + |> Enum.map(fn + %{"name" => ^name} = option -> + Kernel.update_in(option["replies"]["totalItems"], &(&1 + 1)) + + option -> + option + end) + + data = + if multiple do + Map.put(object.data, "anyOf", options) + else + Map.put(object.data, "oneOf", options) + end + + object + |> Object.change(%{data: data}) + |> update_and_set_cache() + else + _ -> :noop + end + end + + @doc "Updates data field of an object" + def update_data(%Object{data: data} = object, attrs \\ %{}) do + object + |> Object.change(%{data: Map.merge(data || %{}, attrs)}) + |> Repo.update() + end + + def local?(%Object{data: %{"id" => id}}) do + String.starts_with?(id, Pleroma.Web.base_url() <> "/") + end + + def replies(object, opts \\ []) do + object = Object.normalize(object) + + query = + Object + |> where( + [o], + fragment("(?)->>'inReplyTo' = ?", o.data, ^object.data["id"]) + ) + |> order_by([o], asc: o.id) + + if opts[:self_only] do + actor = object.data["actor"] + where(query, [o], fragment("(?)->>'actor' = ?", o.data, ^actor)) else - e -> e + query end end + + def self_replies(object, opts \\ []), + do: replies(object, Keyword.put(opts, :self_only, true)) end