bikoret/vav_pipeline/lib/vav_pipeline/l1_index.ex

defmodule VavPipeline.L1Index do
  @moduledoc """
  Inverted index over the edut: consonantal word → the verses that carry it.

  *V'darashta v'chakarta v'sha'alta heitev* — you shall inquire, search and
  ask diligently (Devarim 13:15). A question that quotes Torah names its
  own verse; this index is how the pipeline hears which one.

  Built once from `l1/torah/<book>/<chapter>/<verse>.json` right after the
  process starts (`handle_continue`), so the first pe'ah request does not
  pay for it. Read-only after that — L1 is engraved and so is the index.
  Callers block until the build finishes; no call ever sees a half-built
  index.

  Verse ids are `{book_order, chapter, verse}` so that `Enum.sort/1` is
  Torah order.
  """

  use GenServer
  require Logger

  alias VavPipeline.L1Reader

  @books ~w(bereishit shemot vayikra bamidbar devarim)
  @book_order @books |> Enum.with_index() |> Map.new()

  @type verse_id :: {non_neg_integer(), pos_integer(), pos_integer()}

  def start_link(opts \\ []) do
    GenServer.start_link(__MODULE__, opts, name: Keyword.get(opts, :name, __MODULE__))
  end

  @doc "Canonical book order, as the index sorts it."
  @spec books() :: [String.t()]
  def books, do: @books

  @doc "The verses that contain this consonantal word. Empty list if none."
  @spec lookup(String.t(), GenServer.server()) :: [verse_id()]
  def lookup(consonants, server \\ __MODULE__) do
    GenServer.call(server, {:lookup, consonants}, 60_000)
  end

  @doc "The verses whose words L1 files under this shoresh. Empty list if none."
  @spec lookup_root(String.t(), GenServer.server()) :: [verse_id()]
  def lookup_root(shoresh, server \\ __MODULE__) do
    GenServer.call(server, {:lookup_root, shoresh}, 60_000)
  end

  @doc "`{verses_indexed, words_indexed}` — for health and for tests."
  @spec stats(GenServer.server()) :: {non_neg_integer(), non_neg_integer()}
  def stats(server \\ __MODULE__), do: GenServer.call(server, :stats, 60_000)

  @doc "Turn an index verse id back into a scope map."
  @spec to_scope(verse_id()) :: %{book: String.t(), chapter: pos_integer(), verse: pos_integer()}
  def to_scope({order, chapter, verse}) do
    %{book: Enum.at(@books, order), chapter: chapter, verse: verse}
  end

  # ------------------------------------------------------------
  # GenServer
  # ------------------------------------------------------------

  @impl true
  def init(opts) do
    root = Keyword.get(opts, :l1_root, L1Reader.l1_root())
    {:ok, %{root: root, words: %{}, roots: %{}, verse_count: 0}, {:continue, :build}}
  end

  @impl true
  def handle_continue(:build, state) do
    {time_us, built} = :timer.tc(fn -> build(state.root) end)

    case built do
      {:ok, words, roots, verse_count} ->
        Logger.info(
          "L1Index: #{verse_count} verses, #{map_size(words)} distinct words, " <>
            "#{map_size(roots)} roots, #{div(time_us, 1000)}ms"
        )

        {:noreply, %{state | words: words, roots: roots, verse_count: verse_count}}

      {:error, reason} ->
        # No L1 → no index. Resolution falls back to the default scope and
        # says so; the health check already reports l1_hash separately.
        Logger.warning("L1Index: not built — #{reason}")
        {:noreply, state}
    end
  end

  @impl true
  def handle_call({:lookup, consonants}, _from, state) do
    {:reply, Map.get(state.words, consonants, []), state}
  end

  def handle_call({:lookup_root, shoresh}, _from, state) do
    {:reply, Map.get(state.roots, shoresh, []), state}
  end

  def handle_call(:stats, _from, state) do
    {:reply, {state.verse_count, map_size(state.words)}, state}
  end

  # ------------------------------------------------------------
  # Build
  # ------------------------------------------------------------

  defp build(root) do
    torah = Path.join(root, "torah")

    if File.dir?(torah) do
      {words, roots, count} =
        Enum.reduce(@books, {%{}, %{}, 0}, fn book, acc ->
          index_book(Path.join(torah, book), @book_order[book], acc)
        end)

      if count == 0, do: {:error, "no verses under #{torah}"}, else: {:ok, words, roots, count}
    else
      {:error, "#{torah} is not a directory"}
    end
  end

  defp index_book(book_dir, order, acc) do
    book_dir
    |> list_numeric()
    |> Enum.reduce(acc, fn chapter, acc ->
      chapter_dir = Path.join(book_dir, to_string(chapter))

      chapter_dir
      |> list_numeric(".json")
      |> Enum.reduce(acc, fn verse, {words, roots, count} ->
        case read_consonants(Path.join(chapter_dir, "#{verse}.json")) do
          {:ok, consonant_words, shorashim} ->
            id = {order, chapter, verse}

            words =
              consonant_words
              |> Enum.uniq()
              |> Enum.reduce(words, fn w, words -> Map.update(words, w, [id], &[id | &1]) end)

            roots =
              shorashim
              |> Enum.uniq()
              |> Enum.reduce(roots, fn r, roots -> Map.update(roots, r, [id], &[id | &1]) end)

            {words, roots, count + 1}

          :error ->
            {words, roots, count}
        end
      end)
    end)
  end

  # Directory entries that are integers (optionally with a suffix), ascending.
  defp list_numeric(dir, suffix \\ "") do
    case File.ls(dir) do
      {:ok, entries} ->
        entries
        |> Enum.flat_map(fn e ->
          case Integer.parse(String.replace_suffix(e, suffix, "")) do
            {n, ""} -> [n]
            _ -> []
          end
        end)
        |> Enum.sort()

      {:error, _} ->
        []
    end
  end

  defp read_consonants(path) do
    with {:ok, contents} <- File.read(path),
         {:ok, %{"words" => words}} when is_list(words) <- Jason.decode(contents) do
      {:ok, Enum.flat_map(words, &word_consonants/1),
       for(%{"shoresh" => sh} when is_binary(sh) <- words, do: sh)}
    else
      _ -> :error
    end
  end

  # A word joined by maqaf is indexed as each half, so that a question
  # written with spaces still finds it.
  defp word_consonants(%{"consonants" => c}) when is_binary(c),
    do: String.split(c, "־", trim: true)

  defp word_consonants(_), do: []
end