Merge pull request 'Add ParadeDB search backend' (#7849) from paradedb-search-backend into develop

Reviewed-on: https://git.pleroma.social/pleroma/pleroma/pulls/7849
This commit is contained in:
lain 2026-07-26 09:28:06 +00:00
commit 2ce6e5f49d
28 changed files with 1717 additions and 93 deletions

View file

@ -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

View file

@ -0,0 +1 @@
Add ParadeDB (pg_search) as an optional external search backend, indexing status content, content warnings, and attachment descriptions.

View file

@ -944,6 +944,15 @@ 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",
fuzzy_distance: 0
config :pleroma, Pleroma.Search.ParadeDB.Repo,
pool_size: 2,
prepare: :unnamed
config :pleroma, Pleroma.Application,
background_migrators: true,
internal_fetch: true,

View file

@ -3536,7 +3536,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
]
}
]
},
@ -3569,6 +3574,34 @@ 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"]
},
%{
key: :fuzzy_distance,
type: :integer,
description: "Edit-distance used for fuzzy token matching (0 disables fuzziness, max 2).",
suggestion: [0, 1, 2]
}
]
},
%{
group: :pleroma,
key: Pleroma.Language.LanguageDetector,

View file

@ -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).
@ -145,3 +145,155 @@ 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'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.
### 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"),
> table: "pleroma_search_documents",
> fuzzy_distance: 0
>
> config :pleroma, Pleroma.Search.ParadeDB.Repo,
> pool_size: 2,
> prepare: :unnamed
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
PARADEDB_DATABASE_URL='OWNER_URL' mix pleroma.search.indexer create_index
```
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
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. 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.

View file

@ -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 Pleroma.Config.get([Pleroma.Search, :module]) do
Pleroma.Search.ParadeDB ->
[Pleroma.Search.ParadeDB.Repo]
_ ->
[]
end
end
end

View file

@ -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.")

View file

@ -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

View file

@ -81,12 +81,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) ++
[
@ -198,6 +201,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))

View file

@ -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()

View file

@ -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()

View file

@ -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,60 @@ 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
%{"summary" => summary} when is_binary(summary) -> summary
%{"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

View file

@ -0,0 +1,137 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2026 Pleroma Authors <https://pleroma.social/>
# 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

View file

@ -0,0 +1,287 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDB do
@behaviour Pleroma.Search.SearchBackend
require Logger
require Pleroma.Constants
import Ecto.Query
import Pleroma.Search.DatabaseSearch,
only: [
maybe_fetch: 3,
maybe_restrict_author: 2,
maybe_restrict_blocked: 2,
maybe_restrict_local: 2
]
alias Pleroma.Activity
alias Pleroma.Config.Getting, as: Config
alias Pleroma.Search
@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
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", []),
{: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
case client_impl().query("DROP TABLE IF EXISTS #{table()}", []) do
{:ok, _} ->
:ok
{: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
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)
if is_nil(dumped_activity_id) do
Logger.error("ParadeDB add_to_index failed: invalid activity id #{inspect(activity.id)}")
:ok
else
sql =
"""
INSERT INTO #{table()} (id, object_id, object_ap_id, actor_ap_id, content, published_at)
VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (id) DO UPDATE SET
object_id = EXCLUDED.object_id,
object_ap_id = EXCLUDED.object_ap_id,
actor_ap_id = EXCLUDED.actor_ap_id,
content = EXCLUDED.content,
published_at = EXCLUDED.published_at
"""
params = [
dumped_activity_id,
maybe_search_data.id,
maybe_search_data.ap,
actor_ap_id,
maybe_search_data.content,
published_at
]
case client_impl().query(sql, params) do
{:ok, _} ->
:ok
{:error, error} ->
Logger.error("ParadeDB add_to_index failed: #{inspect(error)}")
{:error, error}
end
end
else
_ -> :ok
end
end
@impl true
def remove_from_index(object) do
sql = "DELETE FROM #{table()} WHERE object_id = $1"
case client_impl().query(sql, [object.id]) do
{:ok, _} ->
:ok
{:error, error} ->
Logger.error("ParadeDB remove_from_index failed: #{inspect(error)}")
{:error, error}
end
end
@impl true
def search(user, query, options \\ []) do
limit = options |> Keyword.get(:limit, @max_limit) |> min(@max_limit)
offset = Keyword.get(options, :offset, 0)
author = Keyword.get(options, :author)
case search_ids(query, limit, offset, author) do
{:ok, []} ->
maybe_fetch([], user, query)
{:ok, ids} ->
ids
|> fetch_activities(user, query, author)
{:error, error} ->
Logger.error("ParadeDB search failed: #{inspect(error)}")
maybe_fetch([], user, query)
end
end
defp fetch_activities(ids, user, query, author) do
dumped_ids =
ids
|> Enum.map(&dump_activity_id/1)
|> Enum.reject(&is_nil/1)
from(a in Activity, where: a.id in ^dumped_ids)
|> Activity.with_preloaded_object()
|> Activity.restrict_deactivated_users()
|> maybe_restrict_public(user)
|> 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 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 &&& #{search_query_param_sql(fuzzy_distance())}"
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, id 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 search_query_param_sql(0), do: "$1"
defp search_query_param_sql(distance), do: "($1::text)::pdb.fuzzy(#{distance})"
defp create_table_sql(table) do
"""
CREATE TABLE IF NOT EXISTS #{table} (
id uuid PRIMARY KEY,
object_id bigint NOT NULL,
object_ap_id text NOT NULL,
actor_ap_id text,
content text NOT NULL,
published_at timestamptz
)
"""
end
defp create_object_id_index_sql(table) do
"CREATE INDEX IF NOT EXISTS #{table}_object_id_idx ON #{table} (object_id)"
end
defp create_bm25_index_sql(table) do
"""
CREATE INDEX IF NOT EXISTS #{table}_bm25_idx ON #{table}
USING bm25 (id, content, (actor_ap_id::pdb.literal), published_at)
WITH (key_field = 'id')
"""
end
defp dump_activity_id(<<_::binary-size(16)>> = uuid), do: uuid
defp dump_activity_id(id) when is_binary(id) do
case 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

View file

@ -0,0 +1,24 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDB.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

View file

@ -0,0 +1,24 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDB.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

View file

@ -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]})

View file

@ -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

View file

@ -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.config")
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

View file

@ -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 ->

View file

@ -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)

View file

@ -0,0 +1,79 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2026 Pleroma Authors <https://pleroma.social/>
# 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

View file

@ -0,0 +1,323 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDBIntegrationTest do
use Pleroma.DataCase, async: false
@moduletag :integration
@moduletag timeout: 180_000
import Pleroma.Factory
alias Pleroma.Activity
alias Pleroma.Search.ParadeDB
alias Pleroma.Web.CommonAPI
@image System.get_env(
"PARADEDB_TEST_IMAGE",
"paradedb/paradedb:0.24.3-pg15@sha256:997d62f6c2605becb7c45878af3f8c8c698f355006b675b3ce212d4eb2d14f2e"
)
@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
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])}"
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",
"127.0.0.1::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}"
setup_repo(url, %{container: container, container_id: container_id})
end
end
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} =
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)
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)
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
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)
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)
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 wait_for_tcp_ready!(host, port, timeout_ms) do
deadline = System.monotonic_time(:millisecond) + timeout_ms
wait = fn wait ->
case :gen_tcp.connect(String.to_charlist(host), 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 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
end

View file

@ -0,0 +1,286 @@
# Pleroma: A lightweight social networking server
# Copyright © 2017-2022 Pleroma Authors <https://pleroma.social/>
# SPDX-License-Identifier: AGPL-3.0-only
defmodule Pleroma.Search.ParadeDBTest do
use Pleroma.DataCase, async: true
use Oban.Testing, repo: Pleroma.Repo
import Mox
import Pleroma.Factory
alias Pleroma.Search.ParadeDB
alias Pleroma.Search.ParadeDB.ClientMock
alias Pleroma.UnstubbedConfigMock, as: Config
alias Pleroma.Web.CommonAPI
alias Pleroma.Workers.SearchIndexingWorker
describe "ParadeDB" do
test "indexes a public post on creation" do
user = insert(:user)
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: "public"
})
args = %{"op" => "add_to_index", "activity" => activity.id}
assert_enqueued(worker: SearchIndexingWorker, args: args)
ClientMock
|> expect(:query, fn sql, params ->
assert sql =~ "INSERT INTO pleroma_search_documents"
[activity_id, object_id, object_ap_id, actor_ap_id, content, published_at] = params
{:ok, dumped_activity_id} = FlakeId.Ecto.CompatType.dump(activity.id)
assert activity_id == dumped_activity_id
assert is_integer(object_id)
assert is_binary(object_ap_id)
assert actor_ap_id == activity.data["actor"]
assert content == "guys i just don&#39;t wanna leave the swamp"
assert %DateTime{} = published_at
send(self(), :inserted)
{:ok, %{}}
end)
Config
|> expect(:get, 3, fn
[Pleroma.Search, :module], nil ->
ParadeDB
[Pleroma.Search.ParadeDB, :client_impl], nil ->
ClientMock
[Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" ->
"pleroma_search_documents"
end)
assert :ok = perform_job(SearchIndexingWorker, args)
assert_received(:inserted)
end
test "doesn't index posts that are not public" do
user = insert(:user)
Enum.each(["private", "direct"], fn visibility ->
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: visibility
})
args = %{"op" => "add_to_index", "activity" => activity.id}
refute_enqueued(worker: SearchIndexingWorker, args: args)
end)
end
test "backend skips non-public posts when called directly" 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
})
assert :ok = ParadeDB.add_to_index(activity)
end)
end
test "deletes posts from index when deleted locally" do
user = insert(:user)
ClientMock
|> expect(:query, 2, fn sql, params ->
cond do
String.contains?(sql, "INSERT INTO pleroma_search_documents") ->
send(self(), :inserted)
{:ok, %{}}
String.contains?(sql, "DELETE FROM pleroma_search_documents") ->
assert [object_id] = params
assert is_integer(object_id)
send(self(), :deleted)
{:ok, %{}}
true ->
flunk("Unexpected ParadeDB SQL: #{inspect(sql)}")
end
end)
Config
|> expect(:get, 6, fn
[Pleroma.Search, :module], nil ->
ParadeDB
[Pleroma.Search.ParadeDB, :client_impl], nil ->
ClientMock
[Pleroma.Search.ParadeDB, :table], "pleroma_search_documents" ->
"pleroma_search_documents"
end)
{:ok, activity} =
CommonAPI.post(user, %{
status: "guys i just don't wanna leave the swamp",
visibility: "public"
})
args = %{"op" => "add_to_index", "activity" => activity.id}
assert_enqueued(worker: SearchIndexingWorker, args: args)
assert :ok = perform_job(SearchIndexingWorker, args)
assert_received(:inserted)
{:ok, _} = CommonAPI.delete(activity.id, user)
delete_args = %{"op" => "remove_from_index", "object" => activity.object.id}
assert_enqueued(worker: SearchIndexingWorker, args: delete_args)
assert :ok = perform_job(SearchIndexingWorker, delete_args)
assert_received(:deleted)
end
test "search returns activities in backend order" do
user = insert(:user)
{:ok, activity1} = CommonAPI.post(user, %{status: "first swamp", visibility: "public"})
{:ok, activity2} = CommonAPI.post(user, %{status: "second swamp", visibility: "public"})
activity1_id = activity1.id
activity2_id = activity2.id
{:ok, dumped_activity1_id} = FlakeId.Ecto.CompatType.dump(activity1_id)
{:ok, dumped_activity2_id} = FlakeId.Ecto.CompatType.dump(activity2_id)
ClientMock
|> expect(:query, fn sql, params ->
assert sql =~ "SELECT id FROM pleroma_search_documents"
assert ["swamp", _limit, _offset] = params
{:ok, %{rows: [[dumped_activity2_id], [dumped_activity1_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 ->
0
end)
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, 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)
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, 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::text)::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

View file

@ -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,83 @@ 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" => "remote-image.jpg",
"summary" => "remote 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"
assert content =~ "remote image description"
refute content =~ "remote-image.jpg"
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

View file

@ -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

View file

@ -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)

View file

@ -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, [])