Page MenuHomePhorge
Files F85827488

No OneTemporary
Actions

Authored By
Unknown
Size
13 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/paradedb.ex
index 963d573e8..e5d3018d7 100644
--- a/lib/pleroma/search/paradedb.ex
+++ b/lib/pleroma/search/paradedb.ex
@@ -1,213 +1,233 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDB do
@behaviour Pleroma.Search.SearchBackend
require Logger
import Ecto.Query
import Pleroma.Search.DatabaseSearch,
only: [
maybe_fetch: 3,
maybe_restrict_author: 2,
maybe_restrict_blocked: 2,
maybe_restrict_local: 2
]
import Pleroma.Search.Meilisearch, only: [object_to_search_data: 1]
alias Pleroma.Activity
alias Pleroma.Config.Getting, as: Config
@max_limit 40
@default_table "pleroma_search_documents"
defp client_impl do
Config.get([Pleroma.Search.ParadeDB, :client_impl]) ||
Pleroma.Search.ParadeDB.Client.Postgres
end
defp table do
Config.get([Pleroma.Search.ParadeDB, :table], @default_table)
end
@impl true
def create_index do
with {:ok, _} <- client_impl().query("CREATE EXTENSION IF NOT EXISTS pg_search", []),
{:ok, _} <- client_impl().query(create_table_sql(table()), []),
{:ok, _} <- client_impl().query(create_object_id_index_sql(table()), []),
{:ok, _} <- client_impl().query(create_bm25_index_sql(table()), []) do
:ok
else
{:error, error} ->
Logger.error("ParadeDB create_index failed: #{inspect(error)}")
{:error, error}
end
end
@impl true
def drop_index do
with {:ok, _} <- client_impl().query("DROP TABLE IF EXISTS #{table()}", []) do
:ok
else
{:error, error} ->
Logger.error("ParadeDB drop_index failed: #{inspect(error)}")
{:error, error}
end
end
@impl true
def healthcheck_endpoints, do: nil
@impl true
def add_to_index(activity) do
maybe_search_data = object_to_search_data(activity.object)
if activity.data["type"] == "Create" and maybe_search_data do
+ dumped_activity_id = dump_activity_id(activity.id)
actor_ap_id = activity.data["actor"]
published_at = DateTime.from_unix!(maybe_search_data.published)
- sql =
- """
- INSERT INTO #{table()} (id, object_id, object_ap_id, actor_ap_id, content, published_at)
- VALUES ($1, $2, $3, $4, $5, $6)
- ON CONFLICT (id) DO UPDATE SET
- object_id = EXCLUDED.object_id,
- object_ap_id = EXCLUDED.object_ap_id,
- actor_ap_id = EXCLUDED.actor_ap_id,
- content = EXCLUDED.content,
- published_at = EXCLUDED.published_at
- """
-
- params = [
- activity.id,
- maybe_search_data.id,
- maybe_search_data.ap,
- actor_ap_id,
- maybe_search_data.content,
- published_at
- ]
-
- case client_impl().query(sql, params) do
- {:ok, _} ->
- :ok
-
- {:error, error} ->
- Logger.error("ParadeDB add_to_index failed: #{inspect(error)}")
- {:error, error}
+ if is_nil(dumped_activity_id) do
+ Logger.error("ParadeDB add_to_index failed: invalid activity id #{inspect(activity.id)}")
+ :ok
+ else
+ sql =
+ """
+ INSERT INTO #{table()} (id, object_id, object_ap_id, actor_ap_id, content, published_at)
+ VALUES ($1, $2, $3, $4, $5, $6)
+ ON CONFLICT (id) DO UPDATE SET
+ object_id = EXCLUDED.object_id,
+ object_ap_id = EXCLUDED.object_ap_id,
+ actor_ap_id = EXCLUDED.actor_ap_id,
+ content = EXCLUDED.content,
+ published_at = EXCLUDED.published_at
+ """
+
+ params = [
+ dumped_activity_id,
+ maybe_search_data.id,
+ maybe_search_data.ap,
+ actor_ap_id,
+ maybe_search_data.content,
+ published_at
+ ]
+
+ case client_impl().query(sql, params) do
+ {:ok, _} ->
+ :ok
+
+ {:error, error} ->
+ Logger.error("ParadeDB add_to_index failed: #{inspect(error)}")
+ {:error, error}
+ end
end
else
:ok
end
end
@impl true
def remove_from_index(object) do
sql = "DELETE FROM #{table()} WHERE object_id = $1"
case client_impl().query(sql, [object.id]) do
{:ok, _} ->
:ok
{:error, error} ->
Logger.error("ParadeDB remove_from_index failed: #{inspect(error)}")
{:error, error}
end
end
@impl true
def search(user, query, options \\ []) do
limit = options |> Keyword.get(:limit, @max_limit) |> min(@max_limit)
offset = Keyword.get(options, :offset, 0)
author = Keyword.get(options, :author)
case search_ids(query, limit, offset, author) do
{:ok, []} ->
maybe_fetch([], user, query)
{:ok, ids} ->
ids
|> fetch_activities(user, query, author)
{:error, error} ->
Logger.error("ParadeDB search failed: #{inspect(error)}")
maybe_fetch([], user, query)
end
end
defp fetch_activities(ids, user, query, author) do
- dumped_ids = Enum.map(ids, &FlakeId.from_string/1)
+ dumped_ids =
+ ids
+ |> Enum.map(&dump_activity_id/1)
+ |> Enum.reject(&is_nil/1)
from(a in Activity, where: a.id in ^dumped_ids)
|> Activity.with_preloaded_object()
|> Activity.restrict_deactivated_users()
|> maybe_restrict_local(user)
|> maybe_restrict_author(author)
|> maybe_restrict_blocked(user)
|> order_by(
[a],
fragment(
"array_position(?, ?)",
^dumped_ids,
a.id
)
)
|> Pleroma.Repo.all()
|> maybe_fetch(user, query)
end
defp search_ids(query, limit, offset, author) do
base_sql = "SELECT id FROM #{table()} WHERE content &&& $1"
params = [query]
{base_sql, params} =
case author do
%Pleroma.User{ap_id: ap_id} when is_binary(ap_id) ->
{base_sql <> " AND actor_ap_id = $2", params ++ [ap_id]}
_ ->
{base_sql, params}
end
limit_arg = length(params) + 1
offset_arg = length(params) + 2
sql =
base_sql <>
" ORDER BY published_at DESC LIMIT $#{limit_arg} OFFSET $#{offset_arg}"
params = params ++ [limit, offset]
with {:ok, %{rows: rows}} <- client_impl().query(sql, params) do
{:ok, Enum.map(rows, fn [id] -> id end)}
end
end
defp create_table_sql(table) do
"""
CREATE TABLE IF NOT EXISTS #{table} (
- id text PRIMARY KEY,
+ id uuid PRIMARY KEY,
object_id bigint NOT NULL,
object_ap_id text NOT NULL,
actor_ap_id text,
content text NOT NULL,
published_at timestamptz
)
"""
end
defp create_object_id_index_sql(table) do
"CREATE INDEX IF NOT EXISTS #{table}_object_id_idx ON #{table} (object_id)"
end
defp create_bm25_index_sql(table) do
"""
CREATE INDEX IF NOT EXISTS #{table}_bm25_idx ON #{table}
USING bm25 (id, content, (actor_ap_id::pdb.literal), published_at)
WITH (key_field = 'id')
"""
end
+
+ defp dump_activity_id(<<_::binary-size(16)>> = uuid), do: uuid
+
+ defp dump_activity_id(id) when is_binary(id) do
+ case FlakeId.Ecto.CompatType.dump(id) do
+ {:ok, <<_::binary-size(16)>> = uuid} -> uuid
+ _ -> nil
+ end
+ end
+
+ defp dump_activity_id(_), do: nil
end
diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/paradedb_test.exs
index c55e2aa47..3338c78ef 100644
--- a/test/pleroma/search/paradedb_test.exs
+++ b/test/pleroma/search/paradedb_test.exs
@@ -1,168 +1,176 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDBTest do
use Pleroma.DataCase, async: true
use Oban.Testing, repo: Pleroma.Repo
import Mox
import Pleroma.Factory
alias Pleroma.Search.ParadeDB
alias Pleroma.Search.ParadeDB.ClientMock
alias Pleroma.UnstubbedConfigMock, as: Config
alias Pleroma.Web.CommonAPI
alias Pleroma.Workers.SearchIndexingWorker
describe "ParadeDB" do
test "indexes a public post on creation" do
user = insert(:user)
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: "public"
})
args = %{"op" => "add_to_index", "activity" => activity.id}
assert_enqueued(worker: SearchIndexingWorker, args: args)
ClientMock
|> expect(:query, fn sql, params ->
assert sql =~ "INSERT INTO pleroma_search_documents"
[activity_id, object_id, object_ap_id, actor_ap_id, content, published_at] = params
- assert activity_id == activity.id
+ {:ok, dumped_activity_id} = FlakeId.Ecto.CompatType.dump(activity.id)
+
+ assert activity_id == dumped_activity_id
assert is_integer(object_id)
assert is_binary(object_ap_id)
assert actor_ap_id == activity.data["actor"]
assert content == "guys i just don&#39;t wanna leave the swamp"
assert %DateTime{} = published_at
send(self(), :inserted)
{:ok, %{}}
end)
Config
|> expect(:get, 3, fn
[Pleroma.Search, :module], nil ->
ParadeDB
[Pleroma.Search.ParadeDB, :client_impl], nil ->
ClientMock
[Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" ->
"pleroma_search_documents"
end)
assert :ok = perform_job(SearchIndexingWorker, args)
assert_received(:inserted)
end
test "doesn't index posts that are not public" do
user = insert(:user)
Enum.each(["private", "direct"], fn visibility ->
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: visibility
})
args = %{"op" => "add_to_index", "activity" => activity.id}
assert_enqueued(worker: SearchIndexingWorker, args: args)
Config
|> expect(:get, fn
[Pleroma.Search, :module], nil ->
ParadeDB
end)
assert :ok = perform_job(SearchIndexingWorker, args)
end)
end
test "deletes posts from index when deleted locally" do
user = insert(:user)
ClientMock
|> expect(:query, 2, fn sql, params ->
cond do
String.contains?(sql, "INSERT INTO pleroma_search_documents") ->
send(self(), :inserted)
{:ok, %{}}
String.contains?(sql, "DELETE FROM pleroma_search_documents") ->
assert [object_id] = params
assert is_integer(object_id)
send(self(), :deleted)
{:ok, %{}}
true ->
flunk("Unexpected ParadeDB SQL: #{inspect(sql)}")
end
end)
Config
|> expect(:get, 6, fn
[Pleroma.Search, :module], nil ->
ParadeDB
[Pleroma.Search.ParadeDB, :client_impl], nil ->
ClientMock
[Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" ->
"pleroma_search_documents"
end)
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: "public"
})
args = %{"op" => "add_to_index", "activity" => activity.id}
assert_enqueued(worker: SearchIndexingWorker, args: args)
assert :ok = perform_job(SearchIndexingWorker, args)
assert_received(:inserted)
{:ok, _} = CommonAPI.delete(activity.id, user)
delete_args = %{"op" => "remove_from_index", "object" => activity.object.id}
assert_enqueued(worker: SearchIndexingWorker, args: delete_args)
assert :ok = perform_job(SearchIndexingWorker, delete_args)
assert_received(:deleted)
end
test "search returns activities in backend order" do
user = insert(:user)
{:ok, activity1} = CommonAPI.post(user, %{status: "first swamp", visibility: "public"})
{:ok, activity2} = CommonAPI.post(user, %{status: "second swamp", visibility: "public"})
+ activity1_id = activity1.id
+ activity2_id = activity2.id
+
+ {:ok, dumped_activity1_id} = FlakeId.Ecto.CompatType.dump(activity1_id)
+ {:ok, dumped_activity2_id} = FlakeId.Ecto.CompatType.dump(activity2_id)
+
ClientMock
|> expect(:query, fn sql, params ->
assert sql =~ "SELECT id FROM pleroma_search_documents"
assert ["swamp", _limit, _offset] = params
- {:ok, %{rows: [[activity2.id], [activity1.id]]}}
+ {:ok, %{rows: [[dumped_activity2_id], [dumped_activity1_id]]}}
end)
Config
|> expect(:get, 2, fn
[Pleroma.Search.ParadeDB, :client_impl], nil ->
ClientMock
[Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" ->
"pleroma_search_documents"
end)
- assert [%{id: ^activity2.id}, %{id: ^activity1.id}] =
+ assert [%{id: ^activity2_id}, %{id: ^activity1_id}] =
ParadeDB.search(nil, "swamp", limit: 40, offset: 0)
end
end
end

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 30, 8:01 PM (14 h, 43 m)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1800496
Default Alt Text
(13 KB)

Event Timeline