Packages
fnord
0.8.76
0.9.40
0.9.39
0.9.38
0.9.37
0.9.36
0.9.35
0.9.34
0.9.33
0.9.32
0.9.31
0.9.30
0.9.29
0.9.28
0.9.27
0.9.26
0.9.25
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.14
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.99
0.8.98
0.8.97
0.8.96
0.8.95
0.8.94
0.8.93
0.8.92
0.8.91
0.8.90
0.8.89
0.8.88
0.8.87
0.8.86
0.8.85
0.8.84
0.8.83
0.8.82
0.8.81
0.8.80
0.8.79
0.8.78
0.8.77
0.8.76
0.8.75
0.8.74
0.8.73
0.8.72
0.8.71
0.8.70
0.8.69
0.8.68
0.8.67
0.8.66
0.8.65
0.8.64
0.8.63
0.8.62
0.8.61
0.8.60
0.8.59
0.8.58
0.8.57
0.8.56
0.8.55
0.8.54
0.8.53
0.8.52
0.8.51
0.8.50
0.8.49
0.8.48
0.8.47
0.8.46
0.8.45
0.8.44
0.8.43
0.8.42
0.8.41
0.8.40
0.8.39
0.8.38
0.8.37
0.8.36
0.8.35
0.8.34
0.8.33
0.8.32
0.8.31
0.8.30
0.8.29
0.8.27
0.8.26
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.1
0.8.0
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.3
0.7.2
0.7.1
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.44
0.4.43
0.4.42
0.4.41
0.4.40
0.4.39
0.4.38
0.4.37
0.4.36
0.4.35
0.4.34
0.4.33
0.4.32
0.4.30
0.4.29
0.4.28
0.4.27
0.4.26
0.4.25
0.4.24
0.4.23
0.4.22
0.4.21
0.4.20
0.4.19
0.4.18
0.4.17
0.4.16
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.0
0.1.0
AI code archaeology
Current section
Files
Jump to
Current section
Files
lib/services/conversation_indexer.ex
defmodule Services.ConversationIndexer do
@moduledoc """
Background indexer for conversations.
This GenServer mirrors `Services.BackgroundIndexer`, but operates on
conversations instead of file entries. It processes one conversation at a
time, generating embeddings from the conversation messages JSON and writing
them to the conversation index.
"""
use GenServer, restart: :temporary
@type state :: %{
project: Store.Project.t() | nil,
impl: module(),
convo_queue: [Store.Project.Conversation.t()],
task: pid() | nil,
mon_ref: reference() | nil
}
@spec start_link(keyword()) :: GenServer.on_start()
def start_link(opts \\ []) do
GenServer.start_link(__MODULE__, opts)
end
@spec stop(pid() | any()) :: :ok
def stop(pid) when is_pid(pid) do
try do
GenServer.stop(pid, :normal, 5_000)
catch
:exit, _reason -> :ok
end
end
def stop(_), do: :ok
@impl true
def init(opts) do
# Use the dedicated AI indexer HTTP pool
HttpPool.set(:ai_indexer)
project =
case Keyword.get(opts, :project) do
%Store.Project{} = prj ->
prj
_ ->
case Store.get_project() do
{:ok, prj} -> prj
_ -> nil
end
end
convo_queue =
case Keyword.get(opts, :conversations) do
list when is_list(list) -> list
_ -> []
end
state = %{
project: project,
impl: Indexer.impl(),
convo_queue: convo_queue,
task: nil,
mon_ref: nil
}
{:ok, state, {:continue, :process_next}}
end
@impl true
def handle_continue(:process_next, %{task: pid} = state) when is_pid(pid) do
{:noreply, state}
end
@impl true
def handle_continue(:process_next, %{task: nil, convo_queue: [convo | rest]} = state) do
{:ok, task_pid} = Task.start_link(fn -> safe_process(convo, state.impl, state.project) end)
mon_ref = Process.monitor(task_pid)
new_state = %{state | task: task_pid, mon_ref: mon_ref, convo_queue: rest}
{:noreply, new_state}
end
@impl true
def handle_continue(:process_next, %{task: nil, convo_queue: [], project: project} = state)
when not is_nil(project) do
case next_stale_conversation(project) do
nil ->
{:stop, :normal, state}
convo ->
{:ok, task_pid} =
Task.start_link(fn -> safe_process(convo, state.impl, state.project) end)
mon_ref = Process.monitor(task_pid)
new_state = %{state | task: task_pid, mon_ref: mon_ref}
{:noreply, new_state}
end
end
@impl true
def handle_continue(:process_next, %{task: nil, convo_queue: [], project: nil} = state) do
{:stop, :normal, state}
end
@impl true
def handle_info({:DOWN, ref, :process, pid, _reason}, %{mon_ref: ref, task: pid} = state) do
new_state = %{state | task: nil, mon_ref: nil}
{:noreply, new_state, {:continue, :process_next}}
end
@impl true
def handle_info(_msg, state), do: {:noreply, state}
@impl true
def terminate(_reason, state) do
case state do
%{task: pid} when is_pid(pid) -> Process.exit(pid, :kill)
_ -> :ok
end
HttpPool.clear()
:ok
end
# ---------------------------------------------------------------------------
# Per-conversation processing
# ---------------------------------------------------------------------------
@spec safe_process(Store.Project.Conversation.t(), module(), Store.Project.t() | nil) :: :ok
defp safe_process(convo, impl, project) do
try do
case Store.Project.Conversation.read(convo) do
{:ok, ts, messages, metadata} ->
json = Jason.encode!(%{"messages" => messages})
with {:ok, embeddings} <- impl.get_embeddings(json) do
meta =
metadata
|> Map.merge(%{
"conversation_id" => convo.id,
"last_indexed_ts" => DateTime.to_unix(ts),
"message_count" => length(messages)
})
case project do
%Store.Project{} = prj ->
:ok =
Store.Project.ConversationIndex.write_embeddings(
prj,
convo.id,
embeddings,
meta
)
detail = convo.id
UI.end_step("Reindexed conversation", detail)
_ ->
:ok
end
end
_ ->
:ok
end
rescue
_ -> :ok
end
end
@spec next_stale_conversation(Store.Project.t()) :: Store.Project.Conversation.t() | nil
defp next_stale_conversation(project) do
project
|> Store.Project.ConversationIndex.index_status()
|> Map.get(:stale, [])
|> List.first()
end
end