diff --git a/changelog.d/paradedb-search.add b/changelog.d/paradedb-search.add new file mode 100644 index 000000000..00a6e2361 --- /dev/null +++ b/changelog.d/paradedb-search.add @@ -0,0 +1 @@ +Add ParadeDB (pg_search) as an optional external search backend. diff --git a/config/config.exs b/config/config.exs index 683805fe3..7e374b22b 100644 --- a/config/config.exs +++ b/config/config.exs @@ -938,6 +938,14 @@ config :pleroma, Pleroma.Search.Meilisearch, private_key: nil, initial_indexing_chunk_size: 100_000 +config :pleroma, Pleroma.Search.ParadeDB, + url: nil, + table: "pleroma_search_documents" + +config :pleroma, Pleroma.Search.ParadeDB.Repo, + pool_size: 10, + prepare: :unnamed + config :pleroma, Pleroma.Application, background_migrators: true, internal_fetch: true, diff --git a/config/description.exs b/config/description.exs index c388d17c3..de454dfb2 100644 --- a/config/description.exs +++ b/config/description.exs @@ -3503,7 +3503,12 @@ config :pleroma, :config_description, [ key: :module, type: :keyword, description: "Selected search module.", - suggestion: [Pleroma.Search.DatabaseSearch, Pleroma.Search.Meilisearch] + suggestion: [ + Pleroma.Search.DatabaseSearch, + Pleroma.Search.Meilisearch, + Pleroma.Search.QdrantSearch, + Pleroma.Search.ParadeDB + ] } ] }, @@ -3536,6 +3541,28 @@ config :pleroma, :config_description, [ } ] }, + %{ + group: :pleroma, + key: Pleroma.Search.ParadeDB, + type: :group, + description: "ParadeDB settings.", + children: [ + %{ + key: :url, + type: :string, + description: + "ParadeDB database URL (separate Postgres instance). " <> + "Can also be set via `PARADEDB_DATABASE_URL`.", + suggestion: ["postgres://postgres:postgres@127.0.0.1:5432/paradedb"] + }, + %{ + key: :table, + type: :string, + description: "Table name used for search documents.", + suggestion: ["pleroma_search_documents"] + } + ] + }, %{ group: :pleroma, key: Pleroma.Language.LanguageDetector, diff --git a/docs/configuration/search.md b/docs/configuration/search.md index d34f84d4f..780904d10 100644 --- a/docs/configuration/search.md +++ b/docs/configuration/search.md @@ -145,3 +145,37 @@ This will clear **all** the posts from the search index. Note, that deleted post there is no need to actually clear the whole index, unless you want **all** of it gone. That said, the index does not hold any information that cannot be re-created from the database, it should also generally be a lot smaller than the size of your database. Still, the size depends on the amount of text in posts. + +## ParadeDB + +[ParadeDB](https://www.paradedb.com/) is a Postgres extension that provides BM25 full-text search. In Pleroma, it can be used as an +external search backend by pointing Pleroma at a separate Postgres instance with the `pg_search` extension installed. + +Pleroma will maintain a small `pleroma_search_documents` table in that database (via the existing search indexing queue) and run search +queries against it. + +To use it, set the search module to `Pleroma.Search.ParadeDB` and configure the ParadeDB database URL: + +> config :pleroma, Pleroma.Search, module: Pleroma.Search.ParadeDB +> +> config :pleroma, Pleroma.Search.ParadeDB, +> url: System.get_env("PARADEDB_DATABASE_URL") || "postgres://postgres:postgres@127.0.0.1:5432/paradedb", +> table: "pleroma_search_documents" + +Then, create the table and BM25 index once: + +=== "From Source" + ```sh + mix pleroma.search.indexer create_index + ``` + +After that, run an initial indexing pass to backfill existing posts: + +=== "From Source" + ```sh + mix pleroma.search.indexer index + ``` + +Note: Like the other external search backends, only public/unlisted Notes are indexed. + +Indexing runs via the Oban `search_indexing` queue, so ensure that queue is enabled (not paused) in your Oban configuration. diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex index 8e1c5de0d..8de71b816 100644 --- a/lib/pleroma/application.ex +++ b/lib/pleroma/application.ex @@ -80,12 +80,15 @@ defmodule Pleroma.Application do [ Pleroma.PromEx, Pleroma.LDAP, - Pleroma.Repo, - Config.TransferTask, - Pleroma.Emoji, - Pleroma.Web.Plugs.RateLimiter.Supervisor, - {Task.Supervisor, name: Pleroma.TaskSupervisor} + Pleroma.Repo ] ++ + search_children() ++ + [ + Config.TransferTask, + Pleroma.Emoji, + Pleroma.Web.Plugs.RateLimiter.Supervisor, + {Task.Supervisor, name: Pleroma.TaskSupervisor} + ] ++ cachex_children() ++ http_children(adapter) ++ [ @@ -159,6 +162,16 @@ defmodule Pleroma.Application do ] end + defp search_children do + case Config.get([Pleroma.Search, :module]) do + Pleroma.Search.ParadeDB -> + [Pleroma.Search.ParadeDB.Repo] + + _ -> + [] + end + end + defp emoji_packs_expiration, do: expiration(default: :timer.seconds(5 * 60), interval: :timer.seconds(60)) diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/paradedb.ex new file mode 100644 index 000000000..963d573e8 --- /dev/null +++ b/lib/pleroma/search/paradedb.ex @@ -0,0 +1,213 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 + 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} + 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) + + 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, + 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 +end diff --git a/lib/pleroma/search/paradedb/client.ex b/lib/pleroma/search/paradedb/client.ex new file mode 100644 index 000000000..d6dce0dd4 --- /dev/null +++ b/lib/pleroma/search/paradedb/client.ex @@ -0,0 +1,24 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Search.ParadeDB.Client do + @moduledoc false + + @type sql :: String.t() + @type params :: list() + @type result :: %{optional(:rows) => list(list())} + + @callback query(sql(), params()) :: {:ok, result()} | {:error, any()} +end + +defmodule Pleroma.Search.ParadeDB.Client.Postgres do + @moduledoc false + + @behaviour Pleroma.Search.ParadeDB.Client + + @impl true + def query(sql, params \\ []) do + Ecto.Adapters.SQL.query(Pleroma.Search.ParadeDB.Repo, sql, params) + end +end diff --git a/lib/pleroma/search/paradedb/repo.ex b/lib/pleroma/search/paradedb/repo.ex new file mode 100644 index 000000000..684271847 --- /dev/null +++ b/lib/pleroma/search/paradedb/repo.ex @@ -0,0 +1,24 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Search.ParadeDB.Repo do + use Ecto.Repo, + otp_app: :pleroma, + adapter: Ecto.Adapters.Postgres, + migration_timestamps: [type: :naive_datetime_usec] + + @doc """ + Loads the ParadeDB database URL from `PARADEDB_DATABASE_URL`. + + If the env var is not set, falls back to `config :pleroma, Pleroma.Search.ParadeDB, url: ...`. + """ + def init(_, opts) do + config = Application.get_env(:pleroma, Pleroma.Search.ParadeDB, []) + + url = System.get_env("PARADEDB_DATABASE_URL") || config[:url] + url = if is_nil(url), do: nil, else: to_string(url) + + {:ok, Keyword.put(opts, :url, url)} + end +end diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/paradedb_test.exs new file mode 100644 index 000000000..c55e2aa47 --- /dev/null +++ b/test/pleroma/search/paradedb_test.exs @@ -0,0 +1,168 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# 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 + assert is_integer(object_id) + assert is_binary(object_ap_id) + assert actor_ap_id == activity.data["actor"] + assert content == "guys i just don'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"}) + + 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]]}} + 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}] = + ParadeDB.search(nil, "swamp", limit: 40, offset: 0) + end + end +end diff --git a/test/support/mocks.ex b/test/support/mocks.ex index 7e1244220..84465509a 100644 --- a/test/support/mocks.ex +++ b/test/support/mocks.ex @@ -46,3 +46,5 @@ Mox.defmock(Pleroma.SignatureMock, for: Pleroma.Signature.API) Mox.defmock(Pleroma.Web.ActivityPub.TransmogrifierMock, for: Pleroma.Web.ActivityPub.Transmogrifier.API ) + +Mox.defmock(Pleroma.Search.ParadeDB.ClientMock, for: Pleroma.Search.ParadeDB.Client)