From 367f0c31c3c15f75aed1d3ba66914e4197c19596 Mon Sep 17 00:00:00 2001
From: Ivan Tashkinov <ivantashkinov@gmail.com>
Date: Thu, 31 Dec 2020 09:36:26 +0300
Subject: [PATCH] [#3213] Added query options support for Repo.chunk_stream/4.
 Used infinite timeout in transfer_hashtags select query.

---
 lib/mix/tasks/pleroma/database.ex | 11 +++++------
 lib/pleroma/repo.ex               |  6 +++---
 2 files changed, 8 insertions(+), 9 deletions(-)

diff --git a/lib/mix/tasks/pleroma/database.ex b/lib/mix/tasks/pleroma/database.ex
index d44bd3478..f903cf75b 100644
--- a/lib/mix/tasks/pleroma/database.ex
+++ b/lib/mix/tasks/pleroma/database.ex
@@ -149,9 +149,9 @@ def run(["transfer_hashtags"]) do
         tag: fragment("(?)->>'tag'", object.data)
       }
     )
-    |> Pleroma.Repo.chunk_stream(100, :batches)
+    |> Repo.chunk_stream(100, :batches, timeout: :infinity)
     |> Stream.each(fn objects ->
-      Logger.info("Processing #{length(objects)} objects...")
+      Logger.info("Processing #{length(objects)} objects starting from id #{hd(objects).id}...")
 
       Enum.map(
         objects,
@@ -165,10 +165,9 @@ def run(["transfer_hashtags"]) do
             with {:ok, hashtag_records} <- Hashtag.get_or_create_by_names(hashtags) do
               for hashtag_record <- hashtag_records do
                 with {:ok, _} <-
-                       Ecto.Adapters.SQL.query(
-                         Repo,
-                         "insert into hashtags_objects(hashtag_id, object_id) values " <>
-                           "(#{hashtag_record.id}, #{object.id});"
+                       Repo.query(
+                         "insert into hashtags_objects(hashtag_id, object_id) values ($1, $2);",
+                         [hashtag_record.id, object.id]
                        ) do
                   :noop
                 else
diff --git a/lib/pleroma/repo.ex b/lib/pleroma/repo.ex
index 4524bd5e2..78711e6ac 100644
--- a/lib/pleroma/repo.ex
+++ b/lib/pleroma/repo.ex
@@ -63,8 +63,8 @@ def get_assoc(resource, association) do
   iex> Pleroma.Repo.chunk_stream(Pleroma.Activity.Queries.by_actor(ap_id), 500, :batches)
   """
   @spec chunk_stream(Ecto.Query.t(), integer(), atom()) :: Enumerable.t()
-  def chunk_stream(query, chunk_size, returns_as \\ :one) do
-    # We don't actually need start and end funcitons of resource streaming,
+  def chunk_stream(query, chunk_size, returns_as \\ :one, query_options \\ []) do
+    # We don't actually need start and end functions of resource streaming,
     # but it seems to be the only way to not fetch records one-by-one and
     # have individual records be the elements of the stream, instead of
     # lists of records
@@ -76,7 +76,7 @@ def chunk_stream(query, chunk_size, returns_as \\ :one) do
           |> order_by(asc: :id)
           |> where([r], r.id > ^last_id)
           |> limit(^chunk_size)
-          |> all()
+          |> all(query_options)
           |> case do
             [] ->
               {:halt, last_id}