Merge remote-tracking branch 'origin/develop' into pr-7444-finish
This commit is contained in:
Lain Soykaf
commit
842e6bcaff
28 changed files with 1717 additions and 93 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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.")
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
137
lib/pleroma/search/backfill.ex
Normal file
137
lib/pleroma/search/backfill.ex
Normal 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
|
||||
287
lib/pleroma/search/parade_db.ex
Normal file
287
lib/pleroma/search/parade_db.ex
Normal 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
|
||||
24
lib/pleroma/search/parade_db/client.ex
Normal file
24
lib/pleroma/search/parade_db/client.ex
Normal 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
|
||||
24
lib/pleroma/search/parade_db/repo.ex
Normal file
24
lib/pleroma/search/parade_db/repo.ex
Normal 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
|
||||
|
|
@ -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]})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue