From 3a87250e15f46b276171f93ba99b8475aa483ebf Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Tue, 3 Mar 2026 12:17:34 +0400 Subject: [PATCH 01/18] Add ParadeDB external search backend Introduce an optional search backend that indexes public/unlisted Notes into a separate ParadeDB (pg_search) Postgres and queries it for status search results. --- changelog.d/paradedb-search.add | 1 + config/config.exs | 8 + config/description.exs | 29 +++- docs/configuration/search.md | 34 ++++ lib/pleroma/application.ex | 23 ++- lib/pleroma/search/paradedb.ex | 213 ++++++++++++++++++++++++++ lib/pleroma/search/paradedb/client.ex | 24 +++ lib/pleroma/search/paradedb/repo.ex | 24 +++ test/pleroma/search/paradedb_test.exs | 168 ++++++++++++++++++++ test/support/mocks.ex | 2 + 10 files changed, 520 insertions(+), 6 deletions(-) create mode 100644 changelog.d/paradedb-search.add create mode 100644 lib/pleroma/search/paradedb.ex create mode 100644 lib/pleroma/search/paradedb/client.ex create mode 100644 lib/pleroma/search/paradedb/repo.ex create mode 100644 test/pleroma/search/paradedb_test.exs 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) From 911d84411dc4eabe8bbca8d90cb898ca9f52f4bd Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Tue, 3 Mar 2026 12:36:01 +0400 Subject: [PATCH 02/18] Add ParadeDB docker integration test Add a skipped-by-default integration test that boots a ParadeDB container, creates the BM25 index, indexes posts, and verifies search results against the real backend. --- .../search/paradedb_integration_test.exs | 209 ++++++++++++++++++ test/test_helper.exs | 2 +- 2 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 test/pleroma/search/paradedb_integration_test.exs diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/paradedb_integration_test.exs new file mode 100644 index 000000000..04e6a9d59 --- /dev/null +++ b/test/pleroma/search/paradedb_integration_test.exs @@ -0,0 +1,209 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2022 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Search.ParadeDBIntegrationTest do + use Pleroma.DataCase, async: false + + @moduletag :integration + @moduletag timeout: 180_000 + + import Mox + import Pleroma.Factory + + alias Pleroma.Activity + alias Pleroma.Search.ParadeDB + alias Pleroma.Web.CommonAPI + + @image "paradedb/paradedb:latest" + @db_user "postgres" + @db_password "postgres" + @db_name "paradedb" + @docker_startup_timeout_ms 60_000 + + setup do + Mox.stub_with(Pleroma.UnstubbedConfigMock, Pleroma.Config) + :ok + end + + setup_all do + with :ok <- ensure_docker_available() do + container = "pleroma-paradedb-it-#{System.unique_integer([:positive])}" + + on_exit(fn -> + _ = docker(["rm", "-f", container]) + end) + + {id, 0} = + docker([ + "run", + "--rm", + "--name", + container, + "-e", + "POSTGRES_USER=#{@db_user}", + "-e", + "POSTGRES_PASSWORD=#{@db_password}", + "-e", + "POSTGRES_DB=#{@db_name}", + "-p", + "5432", + "-d", + @image + ]) + + container_id = String.trim(id) + + host_port = + container + |> wait_for_port!(@docker_startup_timeout_ms) + + wait_for_pg_ready!(container, @docker_startup_timeout_ms) + + url = "postgres://#{@db_user}:#{@db_password}@127.0.0.1:#{host_port}/#{@db_name}" + + prev_env = System.get_env("PARADEDB_DATABASE_URL") + System.put_env("PARADEDB_DATABASE_URL", url) + + on_exit(fn -> + if is_nil(prev_env) do + System.delete_env("PARADEDB_DATABASE_URL") + else + System.put_env("PARADEDB_DATABASE_URL", prev_env) + end + end) + + {:ok, _pid} = start_supervised(Pleroma.Search.ParadeDB.Repo) + + table = "pleroma_search_documents_it_#{System.unique_integer([:positive])}" + Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], table) + + {:ok, + %{ + container: container, + container_id: container_id, + url: url, + table: table + }} + end + end + + test "indexes and searches against real ParadeDB" do + :ok = ParadeDB.drop_index() + assert :ok = ParadeDB.create_index() + + user = insert(:user) + + {:ok, a1} = + CommonAPI.post(user, %{status: "paradedb integration first", visibility: "public"}) + + :timer.sleep(1100) + + {:ok, a2} = + CommonAPI.post(user, %{status: "paradedb integration second", visibility: "public"}) + + {:ok, private} = + CommonAPI.post(user, %{status: "paradedb integration private", visibility: "private"}) + + a1 = Activity.get_by_id_with_object(a1.id) + a2 = Activity.get_by_id_with_object(a2.id) + private = Activity.get_by_id_with_object(private.id) + + assert :ok = ParadeDB.add_to_index(a1) + assert :ok = ParadeDB.add_to_index(a2) + assert :ok = ParadeDB.add_to_index(private) + + results = ParadeDB.search(nil, "paradedb integration", limit: 40, offset: 0) + ids = Enum.map(results, & &1.id) + + assert a1.id in ids + assert a2.id in ids + refute private.id in ids + assert List.first(ids) == a2.id + + # Removing an object should remove it from the ParadeDB index. + assert :ok = ParadeDB.remove_from_index(a2.object) + + results_after_delete = ParadeDB.search(nil, "paradedb integration", limit: 40, offset: 0) + ids_after_delete = Enum.map(results_after_delete, & &1.id) + + assert a1.id in ids_after_delete + refute a2.id in ids_after_delete + end + + defp ensure_docker_available do + if is_nil(System.find_executable("docker")) do + {:skip, "Docker not installed"} + else + case docker(["info"]) do + {_out, 0} -> :ok + {out, _} -> {:skip, "Docker not available: #{String.trim(out)}"} + end + end + end + + defp wait_for_port!(container, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + + wait = fn wait -> + case docker(["port", container, "5432/tcp"]) do + {out, 0} when byte_size(out) > 0 -> + out + |> String.split("\n", trim: true) + |> List.first() + |> parse_host_port!() + + _ -> + if System.monotonic_time(:millisecond) > deadline do + raise "Timed out waiting for docker port mapping" + else + :timer.sleep(250) + wait.(wait) + end + end + end + + wait.(wait) + end + + defp parse_host_port!(mapping) do + mapping + |> String.trim() + |> String.split(":") + |> List.last() + |> String.to_integer() + end + + defp wait_for_pg_ready!(container, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + + wait = fn wait -> + case docker([ + "exec", + container, + "pg_isready", + "-U", + @db_user, + "-d", + @db_name + ]) do + {_out, 0} -> + :ok + + _ -> + if System.monotonic_time(:millisecond) > deadline do + raise "Timed out waiting for ParadeDB Postgres to become ready" + else + :timer.sleep(250) + wait.(wait) + end + end + end + + wait.(wait) + end + + defp docker(args) do + System.cmd("docker", args, stderr_to_stdout: true) + end +end diff --git a/test/test_helper.exs b/test/test_helper.exs index dc6c05a74..bbc00b7df 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -4,7 +4,7 @@ ExUnit.configure(capture_log: true, max_cases: System.schedulers_online()) -ExUnit.start(exclude: [:federated]) +ExUnit.start(exclude: [:federated, :integration]) if match?({:unix, :darwin}, :os.type()) do excluded = ExUnit.configuration() |> Keyword.get(:exclude, []) From efb75488d9f0f8cd8d9e629ec6c925420d7b174c Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Tue, 3 Mar 2026 12:49:23 +0400 Subject: [PATCH 03/18] Tighten ParadeDB integration test Remove warnings that break --warnings-as-errors, add TCP readiness check, and start the ParadeDB Repo with a small pool to reduce connection churn while booting the dockerized backend. --- .../search/paradedb_integration_test.exs | 26 +++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/paradedb_integration_test.exs index 04e6a9d59..1da131c6b 100644 --- a/test/pleroma/search/paradedb_integration_test.exs +++ b/test/pleroma/search/paradedb_integration_test.exs @@ -8,7 +8,6 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do @moduletag :integration @moduletag timeout: 180_000 - import Mox import Pleroma.Factory alias Pleroma.Activity @@ -59,6 +58,7 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do |> wait_for_port!(@docker_startup_timeout_ms) wait_for_pg_ready!(container, @docker_startup_timeout_ms) + wait_for_tcp_ready!(host_port, @docker_startup_timeout_ms) url = "postgres://#{@db_user}:#{@db_password}@127.0.0.1:#{host_port}/#{@db_name}" @@ -73,7 +73,7 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do end end) - {:ok, _pid} = start_supervised(Pleroma.Search.ParadeDB.Repo) + {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 1}) table = "pleroma_search_documents_it_#{System.unique_integer([:positive])}" Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], table) @@ -203,6 +203,28 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do wait.(wait) end + defp wait_for_tcp_ready!(port, timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + + wait = fn wait -> + case :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false], 500) do + {:ok, socket} -> + :gen_tcp.close(socket) + :ok + + {:error, _} -> + if System.monotonic_time(:millisecond) > deadline do + raise "Timed out waiting for TCP port #{port} to accept connections" + else + :timer.sleep(250) + wait.(wait) + end + end + end + + wait.(wait) + end + defp docker(args) do System.cmd("docker", args, stderr_to_stdout: true) end From dee7ac0415e0de1c1f2d44d1dcd1202d5c7d099c Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Wed, 4 Mar 2026 13:04:41 +0400 Subject: [PATCH 04/18] Fix ParadeDB key field handling Use uuid ids in the ParadeDB documents table to satisfy BM25 key_field requirements and adjust indexing/search to dump FlakeIds consistently. Also fix the ParadeDB backend unit test to avoid invalid pinning on struct fields. --- lib/pleroma/search/paradedb.ex | 74 +++++++++++++++++---------- test/pleroma/search/paradedb_test.exs | 14 +++-- 2 files changed, 58 insertions(+), 30 deletions(-) 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 @@ -67,37 +67,43 @@ defmodule Pleroma.Search.ParadeDB 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 - """ + 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 = [ - activity.id, - maybe_search_data.id, - maybe_search_data.ap, - actor_ap_id, - maybe_search_data.content, - 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 + case client_impl().query(sql, params) do + {:ok, _} -> + :ok - {:error, error} -> - Logger.error("ParadeDB add_to_index failed: #{inspect(error)}") - {:error, error} + {:error, error} -> + Logger.error("ParadeDB add_to_index failed: #{inspect(error)}") + {:error, error} + end end else :ok @@ -139,7 +145,10 @@ defmodule Pleroma.Search.ParadeDB do 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() @@ -189,7 +198,7 @@ defmodule Pleroma.Search.ParadeDB do 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, @@ -210,4 +219,15 @@ defmodule Pleroma.Search.ParadeDB do 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 @@ -35,7 +35,9 @@ defmodule Pleroma.Search.ParadeDBTest do [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"] @@ -144,12 +146,18 @@ defmodule Pleroma.Search.ParadeDBTest do {: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 @@ -161,7 +169,7 @@ defmodule Pleroma.Search.ParadeDBTest do "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 From 74da3e01f7f0b27fcbbec313db363118c51388a8 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Thu, 5 Mar 2026 16:36:22 +0400 Subject: [PATCH 05/18] Harden ParadeDB search result safety Re-apply public Create visibility checks when resolving backend hits, improve UUID/FlakeId id parsing, and add tests for poisoned hits and UUID-string ids. Also tighten the docker integration test with localhost-only binding and config restoration. --- lib/pleroma/search/paradedb.ex | 41 ++++++++++-- .../search/paradedb_integration_test.exs | 13 +++- test/pleroma/search/paradedb_test.exs | 63 +++++++++++++++++++ 3 files changed, 111 insertions(+), 6 deletions(-) diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/paradedb.ex index e5d3018d7..01cbc323f 100644 --- a/lib/pleroma/search/paradedb.ex +++ b/lib/pleroma/search/paradedb.ex @@ -6,6 +6,7 @@ defmodule Pleroma.Search.ParadeDB do @behaviour Pleroma.Search.SearchBackend require Logger + require Pleroma.Constants import Ecto.Query @@ -153,6 +154,7 @@ defmodule Pleroma.Search.ParadeDB do from(a in Activity, where: a.id in ^dumped_ids) |> Activity.with_preloaded_object() |> Activity.restrict_deactivated_users() + |> maybe_restrict_public(user) |> maybe_restrict_local(user) |> maybe_restrict_author(author) |> maybe_restrict_blocked(user) @@ -168,6 +170,25 @@ defmodule Pleroma.Search.ParadeDB do |> maybe_fetch(user, query) end + defp maybe_restrict_public(query, %Pleroma.User{}) do + intended_recipients = [ + Pleroma.Constants.as_public(), + Pleroma.Web.ActivityPub.Utils.as_local_public() + ] + + from(a in query, + where: fragment("?->>'type' = 'Create'", a.data), + where: fragment("? && ?", ^intended_recipients, a.recipients) + ) + end + + defp maybe_restrict_public(query, _user) do + from(a in query, + where: fragment("?->>'type' = 'Create'", a.data), + where: ^Pleroma.Constants.as_public() in a.recipients + ) + end + defp search_ids(query, limit, offset, author) do base_sql = "SELECT id FROM #{table()} WHERE content &&& $1" params = [query] @@ -186,7 +207,7 @@ defmodule Pleroma.Search.ParadeDB do sql = base_sql <> - " ORDER BY published_at DESC LIMIT $#{limit_arg} OFFSET $#{offset_arg}" + " ORDER BY published_at DESC, id DESC LIMIT $#{limit_arg} OFFSET $#{offset_arg}" params = params ++ [limit, offset] @@ -223,11 +244,23 @@ defmodule Pleroma.Search.ParadeDB do 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 + case Ecto.UUID.dump(id) do + {:ok, <<_::binary-size(16)>> = uuid} -> + uuid + + :error -> + dump_flake_activity_id(id) end end defp dump_activity_id(_), do: nil + + defp dump_flake_activity_id(id) do + case FlakeId.Ecto.CompatType.dump(id) do + {:ok, <<_::binary-size(16)>> = uuid} -> uuid + _ -> nil + end + rescue + _ -> nil + end end diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/paradedb_integration_test.exs index 1da131c6b..42d12c4d6 100644 --- a/test/pleroma/search/paradedb_integration_test.exs +++ b/test/pleroma/search/paradedb_integration_test.exs @@ -14,7 +14,7 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do alias Pleroma.Search.ParadeDB alias Pleroma.Web.CommonAPI - @image "paradedb/paradedb:latest" + @image System.get_env("PARADEDB_TEST_IMAGE", "paradedb/paradedb:latest") @db_user "postgres" @db_password "postgres" @db_name "paradedb" @@ -46,7 +46,7 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do "-e", "POSTGRES_DB=#{@db_name}", "-p", - "5432", + "127.0.0.1::5432", "-d", @image ]) @@ -75,9 +75,18 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 1}) + previous_table = Pleroma.Config.get([Pleroma.Search.ParadeDB, :table], nil) table = "pleroma_search_documents_it_#{System.unique_integer([:positive])}" Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], table) + on_exit(fn -> + if is_nil(previous_table) do + Pleroma.Config.delete([Pleroma.Search.ParadeDB, :table]) + else + Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], previous_table) + end + end) + {:ok, %{ container: container, diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/paradedb_test.exs index 3338c78ef..c9bb216df 100644 --- a/test/pleroma/search/paradedb_test.exs +++ b/test/pleroma/search/paradedb_test.exs @@ -172,5 +172,68 @@ defmodule Pleroma.Search.ParadeDBTest do assert [%{id: ^activity2_id}, %{id: ^activity1_id}] = ParadeDB.search(nil, "swamp", limit: 40, offset: 0) end + + test "search excludes non-public activities from backend hits" do + user = insert(:user) + + {:ok, public_activity} = + CommonAPI.post(user, %{status: "public swamp", visibility: "public"}) + + {:ok, private_activity} = + CommonAPI.post(user, %{status: "private swamp", visibility: "private"}) + + public_activity_id = public_activity.id + + {:ok, dumped_public_id} = FlakeId.Ecto.CompatType.dump(public_activity.id) + {:ok, dumped_private_id} = FlakeId.Ecto.CompatType.dump(private_activity.id) + + ClientMock + |> expect(:query, fn sql, params -> + assert sql =~ "SELECT id FROM pleroma_search_documents" + assert ["swamp", _limit, _offset] = params + + {:ok, %{rows: [[dumped_private_id], [dumped_public_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: ^public_activity_id}] = ParadeDB.search(nil, "swamp", limit: 40, offset: 0) + end + + test "search supports backend ids returned as uuid strings" do + user = insert(:user) + + {:ok, activity} = CommonAPI.post(user, %{status: "uuid swamp", visibility: "public"}) + + activity_id = activity.id + {:ok, dumped_id} = FlakeId.Ecto.CompatType.dump(activity_id) + {:ok, uuid_string} = Ecto.UUID.load(dumped_id) + + ClientMock + |> expect(:query, fn sql, params -> + assert sql =~ "SELECT id FROM pleroma_search_documents" + assert ["swamp", _limit, _offset] = params + + {:ok, %{rows: [[uuid_string]]}} + 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: ^activity_id}] = ParadeDB.search(nil, "swamp", limit: 40, offset: 0) + end end end From 7efd296afbf49e5fb4ea8cdf659e1f0cdedd7de9 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Thu, 5 Mar 2026 17:12:03 +0400 Subject: [PATCH 06/18] Add fuzzy matching to ParadeDB search Make ParadeDB search typo-tolerant via configurable fuzzy distance and document the new setting. Add coverage for fuzzy query SQL and id parsing, and keep integration test stability under connection load. --- config/config.exs | 3 +- config/description.exs | 6 +++ docs/configuration/search.md | 5 +- lib/pleroma/search/paradedb.ex | 23 ++++++++- .../search/paradedb_integration_test.exs | 2 +- test/pleroma/search/paradedb_test.exs | 47 +++++++++++++++++-- 6 files changed, 79 insertions(+), 7 deletions(-) diff --git a/config/config.exs b/config/config.exs index 7e374b22b..01dd09a18 100644 --- a/config/config.exs +++ b/config/config.exs @@ -940,7 +940,8 @@ config :pleroma, Pleroma.Search.Meilisearch, config :pleroma, Pleroma.Search.ParadeDB, url: nil, - table: "pleroma_search_documents" + table: "pleroma_search_documents", + fuzzy_distance: 0 config :pleroma, Pleroma.Search.ParadeDB.Repo, pool_size: 10, diff --git a/config/description.exs b/config/description.exs index de454dfb2..9a00afdcd 100644 --- a/config/description.exs +++ b/config/description.exs @@ -3560,6 +3560,12 @@ config :pleroma, :config_description, [ type: :string, description: "Table name used for search documents.", suggestion: ["pleroma_search_documents"] + }, + %{ + key: :fuzzy_distance, + type: :integer, + description: "Edit-distance used for fuzzy token matching (0 disables fuzziness, max 2).", + suggestion: [0, 1, 2] } ] }, diff --git a/docs/configuration/search.md b/docs/configuration/search.md index 780904d10..c0d3f8462 100644 --- a/docs/configuration/search.md +++ b/docs/configuration/search.md @@ -160,7 +160,8 @@ To use it, set the search module to `Pleroma.Search.ParadeDB` and configure the > > 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" +> table: "pleroma_search_documents", +> fuzzy_distance: 0 Then, create the table and BM25 index once: @@ -178,4 +179,6 @@ After that, run an initial indexing pass to backfill existing posts: Note: Like the other external search backends, only public/unlisted Notes are indexed. +Set `fuzzy_distance` to `1` or `2` to allow typo-tolerant matching (`2` is the maximum ParadeDB supports). + 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/search/paradedb.ex b/lib/pleroma/search/paradedb.ex index 01cbc323f..57e5599b5 100644 --- a/lib/pleroma/search/paradedb.ex +++ b/lib/pleroma/search/paradedb.ex @@ -35,6 +35,22 @@ defmodule Pleroma.Search.ParadeDB do Config.get([Pleroma.Search.ParadeDB, :table], @default_table) end + defp fuzzy_distance do + case Config.get([Pleroma.Search.ParadeDB, :fuzzy_distance], 0) do + value when is_integer(value) and value in 0..2 -> + value + + value when is_binary(value) -> + case Integer.parse(value) do + {parsed, ""} when parsed in 0..2 -> parsed + _ -> 0 + end + + _ -> + 0 + end + end + @impl true def create_index do with {:ok, _} <- client_impl().query("CREATE EXTENSION IF NOT EXISTS pg_search", []), @@ -190,7 +206,9 @@ defmodule Pleroma.Search.ParadeDB do end defp search_ids(query, limit, offset, author) do - base_sql = "SELECT id FROM #{table()} WHERE content &&& $1" + base_sql = + "SELECT id FROM #{table()} WHERE content &&& #{search_query_param_sql(fuzzy_distance())}" + params = [query] {base_sql, params} = @@ -216,6 +234,9 @@ defmodule Pleroma.Search.ParadeDB do end end + defp search_query_param_sql(0), do: "$1" + defp search_query_param_sql(distance), do: "$1::pdb.fuzzy(#{distance})" + defp create_table_sql(table) do """ CREATE TABLE IF NOT EXISTS #{table} ( diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/paradedb_integration_test.exs index 42d12c4d6..743dcbaff 100644 --- a/test/pleroma/search/paradedb_integration_test.exs +++ b/test/pleroma/search/paradedb_integration_test.exs @@ -73,7 +73,7 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do end end) - {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 1}) + {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 5}) previous_table = Pleroma.Config.get([Pleroma.Search.ParadeDB, :table], nil) table = "pleroma_search_documents_it_#{System.unique_integer([:positive])}" diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/paradedb_test.exs index c9bb216df..c4fb57cb3 100644 --- a/test/pleroma/search/paradedb_test.exs +++ b/test/pleroma/search/paradedb_test.exs @@ -161,12 +161,15 @@ defmodule Pleroma.Search.ParadeDBTest do end) Config - |> expect(:get, 2, fn + |> expect(:get, 3, fn [Pleroma.Search.ParadeDB, :client_impl], nil -> ClientMock [Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" -> "pleroma_search_documents" + + [Pleroma.Search.ParadeDB, :fuzzy_distance], 0 -> + 0 end) assert [%{id: ^activity2_id}, %{id: ^activity1_id}] = @@ -196,12 +199,15 @@ defmodule Pleroma.Search.ParadeDBTest do end) Config - |> expect(:get, 2, fn + |> expect(:get, 3, fn [Pleroma.Search.ParadeDB, :client_impl], nil -> ClientMock [Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" -> "pleroma_search_documents" + + [Pleroma.Search.ParadeDB, :fuzzy_distance], 0 -> + 0 end) assert [%{id: ^public_activity_id}] = ParadeDB.search(nil, "swamp", limit: 40, offset: 0) @@ -225,15 +231,50 @@ defmodule Pleroma.Search.ParadeDBTest do end) Config - |> expect(:get, 2, fn + |> expect(:get, 3, fn [Pleroma.Search.ParadeDB, :client_impl], nil -> ClientMock [Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" -> "pleroma_search_documents" + + [Pleroma.Search.ParadeDB, :fuzzy_distance], 0 -> + 0 end) assert [%{id: ^activity_id}] = ParadeDB.search(nil, "swamp", limit: 40, offset: 0) end + + test "search uses configured fuzzy distance" do + user = insert(:user) + + {:ok, activity} = CommonAPI.post(user, %{status: "runing shoes", visibility: "public"}) + + activity_id = activity.id + {:ok, dumped_id} = FlakeId.Ecto.CompatType.dump(activity_id) + + ClientMock + |> expect(:query, fn sql, params -> + assert sql =~ "SELECT id FROM pleroma_search_documents" + assert sql =~ "$1::pdb.fuzzy(1)" + assert ["running shoes", _limit, _offset] = params + + {:ok, %{rows: [[dumped_id]]}} + end) + + Config + |> expect(:get, 3, fn + [Pleroma.Search.ParadeDB, :client_impl], nil -> + ClientMock + + [Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" -> + "pleroma_search_documents" + + [Pleroma.Search.ParadeDB, :fuzzy_distance], 0 -> + 1 + end) + + assert [%{id: ^activity_id}] = ParadeDB.search(nil, "running shoes", limit: 40, offset: 0) + end end end From 4e65bbf13469d8147dc780c7e655fefdb12bdebf Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Thu, 5 Mar 2026 17:34:48 +0400 Subject: [PATCH 07/18] Fix ParadeDB fuzzy query parameter casting Cast the bound search parameter through text before so Postgrex can handle the prepared statement type inference. Add matching test assertion updates. --- lib/pleroma/search/paradedb.ex | 2 +- test/pleroma/search/paradedb_test.exs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/paradedb.ex index 57e5599b5..a6649a894 100644 --- a/lib/pleroma/search/paradedb.ex +++ b/lib/pleroma/search/paradedb.ex @@ -235,7 +235,7 @@ defmodule Pleroma.Search.ParadeDB do end defp search_query_param_sql(0), do: "$1" - defp search_query_param_sql(distance), do: "$1::pdb.fuzzy(#{distance})" + defp search_query_param_sql(distance), do: "($1::text)::pdb.fuzzy(#{distance})" defp create_table_sql(table) do """ diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/paradedb_test.exs index c4fb57cb3..3a3b8298f 100644 --- a/test/pleroma/search/paradedb_test.exs +++ b/test/pleroma/search/paradedb_test.exs @@ -256,7 +256,7 @@ defmodule Pleroma.Search.ParadeDBTest do ClientMock |> expect(:query, fn sql, params -> assert sql =~ "SELECT id FROM pleroma_search_documents" - assert sql =~ "$1::pdb.fuzzy(1)" + assert sql =~ "($1::text)::pdb.fuzzy(1)" assert ["running shoes", _limit, _offset] = params {:ok, %{rows: [[dumped_id]]}} From cae3c9a980945f107b2cd6a321b0d1d1e0d243ea Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Thu, 5 Mar 2026 17:39:34 +0400 Subject: [PATCH 08/18] Add ParadeDB fuzzy integration coverage Extend the docker-backed ParadeDB integration test to verify typo-tolerant status search with fuzzy_distance enabled, covering prepared-parameter fuzzy casts against a real backend. --- test/pleroma/search/paradedb_integration_test.exs | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/paradedb_integration_test.exs index 743dcbaff..2687260cf 100644 --- a/test/pleroma/search/paradedb_integration_test.exs +++ b/test/pleroma/search/paradedb_integration_test.exs @@ -130,6 +130,20 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do refute private.id in ids assert List.first(ids) == a2.id + previous_fuzzy_distance = Pleroma.Config.get([Pleroma.Search.ParadeDB, :fuzzy_distance], 0) + Pleroma.Config.put([Pleroma.Search.ParadeDB, :fuzzy_distance], 1) + + on_exit(fn -> + Pleroma.Config.put([Pleroma.Search.ParadeDB, :fuzzy_distance], previous_fuzzy_distance) + end) + + fuzzy_results = ParadeDB.search(nil, "paradedb integraion", limit: 40, offset: 0) + fuzzy_ids = Enum.map(fuzzy_results, & &1.id) + + assert a1.id in fuzzy_ids + assert a2.id in fuzzy_ids + refute private.id in fuzzy_ids + # Removing an object should remove it from the ParadeDB index. assert :ok = ParadeDB.remove_from_index(a2.object) From 8ac967518b1729aa2530bb2e3003a31bc40cabfb Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Wed, 25 Mar 2026 15:11:45 -0700 Subject: [PATCH 09/18] Switch to the newly relocated Pleroma.Search.object_to_search_data/1 --- lib/pleroma/search/paradedb.ex | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/paradedb.ex index a6649a894..20be5b6f9 100644 --- a/lib/pleroma/search/paradedb.ex +++ b/lib/pleroma/search/paradedb.ex @@ -18,10 +18,9 @@ defmodule Pleroma.Search.ParadeDB do maybe_restrict_local: 2 ] - import Pleroma.Search.Meilisearch, only: [object_to_search_data: 1] - alias Pleroma.Activity alias Pleroma.Config.Getting, as: Config + alias Pleroma.Search @max_limit 40 @default_table "pleroma_search_documents" @@ -81,7 +80,7 @@ defmodule Pleroma.Search.ParadeDB do @impl true def add_to_index(activity) do - maybe_search_data = object_to_search_data(activity.object) + maybe_search_data = Search.object_to_search_data(activity.object) if activity.data["type"] == "Create" and maybe_search_data do dumped_activity_id = dump_activity_id(activity.id) From 1e3456a1604ebe5818e319f47bbeb79828c3d6e5 Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Wed, 25 Mar 2026 15:14:35 -0700 Subject: [PATCH 10/18] Ensure the Ecto Repo for ParadeDB is started for the data indexing Mix task --- lib/mix/pleroma.ex | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/lib/mix/pleroma.ex b/lib/mix/pleroma.ex index e7869919c..372bf3399 100644 --- a/lib/mix/pleroma.ex +++ b/lib/mix/pleroma.ex @@ -62,7 +62,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) ++ + search_children() cachex_children = Enum.map(@cachex_children, &Pleroma.Application.build_cachex(&1, [])) @@ -141,4 +142,14 @@ defmodule Mix.Pleroma do end defp http_children(_), do: [] + + defp search_children do + case Config.get([Pleroma.Search, :module]) do + Pleroma.Search.ParadeDB -> + [Pleroma.Search.ParadeDB.Repo] + + _ -> + [] + end + end end From 3776f13f20172fa7c88127ba28583fba27cb1e1c Mon Sep 17 00:00:00 2001 From: Mark Felder Date: Wed, 1 Apr 2026 19:10:29 +0000 Subject: [PATCH 11/18] Fix warning --- lib/mix/pleroma.ex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/mix/pleroma.ex b/lib/mix/pleroma.ex index 372bf3399..aebad6984 100644 --- a/lib/mix/pleroma.ex +++ b/lib/mix/pleroma.ex @@ -144,7 +144,7 @@ defmodule Mix.Pleroma do defp http_children(_), do: [] defp search_children do - case Config.get([Pleroma.Search, :module]) do + case Pleroma.Config.get([Pleroma.Search, :module]) do Pleroma.Search.ParadeDB -> [Pleroma.Search.ParadeDB.Repo] From cd28c2089c2fddce4e480bd097830ce22f923c47 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Mon, 25 May 2026 18:45:47 +0400 Subject: [PATCH 12/18] Fix ParadeDB search indexing --- changelog.d/paradedb-search.add | 2 +- config/config.exs | 2 +- docs/configuration/search.md | 8 +- lib/pleroma/search.ex | 74 +++++++++------ .../search/{paradedb.ex => parade_db.ex} | 15 +-- .../search/{paradedb => parade_db}/client.ex | 0 .../search/{paradedb => parade_db}/repo.ex | 0 .../activity_pub/mrf/anti_followbot_policy.ex | 6 +- .../web/activity_pub/mrf/simple_policy.ex | 2 +- .../web/mastodon_api/views/status_view.ex | 4 +- lib/pleroma/workers/mailer_worker.ex | 4 +- test/pleroma/search/meilisearch_test.exs | 2 +- ...est.exs => parade_db_integration_test.exs} | 0 .../{paradedb_test.exs => parade_db_test.exs} | 20 ++-- test/pleroma/search_test.exs | 94 +++++++++++++++++-- 15 files changed, 175 insertions(+), 58 deletions(-) rename lib/pleroma/search/{paradedb.ex => parade_db.ex} (96%) rename lib/pleroma/search/{paradedb => parade_db}/client.ex (100%) rename lib/pleroma/search/{paradedb => parade_db}/repo.ex (100%) rename test/pleroma/search/{paradedb_integration_test.exs => parade_db_integration_test.exs} (100%) rename test/pleroma/search/{paradedb_test.exs => parade_db_test.exs} (94%) diff --git a/changelog.d/paradedb-search.add b/changelog.d/paradedb-search.add index 00a6e2361..f72ebca14 100644 --- a/changelog.d/paradedb-search.add +++ b/changelog.d/paradedb-search.add @@ -1 +1 @@ -Add ParadeDB (pg_search) as an optional external search backend. +Add ParadeDB (pg_search) as an optional external search backend, indexing status content, content warnings, and attachment descriptions. diff --git a/config/config.exs b/config/config.exs index 2e405413c..3c308f093 100644 --- a/config/config.exs +++ b/config/config.exs @@ -946,7 +946,7 @@ config :pleroma, Pleroma.Search.ParadeDB, fuzzy_distance: 0 config :pleroma, Pleroma.Search.ParadeDB.Repo, - pool_size: 10, + pool_size: 2, prepare: :unnamed config :pleroma, Pleroma.Application, diff --git a/docs/configuration/search.md b/docs/configuration/search.md index c0d3f8462..1d5056ddf 100644 --- a/docs/configuration/search.md +++ b/docs/configuration/search.md @@ -85,7 +85,7 @@ You will see a "Default Admin API Key", this is the key you actually put into yo ### Initial indexing -After setting up the configuration, you'll want to index all of your already existing posts. Only public posts are indexed. You'll only +After setting up the configuration, you'll want to index all of your already existing posts. Only public/unlisted posts are indexed. You'll only have to do it one time, but it might take a while, depending on the amount of posts your instance has seen. This is also a fairly RAM consuming process for `meilisearch`, and it will take a lot of RAM when running if you have a lot of posts (seems to be around 5G for ~1.2 million posts while idle and up to 7G while indexing initially, but your experience may be different). @@ -153,8 +153,10 @@ external search backend by pointing Pleroma at a separate Postgres instance with Pleroma will maintain a small `pleroma_search_documents` table in that database (via the existing search indexing queue) and run search queries against it. +The indexed text includes status content, content warnings, and attachment descriptions. To use it, set the search module to `Pleroma.Search.ParadeDB` and configure the ParadeDB database URL: +These settings must be in static/runtime config (or `PARADEDB_DATABASE_URL`) so the dedicated ParadeDB Repo is started with the application. > config :pleroma, Pleroma.Search, module: Pleroma.Search.ParadeDB > @@ -162,6 +164,10 @@ To use it, set the search module to `Pleroma.Search.ParadeDB` and configure the > url: System.get_env("PARADEDB_DATABASE_URL") || "postgres://postgres:postgres@127.0.0.1:5432/paradedb", > table: "pleroma_search_documents", > fuzzy_distance: 0 +> +> config :pleroma, Pleroma.Search.ParadeDB.Repo, +> pool_size: 2, +> prepare: :unnamed Then, create the table and BM25 index once: diff --git a/lib/pleroma/search.ex b/lib/pleroma/search.ex index 9cd2768c4..cd4907026 100644 --- a/lib/pleroma/search.ex +++ b/lib/pleroma/search.ex @@ -5,13 +5,12 @@ defmodule Pleroma.Search do alias Pleroma.Workers.SearchIndexingWorker @spec add_to_index(Activity.t()) :: {:ok, Oban.Job.t() | :noop} | {:error, Oban.Job.changeset()} - def add_to_index(%Activity{id: activity_id, object: %Object{} = object} = activity) do - with {_, true} <- {:indexable, indexable?(activity)}, - {_, "public"} <- {:visibility, Visibility.get_visibility(object)} do + def add_to_index(%Activity{id: activity_id, object: %Object{}} = activity) do + if indexable?(activity) do SearchIndexingWorker.new(%{"op" => "add_to_index", "activity" => activity_id}) |> Oban.insert() else - _ -> {:ok, :noop} + {:ok, :noop} end end @@ -41,40 +40,59 @@ defmodule Pleroma.Search do def object_to_search_data(%Object{} = object) do data = object.data - content_str = - case data["content"] do - [nil | rest] -> to_string(rest) - str -> str - end - content = - with {:ok, scrubbed} <- - FastSanitize.Sanitizer.scrub(content_str, Pleroma.HTML.Scrubber.SearchIndexing), - trimmed <- String.trim(scrubbed) do - trimmed - end - - # Make sure we have a non-empty string - if content != "" do - {:ok, published, _} = DateTime.from_iso8601(data["published"]) + data + |> search_texts() + |> Enum.map(&sanitize_search_text/1) + |> Enum.reject(&(&1 in ["", "."])) + |> Enum.join(" ") + with true <- content != "", + published when is_binary(published) <- data["published"], + {:ok, published, _} <- DateTime.from_iso8601(published) do %{ id: object.id, content: content, ap: data["id"], published: published |> DateTime.to_unix() } + else + _ -> nil end end - defp indexable?(%Activity{ - data: %{"type" => "Create"}, - object: %Object{ - data: %{"content" => content, "published" => published, "type" => "Note"} - } - }) - when not is_nil(content) and content not in ["", "."] and not is_nil(published), - do: true + def indexable?(%Activity{ + data: %{"type" => "Create"}, + object: %Object{data: %{"published" => published, "type" => "Note"}} = object + }) + when not is_nil(published) do + Visibility.get_visibility(object) in ["public", "unlisted"] and + not is_nil(object_to_search_data(object)) + end - defp indexable?(_), do: false + def indexable?(_), do: false + + defp search_texts(data) do + [data["content"], data["summary"] | attachment_names(data["attachment"])] + end + + defp attachment_names(attachments) when is_list(attachments) do + Enum.map(attachments, fn + %{"name" => name} -> name + _ -> nil + end) + end + + defp attachment_names(_), do: [] + + defp sanitize_search_text([nil | rest]), do: sanitize_search_text(to_string(rest)) + + defp sanitize_search_text(text) when is_binary(text) do + case FastSanitize.Sanitizer.scrub(text, Pleroma.HTML.Scrubber.SearchIndexing) do + {:ok, scrubbed} -> String.trim(scrubbed) + _ -> "" + end + end + + defp sanitize_search_text(_), do: "" end diff --git a/lib/pleroma/search/paradedb.ex b/lib/pleroma/search/parade_db.ex similarity index 96% rename from lib/pleroma/search/paradedb.ex rename to lib/pleroma/search/parade_db.ex index 20be5b6f9..ee00e74af 100644 --- a/lib/pleroma/search/paradedb.ex +++ b/lib/pleroma/search/parade_db.ex @@ -66,9 +66,10 @@ defmodule Pleroma.Search.ParadeDB do @impl true def drop_index do - with {:ok, _} <- client_impl().query("DROP TABLE IF EXISTS #{table()}", []) do - :ok - else + case client_impl().query("DROP TABLE IF EXISTS #{table()}", []) do + {:ok, _} -> + :ok + {:error, error} -> Logger.error("ParadeDB drop_index failed: #{inspect(error)}") {:error, error} @@ -80,9 +81,9 @@ defmodule Pleroma.Search.ParadeDB do @impl true def add_to_index(activity) do - maybe_search_data = Search.object_to_search_data(activity.object) - - if activity.data["type"] == "Create" and maybe_search_data do + with true <- Search.indexable?(activity), + maybe_search_data when not is_nil(maybe_search_data) <- + Search.object_to_search_data(activity.object) do dumped_activity_id = dump_activity_id(activity.id) actor_ap_id = activity.data["actor"] published_at = DateTime.from_unix!(maybe_search_data.published) @@ -122,7 +123,7 @@ defmodule Pleroma.Search.ParadeDB do end end else - :ok + _ -> :ok end end diff --git a/lib/pleroma/search/paradedb/client.ex b/lib/pleroma/search/parade_db/client.ex similarity index 100% rename from lib/pleroma/search/paradedb/client.ex rename to lib/pleroma/search/parade_db/client.ex diff --git a/lib/pleroma/search/paradedb/repo.ex b/lib/pleroma/search/parade_db/repo.ex similarity index 100% rename from lib/pleroma/search/paradedb/repo.ex rename to lib/pleroma/search/parade_db/repo.ex diff --git a/lib/pleroma/web/activity_pub/mrf/anti_followbot_policy.ex b/lib/pleroma/web/activity_pub/mrf/anti_followbot_policy.ex index 8ea61aec2..d0320ce49 100644 --- a/lib/pleroma/web/activity_pub/mrf/anti_followbot_policy.ex +++ b/lib/pleroma/web/activity_pub/mrf/anti_followbot_policy.ex @@ -24,7 +24,11 @@ defmodule Pleroma.Web.ActivityPub.MRF.AntiFollowbotPolicy do defp score_displayname("fedibot"), do: 1.0 defp score_displayname(_), do: 0.0 - defp determine_if_followbot(%User{nickname: nickname, name: displayname, actor_type: actor_type}) do + defp determine_if_followbot(%User{ + nickname: nickname, + name: displayname, + actor_type: actor_type + }) do # nickname will be a binary string except when following a relay nick_score = if is_binary(nickname) do diff --git a/lib/pleroma/web/activity_pub/mrf/simple_policy.ex b/lib/pleroma/web/activity_pub/mrf/simple_policy.ex index a97e8db7b..93f0616ec 100644 --- a/lib/pleroma/web/activity_pub/mrf/simple_policy.ex +++ b/lib/pleroma/web/activity_pub/mrf/simple_policy.ex @@ -108,7 +108,7 @@ defmodule Pleroma.Web.ActivityPub.MRF.SimplePolicy do end defp intersection(list1, list2) do - list1 -- list1 -- list2 + list1 -- (list1 -- list2) end defp check_followers_only(%{host: actor_host} = _actor_info, activity) do diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex index 6949c5f2d..5b18261a7 100644 --- a/lib/pleroma/web/mastodon_api/views/status_view.ex +++ b/lib/pleroma/web/mastodon_api/views/status_view.ex @@ -680,7 +680,9 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do %{ancestors: ancestors, descendants: descendants} = activities |> Enum.reverse() - |> Enum.group_by(fn %{id: id} -> if id < activity.id, do: :ancestors, else: :descendants end) + |> Enum.group_by(fn %{id: id} -> + if id < activity.id, do: :ancestors, else: :descendants + end) |> Map.put_new(:ancestors, []) |> Map.put_new(:descendants, []) diff --git a/lib/pleroma/workers/mailer_worker.ex b/lib/pleroma/workers/mailer_worker.ex index b0259b191..9ef6a80c9 100644 --- a/lib/pleroma/workers/mailer_worker.ex +++ b/lib/pleroma/workers/mailer_worker.ex @@ -6,7 +6,9 @@ defmodule Pleroma.Workers.MailerWorker do use Oban.Worker, queue: :background @impl true - def perform(%Job{args: %{"op" => "email", "encoded_email" => encoded_email, "config" => config}}) do + def perform(%Job{ + args: %{"op" => "email", "encoded_email" => encoded_email, "config" => config} + }) do encoded_email |> Base.decode64!() |> :erlang.binary_to_term() diff --git a/test/pleroma/search/meilisearch_test.exs b/test/pleroma/search/meilisearch_test.exs index ff32491c5..6e4a42c95 100644 --- a/test/pleroma/search/meilisearch_test.exs +++ b/test/pleroma/search/meilisearch_test.exs @@ -99,7 +99,7 @@ defmodule Pleroma.Search.MeilisearchTest do %{method: :delete, url: "http://127.0.0.1:7700/indexes/objects/documents/" <> id} -> send(self(), "called_delete") - assert String.length(id) > 1 + assert id != "" json(%{}) end) diff --git a/test/pleroma/search/paradedb_integration_test.exs b/test/pleroma/search/parade_db_integration_test.exs similarity index 100% rename from test/pleroma/search/paradedb_integration_test.exs rename to test/pleroma/search/parade_db_integration_test.exs diff --git a/test/pleroma/search/paradedb_test.exs b/test/pleroma/search/parade_db_test.exs similarity index 94% rename from test/pleroma/search/paradedb_test.exs rename to test/pleroma/search/parade_db_test.exs index 3a3b8298f..3133b6b38 100644 --- a/test/pleroma/search/paradedb_test.exs +++ b/test/pleroma/search/parade_db_test.exs @@ -75,15 +75,21 @@ defmodule Pleroma.Search.ParadeDBTest do }) args = %{"op" => "add_to_index", "activity" => activity.id} - assert_enqueued(worker: SearchIndexingWorker, args: args) + refute_enqueued(worker: SearchIndexingWorker, args: args) + end) + end - Config - |> expect(:get, fn - [Pleroma.Search, :module], nil -> - ParadeDB - end) + test "backend skips non-public posts when called directly" do + user = insert(:user) - assert :ok = perform_job(SearchIndexingWorker, args) + Enum.each(["private", "direct"], fn visibility -> + {:ok, activity} = + CommonAPI.post(user, %{ + status: "guys i just don't wanna leave the swamp", + visibility: visibility + }) + + assert :ok = ParadeDB.add_to_index(activity) end) end diff --git a/test/pleroma/search_test.exs b/test/pleroma/search_test.exs index d777bcda2..00d8b10ed 100644 --- a/test/pleroma/search_test.exs +++ b/test/pleroma/search_test.exs @@ -3,26 +3,32 @@ # SPDX-License-Identifier: AGPL-3.0-only defmodule Pleroma.SearchTest do + require Pleroma.Constants + use Pleroma.DataCase, async: true use Oban.Testing, repo: Pleroma.Repo import Pleroma.Factory + alias Pleroma.Object + alias Pleroma.Search alias Pleroma.Web.CommonAPI alias Pleroma.Workers.SearchIndexingWorker - test "indexes posts that are public" do + test "indexes posts that are public or unlisted" do user = insert(:user) - {:ok, activity} = - CommonAPI.post(user, %{ - status: "Well this is a story all about how my life got flipped turned upside down", - visibility: "public" - }) + Enum.each(["public", "unlisted"], fn visibility -> + {:ok, activity} = + CommonAPI.post(user, %{ + status: "Well this is a story all about how my life got flipped turned upside down", + visibility: visibility + }) - args = %{"op" => "add_to_index", "activity" => activity.id} + args = %{"op" => "add_to_index", "activity" => activity.id} - assert_enqueued(worker: SearchIndexingWorker, args: args) + assert_enqueued(worker: SearchIndexingWorker, args: args) + end) end test "doesn't index posts that are not public" do @@ -66,4 +72,76 @@ defmodule Pleroma.SearchTest do refute_enqueued(worker: SearchIndexingWorker, args: args) end + + test "object_to_search_data includes content warnings and attachment descriptions" do + object = %Object{ + id: 1, + data: %{ + "id" => "https://example.com/objects/1", + "type" => "Note", + "content" => "body text", + "summary" => "subject text", + "attachment" => [ + %{"type" => "Document", "name" => "image description"}, + %{"type" => "Document", "name" => ""} + ], + "published" => DateTime.utc_now() |> DateTime.to_iso8601() + } + } + + assert %{content: content} = Search.object_to_search_data(object) + assert content =~ "body text" + assert content =~ "subject text" + assert content =~ "image description" + end + + test "object_to_search_data ignores objects with invalid published dates" do + for published <- [nil, 123, "not a date"] do + object = %Object{ + id: 1, + data: %{ + "id" => "https://example.com/objects/1", + "type" => "Note", + "content" => "body text", + "published" => published + } + } + + refute Search.object_to_search_data(object) + end + end + + test "indexable? accepts searchable content warnings and attachment descriptions" do + public = Pleroma.Constants.as_public() + published = DateTime.utc_now() |> DateTime.to_iso8601() + + for data <- [ + %{"content" => ".", "summary" => "subject text", "attachment" => []}, + %{ + "content" => ".", + "summary" => "", + "attachment" => [%{"type" => "Document", "name" => "image description"}] + } + ] do + object = %Object{ + id: 1, + data: + Map.merge(data, %{ + "id" => "https://example.com/objects/1", + "type" => "Note", + "to" => [public], + "published" => published + }) + } + + activity = %Pleroma.Activity{ + id: Ecto.UUID.generate(), + data: %{"type" => "Create"}, + object: object, + recipients: [public] + } + + assert Search.indexable?(activity) + end + end end From 91089c93ada742ff2b709601e9f0ea217c9f5a06 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 11:38:24 +0400 Subject: [PATCH 13/18] feat(search): add resumable runtime backfill --- lib/mix/tasks/pleroma/search/indexer.ex | 58 ++++------ lib/pleroma/search/backfill.ex | 137 ++++++++++++++++++++++++ test/pleroma/search/backfill_test.exs | 79 ++++++++++++++ 3 files changed, 234 insertions(+), 40 deletions(-) create mode 100644 lib/pleroma/search/backfill.ex create mode 100644 test/pleroma/search/backfill_test.exs diff --git a/lib/mix/tasks/pleroma/search/indexer.ex b/lib/mix/tasks/pleroma/search/indexer.ex index 2a52472f9..62cbf2e50 100644 --- a/lib/mix/tasks/pleroma/search/indexer.ex +++ b/lib/mix/tasks/pleroma/search/indexer.ex @@ -4,9 +4,6 @@ defmodule Mix.Tasks.Pleroma.Search.Indexer do import Mix.Pleroma - import Ecto.Query - - alias Pleroma.Workers.SearchIndexingWorker def run(["create_index"]) do start_pleroma() @@ -35,49 +32,30 @@ defmodule Mix.Tasks.Pleroma.Search.Indexer do strict: [ chunk: :integer, limit: :integer, - step: :integer + step: :integer, + before: :string ] ) start_pleroma() - chunk_size = Keyword.get(options, :chunk, 100) - limit = Keyword.get(options, :limit, 100_000) - per_step = Keyword.get(options, :step, 1000) + result = + Pleroma.Search.Backfill.run( + options ++ + [ + on_page: fn page -> + IO.puts( + "Queued #{page.count} activities (#{page.enqueued} total), " <> + "checkpoint: #{page.next_cursor}" + ) + end + ] + ) - chunks = max(div(limit, per_step), 1) + IO.puts("Queued #{result.enqueued} activities") - 1..chunks - |> Enum.each(fn step -> - q = - from(a in Pleroma.Activity, - limit: ^per_step, - offset: ^per_step * (^step - 1), - select: [:id], - order_by: [desc: :id] - ) - - {:ok, ids} = - Pleroma.Repo.transaction(fn -> - Pleroma.Repo.stream(q, timeout: :infinity) - |> Enum.map(fn a -> - a.id - end) - end) - - IO.puts("Got #{length(ids)} activities, adding to indexer") - - ids - |> Enum.chunk_every(chunk_size) - |> Enum.each(fn chunk -> - IO.puts("Adding #{length(chunk)} activities to indexing queue") - - chunk - |> Enum.map(fn id -> - SearchIndexingWorker.new(%{"op" => "add_to_index", "activity" => id}) - end) - |> Oban.insert_all() - end) - end) + unless result.exhausted do + IO.puts("Resume with --before #{result.next_cursor}") + end end end diff --git a/lib/pleroma/search/backfill.ex b/lib/pleroma/search/backfill.ex new file mode 100644 index 000000000..1b986bb59 --- /dev/null +++ b/lib/pleroma/search/backfill.ex @@ -0,0 +1,137 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2026 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Search.Backfill do + import Ecto.Query + + alias Pleroma.Activity + alias Pleroma.Workers.SearchIndexingWorker + + @default_chunk_size 100 + @default_step_size 1_000 + @default_limit 100_000 + + @doc """ + Enqueues existing Create activities for search indexing. + + The returned `next_cursor` can be passed as `before` to resume a bounded run. + Pass `limit: :infinity` to continue until the oldest activity is reached. + """ + def run(options \\ []) do + options = + Keyword.validate!(options, + chunk: @default_chunk_size, + step: @default_step_size, + limit: @default_limit, + before: nil, + on_page: nil + ) + + chunk_size = positive_option!(options, :chunk, @default_chunk_size) + step_size = positive_option!(options, :step, @default_step_size) + limit = limit_option!(options) + before = cursor_option!(options[:before]) + on_page = callback_option!(options[:on_page]) + + do_run(before, limit, step_size, chunk_size, on_page, 0) + end + + defp do_run(before, remaining, step_size, chunk_size, on_page, enqueued) do + page_size = page_size(remaining, step_size) + ids = fetch_ids(before, page_size) + + case ids do + [] -> + result(enqueued, before, true) + + ids -> + enqueue(ids, chunk_size) + + page_count = length(ids) + next_cursor = List.last(ids) + total = enqueued + page_count + remaining = decrement(remaining, page_count) + + on_page.(%{count: page_count, enqueued: total, next_cursor: next_cursor}) + + cond do + remaining == 0 -> result(total, next_cursor, false) + page_count < page_size -> result(total, next_cursor, true) + true -> do_run(next_cursor, remaining, step_size, chunk_size, on_page, total) + end + end + end + + defp fetch_ids(before, limit) do + Activity + |> where([a], fragment("?->>'type' = 'Create'", a.data)) + |> maybe_before(before) + |> order_by([a], desc: a.id) + |> limit(^limit) + |> select([a], a.id) + |> Pleroma.Repo.all() + end + + defp maybe_before(query, nil), do: query + defp maybe_before(query, before), do: where(query, [a], a.id < ^before) + + defp enqueue(ids, chunk_size) do + ids + |> Enum.chunk_every(chunk_size) + |> Enum.each(fn chunk -> + chunk + |> Enum.map(&SearchIndexingWorker.new(%{"op" => "add_to_index", "activity" => &1})) + |> Oban.insert_all() + end) + end + + defp page_size(:infinity, step_size), do: step_size + defp page_size(remaining, step_size), do: min(remaining, step_size) + + defp decrement(:infinity, _count), do: :infinity + defp decrement(remaining, count), do: remaining - count + + defp result(enqueued, next_cursor, exhausted) do + %{enqueued: enqueued, next_cursor: next_cursor, exhausted: exhausted} + end + + defp positive_option!(options, key, default) do + case Keyword.get(options, key, default) do + value when is_integer(value) and value > 0 -> value + value -> raise ArgumentError, "#{key} must be a positive integer, got: #{inspect(value)}" + end + end + + defp limit_option!(options) do + case Keyword.fetch!(options, :limit) do + :infinity -> :infinity + value when is_integer(value) and value > 0 -> value + value -> raise ArgumentError, "limit must be a positive integer, got: #{inspect(value)}" + end + end + + defp cursor_option!(nil), do: nil + + defp cursor_option!(cursor) do + valid? = + try do + match?({:ok, <<_::binary-size(16)>>}, FlakeId.Ecto.CompatType.dump(cursor)) + rescue + _ -> false + end + + if valid? do + cursor + else + raise ArgumentError, "before must be a valid activity id, got: #{inspect(cursor)}" + end + end + + defp callback_option!(nil), do: fn _ -> :ok end + defp callback_option!(callback) when is_function(callback, 1), do: callback + + defp callback_option!(callback) do + raise ArgumentError, "on_page must be a function of arity 1, got: #{inspect(callback)}" + end +end diff --git a/test/pleroma/search/backfill_test.exs b/test/pleroma/search/backfill_test.exs new file mode 100644 index 000000000..d707524c3 --- /dev/null +++ b/test/pleroma/search/backfill_test.exs @@ -0,0 +1,79 @@ +# Pleroma: A lightweight social networking server +# Copyright © 2017-2026 Pleroma Authors +# SPDX-License-Identifier: AGPL-3.0-only + +defmodule Pleroma.Search.BackfillTest do + use Pleroma.DataCase, async: false + use Oban.Testing, repo: Pleroma.Repo + + import Pleroma.Factory + + alias Pleroma.Search.Backfill + alias Pleroma.Workers.SearchIndexingWorker + + test "enqueues the exact requested number across partial pages" do + activities = insert_list(5, :note_activity) |> Enum.sort_by(& &1.id, :desc) + + assert %{enqueued: 3, exhausted: false, next_cursor: cursor} = + Backfill.run(limit: 3, step: 2, chunk: 2) + + assert search_indexing_ids() == Enum.map(Enum.take(activities, 3), & &1.id) + assert cursor == Enum.at(activities, 2).id + end + + test "resumes after the last processed activity without duplicates" do + activities = insert_list(5, :note_activity) |> Enum.sort_by(& &1.id, :desc) + + assert %{enqueued: 2, next_cursor: cursor} = Backfill.run(limit: 2, step: 1) + + Pleroma.Repo.delete_all(Oban.Job) + + assert %{enqueued: 2, next_cursor: next_cursor} = + Backfill.run(before: cursor, limit: 2, step: 1) + + assert search_indexing_ids() == + activities + |> Enum.slice(2, 2) + |> Enum.map(& &1.id) + + assert next_cursor == Enum.at(activities, 3).id + end + + test "only queues Create activities" do + create = insert(:note_activity) + insert(:like_activity, note_activity: create) + + assert %{enqueued: 1, exhausted: true, next_cursor: cursor} = Backfill.run(step: 2) + assert search_indexing_ids() == [create.id] + assert cursor == create.id + end + + test "rejects unknown options and malformed cursors before enqueueing" do + assert_raise ArgumentError, fn -> Backfill.run(limt: 1) end + assert_raise ArgumentError, fn -> Backfill.run(before: "not-an-activity-id") end + assert_raise ArgumentError, fn -> Backfill.run(on_page: :invalid) end + + assert all_enqueued(worker: SearchIndexingWorker) == [] + end + + test "does not duplicate or skip the initial activities when newer rows are inserted" do + activities = insert_list(3, :note_activity) |> Enum.sort_by(& &1.id, :desc) + + assert %{enqueued: 3, exhausted: true} = + Backfill.run( + limit: :infinity, + step: 1, + on_page: fn %{enqueued: enqueued} -> + if enqueued == 1, do: insert(:note_activity) + end + ) + + assert search_indexing_ids() == Enum.map(activities, & &1.id) + end + + defp search_indexing_ids do + all_enqueued(worker: SearchIndexingWorker) + |> Enum.sort_by(& &1.id) + |> Enum.map(& &1.args["activity"]) + end +end From 271c3a3dd8a22fad2fa4cd6d034c98e1ffc91fb4 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 11:58:31 +0400 Subject: [PATCH 14/18] fix(config): keep search startup settings static --- lib/mix/tasks/pleroma/config.ex | 9 ++++--- lib/pleroma/config/transfer_task.ex | 1 + lib/pleroma/config_db.ex | 26 +++++++++++++++++-- .../controllers/config_controller.ex | 5 +++- test/fixtures/config/static_search.secret.exs | 5 ++++ test/mix/tasks/pleroma/config_test.exs | 8 ++++++ test/pleroma/config/transfer_task_test.exs | 25 ++++++++++++++++++ test/pleroma/config_db_test.exs | 14 ++++++++++ .../controllers/config_controller_test.exs | 20 ++++++++++++-- 9 files changed, 105 insertions(+), 8 deletions(-) create mode 100644 test/fixtures/config/static_search.secret.exs diff --git a/lib/mix/tasks/pleroma/config.ex b/lib/mix/tasks/pleroma/config.ex index a10506236..81c41d967 100644 --- a/lib/mix/tasks/pleroma/config.ex +++ b/lib/mix/tasks/pleroma/config.ex @@ -335,9 +335,12 @@ defmodule Mix.Tasks.Pleroma.Config do group |> Pleroma.Config.Loader.filter_group(settings) |> Enum.each(fn {key, value} -> - {:ok, _} = ConfigDB.update_or_create(%{group: group, key: key, value: value}) - - shell_info("Settings for key #{key} migrated.") + if ConfigDB.static?(group, key) do + shell_info("Settings for key #{key} must remain in static configuration; skipping.") + else + {:ok, _} = ConfigDB.update_or_create(%{group: group, key: key, value: value}) + shell_info("Settings for key #{key} migrated.") + end end) shell_info("Settings for group #{inspect(group)} migrated.") diff --git a/lib/pleroma/config/transfer_task.ex b/lib/pleroma/config/transfer_task.ex index 3e59dc4e0..48cb158ca 100644 --- a/lib/pleroma/config/transfer_task.ex +++ b/lib/pleroma/config/transfer_task.ex @@ -47,6 +47,7 @@ defmodule Pleroma.Config.TransferTask do settings = (Repo.all(ConfigDB) ++ deleted_settings) + |> Enum.reject(&ConfigDB.static?/1) |> Enum.map(&merge_with_default/1) started_applications = Application.started_applications() diff --git a/lib/pleroma/config_db.ex b/lib/pleroma/config_db.ex index 2c3df773b..9aa52f7d6 100644 --- a/lib/pleroma/config_db.ex +++ b/lib/pleroma/config_db.ex @@ -23,6 +23,12 @@ defmodule Pleroma.ConfigDB do {:pleroma, :mrf_keyword, :replace} ] + @static_config [ + {:pleroma, Pleroma.Search}, + {:pleroma, Pleroma.Search.ParadeDB}, + {:pleroma, Pleroma.Search.ParadeDB.Repo} + ] + schema "config" do field(:key, Pleroma.EctoType.Config.Atom) field(:group, Pleroma.EctoType.Config.Atom) @@ -32,16 +38,32 @@ defmodule Pleroma.ConfigDB do timestamps() end - @spec get_all_as_keyword() :: keyword() - def get_all_as_keyword do + @spec get_all_as_keyword(keyword()) :: keyword() + def get_all_as_keyword(options \\ []) do ConfigDB |> select([c], {c.group, c.key, c.value}) |> Repo.all() + |> maybe_reject_static(options) |> Enum.reduce([], fn {group, key, value}, acc -> Keyword.update(acc, group, [{key, value}], &Keyword.merge(&1, [{key, value}])) end) end + def static?(%{group: group, key: key}), do: static?(group, key) + def static?(_config), do: false + + def static?(group, key) do + Enum.any?(@static_config, fn {static_group, static_key} -> + group in [static_group, inspect(static_group)] and key in [static_key, inspect(static_key)] + end) + end + + defp maybe_reject_static(configs, exclude_static: true) do + Enum.reject(configs, fn {group, key, _value} -> static?(group, key) end) + end + + defp maybe_reject_static(configs, _options), do: configs + @spec get_all_by_group(atom() | String.t()) :: [t()] def get_all_by_group(group) do from(c in ConfigDB, where: c.group == ^group) |> Repo.all() diff --git a/lib/pleroma/web/admin_api/controllers/config_controller.ex b/lib/pleroma/web/admin_api/controllers/config_controller.ex index 753690d45..4c76d808d 100644 --- a/lib/pleroma/web/admin_api/controllers/config_controller.ex +++ b/lib/pleroma/web/admin_api/controllers/config_controller.ex @@ -75,6 +75,7 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do Pleroma.Docs.JSON.compiled_descriptions() |> Enum.filter(&allowed_config?(&1, :whitelist)) |> Enum.reject(&allowed_config?(&1, :blacklist)) + |> Enum.reject(&ConfigDB.static?/1) json(conn, translate_descriptions(descriptions)) end @@ -92,7 +93,7 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do def show(conn, _params) do with :ok <- configurable_from_database() do - configs = ConfigDB.get_all_as_keyword() + configs = ConfigDB.get_all_as_keyword(exclude_static: true) merged = Config.Holder.default_config() @@ -123,6 +124,7 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do end) end) |> List.flatten() + |> Enum.reject(&ConfigDB.static?/1) render(conn, "index.json", %{ configs: merged, @@ -137,6 +139,7 @@ defmodule Pleroma.Web.AdminAPI.ConfigController do configs |> Enum.filter(&allowed_config?(&1, :whitelist)) |> Enum.reject(&allowed_config?(&1, :blacklist)) + |> Enum.reject(&(ConfigDB.static?(&1) and not &1[:delete])) |> Enum.map(fn %{group: group, key: key, delete: true} = params -> ConfigDB.delete(%{group: group, key: key, subkeys: params[:subkeys]}) diff --git a/test/fixtures/config/static_search.secret.exs b/test/fixtures/config/static_search.secret.exs new file mode 100644 index 000000000..13f744184 --- /dev/null +++ b/test/fixtures/config/static_search.secret.exs @@ -0,0 +1,5 @@ +import Config + +config :pleroma, Pleroma.Search, module: Pleroma.Search.ParadeDB +config :pleroma, Pleroma.Search.ParadeDB, url: "postgres://localhost/paradedb" +config :pleroma, Pleroma.Search.ParadeDB.Repo, pool_size: 2 diff --git a/test/mix/tasks/pleroma/config_test.exs b/test/mix/tasks/pleroma/config_test.exs index 7513e958e..620bf8662 100644 --- a/test/mix/tasks/pleroma/config_test.exs +++ b/test/mix/tasks/pleroma/config_test.exs @@ -87,6 +87,14 @@ defmodule Mix.Tasks.Pleroma.ConfigTest do config = ConfigDB.get_by_params(%{group: ":pleroma", key: ":first_setting"}) assert config.value == [key: "value", key2: [Repo]] end + + test "static search settings remain in the config file" do + MixTask.migrate_to_db("test/fixtures/config/static_search.secret.exs") + + refute ConfigDB.get_by_group_and_key(:pleroma, Pleroma.Search) + refute ConfigDB.get_by_group_and_key(:pleroma, Pleroma.Search.ParadeDB) + refute ConfigDB.get_by_group_and_key(:pleroma, Pleroma.Search.ParadeDB.Repo) + end end describe "with deletion of temp file" do diff --git a/test/pleroma/config/transfer_task_test.exs b/test/pleroma/config/transfer_task_test.exs index 6295fa888..ea5378ffd 100644 --- a/test/pleroma/config/transfer_task_test.exs +++ b/test/pleroma/config/transfer_task_test.exs @@ -71,6 +71,31 @@ defmodule Pleroma.Config.TransferTaskTest do assert assets_env[:mascots] == [a: 1, b: 2] end + test "does not transfer static search configuration" do + search_config = Application.get_env(:pleroma, Pleroma.Search) + parade_config = Application.get_env(:pleroma, Pleroma.Search.ParadeDB) + + clear_config(:database_config_blacklist, []) + + insert(:config, key: Pleroma.Search, value: [module: Pleroma.Search.ParadeDB]) + + insert(:config, + key: Pleroma.Search.ParadeDB, + value: [url: "postgres://localhost/paradedb"] + ) + + insert(:config, + key: Pleroma.Search.ParadeDB.Repo, + value: [pool_size: 50] + ) + + TransferTask.start_link([]) + + assert Application.get_env(:pleroma, Pleroma.Search) == search_config + assert Application.get_env(:pleroma, Pleroma.Search.ParadeDB) == parade_config + assert Application.get_env(:pleroma, Pleroma.Search.ParadeDB.Repo)[:pool_size] == 2 + end + describe "pleroma restart" do setup do on_exit(fn -> diff --git a/test/pleroma/config_db_test.exs b/test/pleroma/config_db_test.exs index 9d70e6d54..ae17a9d34 100644 --- a/test/pleroma/config_db_test.exs +++ b/test/pleroma/config_db_test.exs @@ -36,6 +36,20 @@ defmodule Pleroma.ConfigDBTest do assert config[:goose][:webhook_url] == "https://gander.com/" end + test "get_all_as_keyword/1 excludes static search configuration" do + insert(:config, key: Pleroma.Search, value: [module: Pleroma.Search.ParadeDB]) + insert(:config, key: Pleroma.Search.ParadeDB, value: [url: "postgres://example/db"]) + insert(:config, key: Pleroma.Search.ParadeDB.Repo, value: [pool_size: 50]) + insert(:config, key: :instance, value: [name: "Pleroma"]) + + config = ConfigDB.get_all_as_keyword(exclude_static: true) + + refute config[:pleroma][Pleroma.Search] + refute config[:pleroma][Pleroma.Search.ParadeDB] + refute config[:pleroma][Pleroma.Search.ParadeDB.Repo] + assert config[:pleroma][:instance] == [name: "Pleroma"] + end + describe "update_or_create/1" do test "common" do config = insert(:config) diff --git a/test/pleroma/web/admin_api/controllers/config_controller_test.exs b/test/pleroma/web/admin_api/controllers/config_controller_test.exs index c11ece10d..e4c8d866f 100644 --- a/test/pleroma/web/admin_api/controllers/config_controller_test.exs +++ b/test/pleroma/web/admin_api/controllers/config_controller_test.exs @@ -121,6 +121,19 @@ defmodule Pleroma.Web.AdminAPI.ConfigControllerTest do end) end + test "effective configuration omits static search settings", %{conn: conn} do + insert(:config, key: Pleroma.Search, value: [module: Pleroma.Search.ParadeDB]) + insert(:config, key: Pleroma.Search.ParadeDB, value: [url: "postgres://example/db"]) + + %{"configs" => configs} = + conn + |> get("/api/pleroma/admin/config") + |> json_response_and_validate_schema(200) + + refute Enum.any?(configs, &(&1["key"] == "Pleroma.Search")) + refute Enum.any?(configs, &(&1["key"] == "Pleroma.Search.ParadeDB")) + end + test "subkeys with full update right merge", %{conn: conn} do insert(:config, key: ":emoji", @@ -1552,12 +1565,15 @@ defmodule Pleroma.Web.AdminAPI.ConfigControllerTest do assert web_endpoint["children"] end - test "all keys from description are whitelisted", %{conn: conn} do + test "descriptions exclude blacklisted keys", %{conn: conn} do conn = get(conn, "/api/pleroma/admin/config/descriptions") assert response = json_response_and_validate_schema(conn, 200) - assert length(response) == length(Pleroma.Docs.JSON.compiled_descriptions()) + refute Enum.any?(response, &(&1["key"] == "Pleroma.Search")) + refute Enum.any?(response, &(&1["key"] == "Pleroma.Search.ParadeDB")) + + assert length(response) == length(Pleroma.Docs.JSON.compiled_descriptions()) - 2 end test "filters by database configuration blacklist and whitelist", %{conn: conn} do From 13c944a04b217a534cd9ef0d0f9d8dfd65609faf Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 12:03:24 +0400 Subject: [PATCH 15/18] fix(search): index remote attachment descriptions --- lib/pleroma/search.ex | 1 + test/pleroma/search_test.exs | 7 +++++++ 2 files changed, 8 insertions(+) diff --git a/lib/pleroma/search.ex b/lib/pleroma/search.ex index cd4907026..0d9c7c6e3 100644 --- a/lib/pleroma/search.ex +++ b/lib/pleroma/search.ex @@ -78,6 +78,7 @@ defmodule Pleroma.Search do defp attachment_names(attachments) when is_list(attachments) do Enum.map(attachments, fn + %{"summary" => summary} when is_binary(summary) -> summary %{"name" => name} -> name _ -> nil end) diff --git a/test/pleroma/search_test.exs b/test/pleroma/search_test.exs index 00d8b10ed..91521b1a5 100644 --- a/test/pleroma/search_test.exs +++ b/test/pleroma/search_test.exs @@ -83,6 +83,11 @@ defmodule Pleroma.SearchTest do "summary" => "subject text", "attachment" => [ %{"type" => "Document", "name" => "image description"}, + %{ + "type" => "Document", + "name" => "remote-image.jpg", + "summary" => "remote image description" + }, %{"type" => "Document", "name" => ""} ], "published" => DateTime.utc_now() |> DateTime.to_iso8601() @@ -93,6 +98,8 @@ defmodule Pleroma.SearchTest do assert content =~ "body text" assert content =~ "subject text" assert content =~ "image description" + assert content =~ "remote image description" + refute content =~ "remote-image.jpg" end test "object_to_search_data ignores objects with invalid published dates" do From f90a52398ba8825d70db9f4594904eca02e248b5 Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 12:16:04 +0400 Subject: [PATCH 16/18] ci(search): gate pinned ParadeDB integration --- .woodpecker/paradedb-integration.yaml | 64 ++++++++ .../search/parade_db_integration_test.exs | 145 +++++++++++++----- 2 files changed, 171 insertions(+), 38 deletions(-) create mode 100644 .woodpecker/paradedb-integration.yaml diff --git a/.woodpecker/paradedb-integration.yaml b/.woodpecker/paradedb-integration.yaml new file mode 100644 index 000000000..b91222837 --- /dev/null +++ b/.woodpecker/paradedb-integration.yaml @@ -0,0 +1,64 @@ +when: + - event: pull_request + path: + - ".woodpecker/**" + - "lib/pleroma/search/**" + - "test/pleroma/search/**" + - "mix.exs" + - "mix.lock" + evaluate: 'CI_COMMIT_SOURCE_BRANCH != "weblate"' + - event: push + branch: ${CI_REPO_DEFAULT_BRANCH} + path: + - ".woodpecker/**" + - "lib/pleroma/search/**" + - "test/pleroma/search/**" + - "mix.exs" + - "mix.lock" + +depends_on: + - lint + +labels: + platform: linux/amd64 + +variables: + script_file_entrypoint: &script_file_entrypoint + - /bin/sh + - -c + - 'printf "%s" "$CI_SCRIPT" | base64 -d > /tmp/ci-script.sh && /bin/sh -xe /tmp/ci-script.sh' + +steps: + paradedb-integration: + image: docker.io/elixir:1.19-otp-28-alpine + entrypoint: *script_file_entrypoint + environment: + MIX_ENV: test + VIX_COMPILATION_MODE: PLATFORM_PROVIDED_LIBVIPS + DB_HOST: postgres + DB_PORT: 5432 + PARADEDB_TEST_URL: postgres://postgres:postgres@paradedb:5432/paradedb + commands: + - apk add --no-cache build-base cmake exiftool ffmpeg vips-dev file-dev git openssl postgresql-client + - mix local.hex --force + - mix local.rebar --force + - mix deps.get + - timeout 60 sh -c 'until pg_isready -h postgres -U postgres -d pleroma_test; do sleep 1; done' || { echo "PostgreSQL did not become ready" >&2; exit 1; } + - timeout 60 sh -c 'until pg_isready -h paradedb -U postgres -d paradedb; do sleep 1; done' || { echo "ParadeDB did not become ready" >&2; exit 1; } + - mix ecto.create + - mix ecto.migrate + - mix test --include integration test/pleroma/search/parade_db_integration_test.exs + +services: + postgres: + image: docker.io/postgres:13-alpine + environment: + POSTGRES_DB: pleroma_test + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres + paradedb: + image: docker.io/paradedb/paradedb:0.24.3-pg15@sha256:997d62f6c2605becb7c45878af3f8c8c698f355006b675b3ce212d4eb2d14f2e + environment: + POSTGRES_DB: paradedb + POSTGRES_USER: postgres + POSTGRES_PASSWORD: postgres diff --git a/test/pleroma/search/parade_db_integration_test.exs b/test/pleroma/search/parade_db_integration_test.exs index 2687260cf..4336ecfc3 100644 --- a/test/pleroma/search/parade_db_integration_test.exs +++ b/test/pleroma/search/parade_db_integration_test.exs @@ -14,7 +14,10 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do alias Pleroma.Search.ParadeDB alias Pleroma.Web.CommonAPI - @image System.get_env("PARADEDB_TEST_IMAGE", "paradedb/paradedb:latest") + @image System.get_env( + "PARADEDB_TEST_IMAGE", + "paradedb/paradedb:0.24.3-pg15@sha256:997d62f6c2605becb7c45878af3f8c8c698f355006b675b3ce212d4eb2d14f2e" + ) @db_user "postgres" @db_password "postgres" @db_name "paradedb" @@ -26,6 +29,13 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do end setup_all do + case System.get_env("PARADEDB_TEST_URL") do + nil -> setup_docker_paradedb() + url -> setup_external_paradedb(url) + end + end + + defp setup_docker_paradedb do with :ok <- ensure_docker_available() do container = "pleroma-paradedb-it-#{System.unique_integer([:positive])}" @@ -58,49 +68,67 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do |> wait_for_port!(@docker_startup_timeout_ms) wait_for_pg_ready!(container, @docker_startup_timeout_ms) - wait_for_tcp_ready!(host_port, @docker_startup_timeout_ms) url = "postgres://#{@db_user}:#{@db_password}@127.0.0.1:#{host_port}/#{@db_name}" - - prev_env = System.get_env("PARADEDB_DATABASE_URL") - System.put_env("PARADEDB_DATABASE_URL", url) - - on_exit(fn -> - if is_nil(prev_env) do - System.delete_env("PARADEDB_DATABASE_URL") - else - System.put_env("PARADEDB_DATABASE_URL", prev_env) - end - end) - - {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 5}) - - previous_table = Pleroma.Config.get([Pleroma.Search.ParadeDB, :table], nil) - table = "pleroma_search_documents_it_#{System.unique_integer([:positive])}" - Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], table) - - on_exit(fn -> - if is_nil(previous_table) do - Pleroma.Config.delete([Pleroma.Search.ParadeDB, :table]) - else - Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], previous_table) - end - end) - - {:ok, - %{ - container: container, - container_id: container_id, - url: url, - table: table - }} + setup_repo(url, %{container: container, container_id: container_id}) end end - test "indexes and searches against real ParadeDB" do + defp setup_external_paradedb(url) do + uri = URI.parse(url) + wait_for_tcp_ready!(uri.host, uri.port || 5432, @docker_startup_timeout_ms) + setup_repo(url, %{}) + end + + defp setup_repo(url, context) do + prev_env = System.get_env("PARADEDB_DATABASE_URL") + System.put_env("PARADEDB_DATABASE_URL", url) + + on_exit(fn -> + if is_nil(prev_env) do + System.delete_env("PARADEDB_DATABASE_URL") + else + System.put_env("PARADEDB_DATABASE_URL", prev_env) + end + end) + + {:ok, _pid} = start_supervised({Pleroma.Search.ParadeDB.Repo, pool_size: 5}) + wait_for_repo_ready!(@docker_startup_timeout_ms) + + previous_table = Pleroma.Config.get([Pleroma.Search.ParadeDB, :table], nil) + suffix = Ecto.UUID.generate() |> String.replace("-", "") + table = "pleroma_search_documents_it_#{suffix}" + Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], table) + + on_exit(fn -> + if is_nil(previous_table) do + Pleroma.Config.delete([Pleroma.Search.ParadeDB, :table]) + else + Pleroma.Config.put([Pleroma.Search.ParadeDB, :table], previous_table) + end + end) + + {:ok, Map.merge(context, %{url: url, table: table})} + end + + test "indexes and searches against real ParadeDB", %{table: table} do + on_exit(fn -> + _ = + Ecto.Adapters.SQL.query( + Pleroma.Search.ParadeDB.Repo, + "DROP TABLE IF EXISTS #{table}" + ) + end) + :ok = ParadeDB.drop_index() assert :ok = ParadeDB.create_index() + assert {:ok, %{rows: [["0.24.3"]]}} = + Ecto.Adapters.SQL.query( + Pleroma.Search.ParadeDB.Repo, + "SELECT extversion FROM pg_extension WHERE extname = 'pg_search'" + ) + user = insert(:user) {:ok, a1} = @@ -118,6 +146,23 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do a2 = Activity.get_by_id_with_object(a2.id) private = Activity.get_by_id_with_object(private.id) + a1 = %{ + a1 + | object: %{ + a1.object + | data: + Map.merge(a1.object.data, %{ + "summary" => "paradedb subject phrase", + "attachment" => [ + %{ + "name" => "remote-image.jpg", + "summary" => "paradedb attachment description" + } + ] + }) + } + } + assert :ok = ParadeDB.add_to_index(a1) assert :ok = ParadeDB.add_to_index(a2) assert :ok = ParadeDB.add_to_index(private) @@ -130,6 +175,9 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do refute private.id in ids assert List.first(ids) == a2.id + assert Enum.any?(ParadeDB.search(nil, "subject phrase"), &(&1.id == a1.id)) + assert Enum.any?(ParadeDB.search(nil, "attachment description"), &(&1.id == a1.id)) + previous_fuzzy_distance = Pleroma.Config.get([Pleroma.Search.ParadeDB, :fuzzy_distance], 0) Pleroma.Config.put([Pleroma.Search.ParadeDB, :fuzzy_distance], 1) @@ -226,11 +274,11 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do wait.(wait) end - defp wait_for_tcp_ready!(port, timeout_ms) do + defp wait_for_tcp_ready!(host, port, timeout_ms) do deadline = System.monotonic_time(:millisecond) + timeout_ms wait = fn wait -> - case :gen_tcp.connect(~c"127.0.0.1", port, [:binary, active: false], 500) do + case :gen_tcp.connect(String.to_charlist(host), port, [:binary, active: false], 500) do {:ok, socket} -> :gen_tcp.close(socket) :ok @@ -248,6 +296,27 @@ defmodule Pleroma.Search.ParadeDBIntegrationTest do wait.(wait) end + defp wait_for_repo_ready!(timeout_ms) do + deadline = System.monotonic_time(:millisecond) + timeout_ms + + wait = fn wait -> + case Ecto.Adapters.SQL.query(Pleroma.Search.ParadeDB.Repo, "SELECT 1") do + {:ok, _} -> + :ok + + _ -> + if System.monotonic_time(:millisecond) > deadline do + raise "Timed out waiting for ParadeDB Repo" + else + :timer.sleep(250) + wait.(wait) + end + end + end + + wait.(wait) + end + defp docker(args) do System.cmd("docker", args, stderr_to_stdout: true) end From cece76e4cbce194a56847057f120550794c4f4fe Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 12:32:24 +0400 Subject: [PATCH 17/18] docs(search): document ParadeDB operations --- docs/configuration/search.md | 125 ++++++++++++++++++++++++++++++++--- 1 file changed, 117 insertions(+), 8 deletions(-) diff --git a/docs/configuration/search.md b/docs/configuration/search.md index 1d5056ddf..5820cdbd9 100644 --- a/docs/configuration/search.md +++ b/docs/configuration/search.md @@ -150,18 +150,22 @@ depends on the amount of text in posts. [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's integration suite tests `pg_search` 0.24.3 on PostgreSQL 15. Pleroma will maintain a small `pleroma_search_documents` table in that database (via the existing search indexing queue) and run search queries against it. The indexed text includes status content, content warnings, and attachment descriptions. -To use it, set the search module to `Pleroma.Search.ParadeDB` and configure the ParadeDB database URL: -These settings must be in static/runtime config (or `PARADEDB_DATABASE_URL`) so the dedicated ParadeDB Repo is started with the application. +### Configuration + +Set the search module to `Pleroma.Search.ParadeDB` and configure the ParadeDB database URL. These settings must be in static/runtime config +(or `PARADEDB_DATABASE_URL`) so the dedicated ParadeDB Repo is started with the application. They cannot be migrated to or changed through +ConfigDB. > 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", +> url: System.get_env("PARADEDB_DATABASE_URL"), > table: "pleroma_search_documents", > fuzzy_distance: 0 > @@ -169,22 +173,127 @@ These settings must be in static/runtime config (or `PARADEDB_DATABASE_URL`) so > pool_size: 2, > prepare: :unnamed -Then, create the table and BM25 index once: +When connecting over an untrusted network, enable certificate verification in the Repo configuration. For example: + +> config :pleroma, Pleroma.Search.ParadeDB.Repo, +> ssl: [ +> verify: :verify_peer, +> cacertfile: "/etc/pleroma/paradedb-ca.pem", +> server_name_indication: ~c"paradedb.example.com" +> ] + +### Initial rollout + +The `pg_search` extension must already be available in the database. Creating the extension normally requires an administrative database +role. The Pleroma runtime role requires `CONNECT` on the database, `USAGE` on the table's schema and the `pdb` schema, and `SELECT`, `INSERT`, `UPDATE`, +and `DELETE` on the generated table. For example, after creating the table as its owner: + +```sql +GRANT CONNECT ON DATABASE paradedb TO pleroma_search; +GRANT USAGE ON SCHEMA public, pdb TO pleroma_search; +GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE pleroma_search_documents TO pleroma_search; +``` + +Use this rollout order to prevent jobs from being handled by the previous backend: + +1. Install `pg_search` 0.24.3 on PostgreSQL 15 or another deliberately tested combination. +2. Stop Pleroma so no search indexing jobs are running. +3. Put the ParadeDB backend in static configuration. +4. Create the table and BM25 index in a one-shot offline process using `PARADEDB_DATABASE_URL` with the table owner/setup credential. +5. Apply the runtime grants shown above, set the service's `PARADEDB_DATABASE_URL` to the restricted runtime credential, and start Pleroma. +6. Start the initial backfill and monitor the `search_indexing` queue. + +Create the table and BM25 index once with: + +=== "OTP" + ```sh + PARADEDB_DATABASE_URL='OWNER_URL' PLEROMA_CTL_RPC_DISABLED=true \ + ./bin/pleroma_ctl search.indexer create_index + ``` === "From Source" ```sh - mix pleroma.search.indexer create_index + PARADEDB_DATABASE_URL='OWNER_URL' mix pleroma.search.indexer create_index ``` -After that, run an initial indexing pass to backfill existing posts: +The task currently prints an error without returning a failing process status. Verify that it prints `Index created` before continuing. + +### Backfill + +Run an initial indexing pass to enqueue existing posts: + +=== "OTP" + ```sh + ./bin/pleroma_ctl search.indexer index + ``` === "From Source" ```sh - mix pleroma.search.indexer index + PARADEDB_DATABASE_URL='RUNTIME_URL' mix pleroma.search.indexer index ``` +The command enqueues up to 100,000 Create activities by default. It uses keyset pagination, so new activities arriving during the backfill +do not shift the remaining pages. The defaults and accepted options are: + +* `--limit`: maximum activities to enqueue in this run; defaults to `100000` and must be positive +* `--step`: activities fetched from Pleroma's database per page; defaults to `1000` and must be positive +* `--chunk`: jobs inserted into Oban per batch; defaults to `100` and must be positive +* `--before`: resume before the checkpoint activity ID printed by an earlier run; the cursor is exclusive + +The checkpoint is the last activity enqueued, not confirmation that its job succeeded. Repeat the command with its last `--before` +checkpoint only when it reports that more activities remain. Inspect failed or discarded Oban jobs as well as waiting jobs before declaring +the backfill complete. + +The same operation can be run from an attached console while Pleroma is online: + +```elixir +Pleroma.Search.Backfill.run(limit: 10_000) +``` + +The returned `next_cursor` can be supplied as `before` when `exhausted` is `false`. The attached API also accepts `limit: :infinity`, but this +can create an unbounded number of Oban rows; bounded runs are safer on large instances. An `on_page` callback may be supplied for progress +reporting. + Note: Like the other external search backends, only public/unlisted Notes are indexed. Set `fuzzy_distance` to `1` or `2` to allow typo-tolerant matching (`2` is the maximum ParadeDB supports). -Indexing runs via the Oban `search_indexing` queue, so ensure that queue is enabled (not paused) in your Oban configuration. +Indexing runs via the Oban `search_indexing` queue. Its stock configuration starts paused so the search health monitor can control it; verify +that the queue is running before backfilling. A manually managed queue can be enabled with `search_indexing: 10` in the Oban `queues` +configuration instead of a `paused: true` entry. +Monitor that queue until the backfill jobs have drained before evaluating search completeness. + +### Rebuilding + +Stop Pleroma, or otherwise ensure no search indexing jobs are running, before rebuilding. While ParadeDB is still the configured backend, +run `drop_index`, then `create_index` with the table owner/setup credential. Reapply runtime grants after recreating the table, restart +Pleroma, and begin a new backfill. The DDL tasks print failures without returning a failing process status, so verify the exact `Index +dropped` and `Index created` success messages. Dropping the table removes only derived search data; Pleroma activities remain in the primary +database. + +### Rolling back + +To retain the ParadeDB table for a later retry, configure `Pleroma.Search.DatabaseSearch` (or the previous external backend) in static config +and restart Pleroma. + +Before switching backends, inspect the `search_indexing` queue. Drain or cancel pending backfill jobs, or explicitly accept that they will run +against the newly configured backend after restart. + +To remove the table during rollback, first stop Pleroma and run `drop_index` with the table owner/setup credential while ParadeDB is still +configured: + +=== "OTP" + ```sh + PARADEDB_DATABASE_URL='OWNER_URL' PLEROMA_CTL_RPC_DISABLED=true \ + ./bin/pleroma_ctl search.indexer drop_index + ``` + +=== "From Source" + ```sh + PARADEDB_DATABASE_URL='OWNER_URL' mix pleroma.search.indexer drop_index + ``` + +Verify that the task prints `Index dropped`. Then change the backend in static config and restart. The ParadeDB environment variable may be +removed after removing the ParadeDB stanza or switching to a backend that does not start its Repo. After switching away from ParadeDB, the +generic `drop_index` task targets the newly configured backend and will not remove the ParadeDB table; remove it directly with the database +owner instead. From 6e73d043ef87c4d0383b0e8d509a4a256f44f3eb Mon Sep 17 00:00:00 2001 From: Lain Soykaf Date: Sun, 26 Jul 2026 12:40:38 +0400 Subject: [PATCH 18/18] test(config): exclude static fixture from test loader --- .../config/{static_search.secret.exs => static_search.config} | 0 test/mix/tasks/pleroma/config_test.exs | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) rename test/fixtures/config/{static_search.secret.exs => static_search.config} (100%) diff --git a/test/fixtures/config/static_search.secret.exs b/test/fixtures/config/static_search.config similarity index 100% rename from test/fixtures/config/static_search.secret.exs rename to test/fixtures/config/static_search.config diff --git a/test/mix/tasks/pleroma/config_test.exs b/test/mix/tasks/pleroma/config_test.exs index 620bf8662..786d4aac2 100644 --- a/test/mix/tasks/pleroma/config_test.exs +++ b/test/mix/tasks/pleroma/config_test.exs @@ -89,7 +89,7 @@ defmodule Mix.Tasks.Pleroma.ConfigTest do end test "static search settings remain in the config file" do - MixTask.migrate_to_db("test/fixtures/config/static_search.secret.exs") + MixTask.migrate_to_db("test/fixtures/config/static_search.config") refute ConfigDB.get_by_group_and_key(:pleroma, Pleroma.Search) refute ConfigDB.get_by_group_and_key(:pleroma, Pleroma.Search.ParadeDB)