fix import from live

config/config.exs

@@ -874,7 +874,7 @@ config :pleroma, Pleroma.Search.Elasticsearch.Cluster,
       settings: "priv/es-mappings/activity.json",
       store: Pleroma.Search.Elasticsearch.Store,
       sources: [Pleroma.Activity],
-      bulk_page_size: 5000,
+      bulk_page_size: 1000,
       bulk_wait_interval: 15_000
     }
   }

docs/configuration/search.md

@@ -154,10 +154,10 @@ To start the initial indexing, run the `build` command:
 
 === "OTP"
 ```sh
-./bin/pleroma_ctl search.elasticsearch index activities --cluster Pleroma.Search.Elasticsearch.Cluster
+./bin/pleroma_ctl search import activities
 ```
 
 === "From Source"
 ```sh
-mix elasticsearch.build activities --cluster Pleroma.Search.Elasticsearch.Cluster
-```
+mix pleroma.search import activities
+```

lib/mix/pleroma.ex

@@ -57,7 +57,8 @@ defmodule Mix.Pleroma do
         {Majic.Pool,
          [name: Pleroma.MajicPool, pool_size: Pleroma.Config.get([:majic_pool, :size], 2)]}
       ] ++
-        http_children(adapter)
+        http_children(adapter) ++
+	elasticsearch_children()
 
     cachex_children = Enum.map(@cachex_children, &Pleroma.Application.build_cachex(&1, []))
 
@@ -136,4 +137,14 @@ defmodule Mix.Pleroma do
   end
 
   defp http_children(_), do: []
+
+  def elasticsearch_children do
+    config = Pleroma.Config.get([Pleroma.Search, :module])
+
+    if config == Pleroma.Search.Elasticsearch do
+      [Pleroma.Search.Elasticsearch.Cluster]
+    else
+      []
+    end
+  end
 end

lib/mix/tasks/pleroma/search.ex

@@ -0,0 +1,52 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2021 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Mix.Tasks.Pleroma.Search do
+  use Mix.Task
+  import Mix.Pleroma
+  import Ecto.Query
+  alias Pleroma.Activity
+  alias Pleroma.Pagination
+  alias Pleroma.User
+  alias Pleroma.Hashtag
+
+  @shortdoc "Manages elasticsearch"
+
+  def run(["import", "activities" | _rest]) do
+    start_pleroma()
+
+    Elasticsearch.Index.Bulk.upload(Pleroma.Search.Elasticsearch.Cluster, 
+	"activities",
+	Pleroma.Config.get([Pleroma.Search.Elasticsearch.Cluster, :indexes, :activities]))
+    #from(a in Activity, where: not ilike(a.actor, "%/relay"))
+    #|> where([a], fragment("(? ->> 'type'::text) = 'Create'", a.data))
+    #|> Activity.with_preloaded_object()
+    #|> Activity.with_preloaded_user_actor()
+    #|> get_all(:activities)
+  end
+
+  defp get_all(query, index, max_id \\ nil) do
+    params = %{limit: 1000}
+
+    params =
+      if max_id == nil do
+        params
+      else
+        Map.put(params, :max_id, max_id)
+      end
+
+    res =
+      query
+      |> Pagination.fetch_paginated(params)
+
+    if res == [] do
+      :ok
+    else
+      res
+      |> Enum.map(fn x -> Pleroma.Search.Elasticsearch.add_to_index(x) end)
+
+      get_all(query, index, List.last(res).id)
+    end
+  end
+end