git.squeep.com Git - akkoma/blob - lib/pleroma/elasticsearch/store.ex

   1 defmodule Pleroma.Elasticsearch do
   2   alias Pleroma.Activity
   3   alias Pleroma.User
   4   alias Pleroma.Object
   5   alias Pleroma.Elasticsearch.DocumentMappings
   6   alias Pleroma.Config
   7   require Logger
   8
   9   defp url do
  10     Config.get([:elasticsearch, :url])
  11   end
  12
  13   defp enabled? do
  14     Config.get([:search, :provider]) == Pleroma.Search.Elasticsearch
  15   end
  16
  17   def delete_by_id(:activity, id) do
  18     if enabled?() do
  19       Elastix.Document.delete(url(), "activities", "activity", id)
  20     end
  21   end
  22
  23   def put_by_id(:activity, id) do
  24     id
  25     |> Activity.get_by_id_with_object()
  26     |> maybe_put_into_elasticsearch()
  27   end
  28
  29   def maybe_put_into_elasticsearch({:ok, item}) do
  30     maybe_put_into_elasticsearch(item)
  31   end
  32
  33   def maybe_put_into_elasticsearch(
  34         %{data: %{"type" => "Create"}, object: %{data: %{"type" => "Note"}}} = activity
  35       ) do
  36     if enabled?() do
  37       actor = Pleroma.Activity.user_actor(activity)
  38
  39       activity
  40       |> Map.put(:user_actor, actor)
  41       |> put()
  42     end
  43   end
  44
  45   def maybe_put_into_elasticsearch(%User{actor_type: "Person"} = user) do
  46     if enabled?() do
  47       put(user)
  48     end
  49   end
  50
  51   def maybe_put_into_elasticsearch(_) do
  52     {:ok, :skipped}
  53   end
  54
  55   def maybe_bulk_post(data, type) do
  56     if enabled?() do
  57       bulk_post(data, type)
  58     end
  59   end
  60
  61   def put(%Activity{} = activity) do
  62     {:ok, _} =
  63       Elastix.Document.index(
  64         url(),
  65         "activities",
  66         "activity",
  67         DocumentMappings.Activity.id(activity),
  68         DocumentMappings.Activity.encode(activity)
  69       )
  70
  71     activity
  72     |> Map.get(:object)
  73     |> Object.hashtags()
  74     |> Enum.map(fn x -> %{id: x, name: x, timestamp: DateTime.to_iso8601(DateTime.utc_now())} end)
  75     |> bulk_post(:hashtags)
  76   end
  77
  78   def put(%User{} = user) do
  79     {:ok, _} =
  80       Elastix.Document.index(
  81         url(),
  82         "users",
  83         "user",
  84         DocumentMappings.User.id(user),
  85         DocumentMappings.User.encode(user)
  86       )
  87   end
  88
  89   def bulk_post(data, :activities) do
  90     d =
  91       data
  92       |> Enum.filter(fn x ->
  93         t =
  94           x.object
  95           |> Map.get(:data, %{})
  96           |> Map.get("type", "")
  97
  98         t == "Note"
  99       end)
 100       |> Enum.map(fn d ->
 101         [
 102           %{index: %{_id: DocumentMappings.Activity.id(d)}},
 103           DocumentMappings.Activity.encode(d)
 104         ]
 105       end)
 106       |> List.flatten()
 107
 108     {:ok, %{body: %{"errors" => false}}} =
 109       Elastix.Bulk.post(
 110         url(),
 111         d,
 112         index: "activities",
 113         type: "activity"
 114       )
 115   end
 116
 117   def bulk_post(data, :users) do
 118     d =
 119       data
 120       |> Enum.filter(fn x -> x.actor_type == "Person" end)
 121       |> Enum.map(fn d ->
 122         [
 123           %{index: %{_id: DocumentMappings.User.id(d)}},
 124           DocumentMappings.User.encode(d)
 125         ]
 126       end)
 127       |> List.flatten()
 128
 129     Elastix.Bulk.post(
 130       url(),
 131       d,
 132       index: "users",
 133       type: "user"
 134     )
 135   end
 136
 137   def bulk_post(data, :hashtags) when is_list(data) do
 138     d =
 139       data
 140       |> Enum.map(fn d ->
 141         [
 142           %{index: %{_id: DocumentMappings.Hashtag.id(d)}},
 143           DocumentMappings.Hashtag.encode(d)
 144         ]
 145       end)
 146       |> List.flatten()
 147
 148     Elastix.Bulk.post(
 149       url(),
 150       d,
 151       index: "hashtags",
 152       type: "hashtag"
 153     )
 154   end
 155
 156   def bulk_post(_, :hashtags), do: {:ok, nil}
 157
 158   def search(_, _, _, :skip), do: []
 159
 160   def search(:raw, index, type, q) do
 161     with {:ok, raw_results} <- Elastix.Search.search(url(), index, [type], q) do
 162       results =
 163         raw_results
 164         |> Map.get(:body, %{})
 165         |> Map.get("hits", %{})
 166         |> Map.get("hits", [])
 167
 168       {:ok, results}
 169     else
 170       {:error, e} ->
 171         Logger.error(e)
 172         {:error, e}
 173     end
 174   end
 175
 176   def search(:activities, q) do
 177     with {:ok, results} <- search(:raw, "activities", "activity", q) do
 178       results
 179       |> Enum.map(fn result -> result["_id"] end)
 180       |> Pleroma.Activity.all_by_ids_with_object()
 181       |> Enum.sort(&(&1.inserted_at >= &2.inserted_at))
 182     else
 183       e ->
 184         Logger.error(e)
 185         []
 186     end
 187   end
 188
 189   def search(:users, q) do
 190     with {:ok, results} <- search(:raw, "users", "user", q) do
 191       results
 192       |> Enum.map(fn result -> result["_id"] end)
 193       |> Pleroma.User.get_all_by_ids()
 194     else
 195       e ->
 196         Logger.error(e)
 197         []
 198     end
 199   end
 200
 201   def search(:hashtags, q) do
 202     with {:ok, results} <- search(:raw, "hashtags", "hashtag", q) do
 203       results
 204       |> Enum.map(fn result -> result["_source"]["hashtag"] end)
 205     else
 206       e ->
 207         Logger.error(e)
 208         []
 209     end
 210   end
 211 end