bikoret/vav_pipeline/lib/vav_pipeline/azazel_ingest.ex

defmodule VavPipeline.AzazelIngest do
  @moduledoc """
  Cross-language bridge between the Python Azazel Processor and the
  Elixir L3 layer.

  The Python orchestrator (azazel_processor/src/azazel_processor/
  azazel.py) serializes completed AzazelRecords to JSON files in a
  pending-l3 directory. This GenServer watches that directory,
  reads each file, calls L3.Store.record_disposal/1, and on success
  shreds the source file.

  The pattern matches Shorashim's other cross-language file bridges
  (the manna provision file). Recoverable: if either the Python side
  or this GenServer crashes mid-write, the file remains and is
  re-processed on the next poll.

  POLL-BASED, NOT FileSystem.event-BASED:

  We do not bring the :file_system dependency into vav_pipeline.
  Polling is simpler and adequate for the disposal cadence (the
  Azazel channel is for exceptional content, not a high-throughput
  stream). Default poll interval 5 seconds; configurable.

  RETRY SEMANTICS:

  If L3.Store.record_disposal/1 fails (validation error, DB
  unavailable), the file is LEFT IN PLACE — it will be retried on
  the next poll. The reasoning: L3 is append-only; a write that
  failed for a transient reason should be retried, and a write
  that failed for a permanent reason (schema mismatch) should
  remain visible to operators rather than be silently destroyed.

  Operator intervention path: if a file persists across many polls
  with the same error, the operator inspects it directly in the
  pending directory and decides whether to fix the schema, fix the
  record, or move it to an error-archive subdirectory for analysis.

  PROPERTY COVERAGE:

  Property 9 (operator-internal half) — the L3 record exists. The
  public-side endpoint (commit 8) reads from the same table.
  """

  use GenServer

  require Logger

  alias VavPipeline.{Chatzotzrot, L3.Store}

  @default_poll_interval_ms 5_000
  @default_pending_dir "/tmp/shorashim/azazel/pending_l3"

  # ============================================================
  # Client API
  # ============================================================

  @doc """
  Start the AzazelIngest GenServer. Options:
    - :pending_dir       — directory to watch (default
                           /tmp/shorashim/azazel/pending_l3)
    - :poll_interval_ms  — milliseconds between polls (default 5000)
    - :name              — process name (default __MODULE__)
    - :recorder          — function used to write the L3 record;
                           defaults to &Store.record_disposal/1.
                           Tests inject mocks here so they don't
                           need PostgreSQL.
  """
  def start_link(opts \\ []) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  @doc """
  Run a single poll synchronously. Useful for tests and for
  operator-triggered manual flushes.
  """
  def poll_now(server \\ __MODULE__) do
    GenServer.call(server, :poll_now)
  end

  @doc "Return the GenServer's current configuration and state."
  def state(server \\ __MODULE__) do
    GenServer.call(server, :state)
  end

  # ============================================================
  # GenServer
  # ============================================================

  @impl true
  def init(opts) do
    state = %{
      pending_dir: Keyword.get(opts, :pending_dir, @default_pending_dir),
      poll_interval_ms: Keyword.get(opts, :poll_interval_ms, @default_poll_interval_ms),
      recorder: Keyword.get(opts, :recorder, &Store.record_disposal/1),
      polls_completed: 0,
      records_ingested: 0,
      errors: 0
    }

    File.mkdir_p!(state.pending_dir)
    schedule_poll(state.poll_interval_ms)
    {:ok, state}
  end

  @impl true
  def handle_call(:poll_now, _from, state) do
    new_state = do_poll(state)
    {:reply, :ok, new_state}
  end

  def handle_call(:state, _from, state), do: {:reply, state, state}

  @impl true
  def handle_info(:poll, state) do
    new_state = do_poll(state)
    schedule_poll(new_state.poll_interval_ms)
    {:noreply, new_state}
  end

  # ============================================================
  # Internal
  # ============================================================

  defp schedule_poll(interval_ms) do
    Process.send_after(self(), :poll, interval_ms)
  end

  defp do_poll(state) do
    files = list_pending(state.pending_dir)

    {ingested, errors} =
      Enum.reduce(files, {0, 0}, fn path, {i, e} ->
        case ingest_file(path, state.recorder) do
          :ok -> {i + 1, e}
          {:error, _reason} -> {i, e + 1}
        end
      end)

    %{
      state
      | polls_completed: state.polls_completed + 1,
        records_ingested: state.records_ingested + ingested,
        errors: state.errors + errors
    }
  end

  defp list_pending(dir) do
    case File.ls(dir) do
      {:ok, names} ->
        names
        |> Enum.filter(&String.ends_with?(&1, ".json"))
        |> Enum.sort()
        |> Enum.map(&Path.join(dir, &1))

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

  defp ingest_file(path, recorder) do
    with {:ok, body} <- File.read(path),
         {:ok, raw} <- Jason.decode(body),
         {:ok, attrs} <- normalize_record(raw),
         {:ok, _record} <- recorder.(attrs) do
      # Property 6 (out-of-band acknowledgment) — fire the Chatzotzrot
      # signal AFTER the L3 write succeeds. The chatzotzrot is the
      # side channel; the disposal flow itself does not return
      # acknowledgment to the caller (the caller already knows they
      # handed the input to disposal).
      #
      # :moadim is the right signal type — Yom Kippur completion
      # semantics. This disposal cycle is closed; the public fact
      # (property 9) is now witnessable via /peah/azazel_log.
      #
      # Process.whereis-guarded so this works in tests that don't
      # bring up the full supervision tree.
      fire_chatzotzrot(attrs)

      # Property 10 follow-through: shred the source file after
      # successful L3 write. The disposal is now permanent oral
      # tradition; the pending artifact must not linger.
      File.rm(path)
      :ok
    else
      {:error, reason} ->
        Logger.warning(
          "AzazelIngest: failed to ingest #{path} — #{inspect(reason)}. " <>
            "File left in place for retry."
        )

        {:error, reason}
    end
  end

  defp fire_chatzotzrot(attrs) do
    if Process.whereis(Chatzotzrot) do
      summary = chatzotzrot_summary(attrs)

      Chatzotzrot.signal(:moadim,
        reason: "Azazel disposal completed: #{summary}"
      )
    end

    :ok
  end

  defp chatzotzrot_summary(attrs) do
    case_ref = Map.get(attrs, :case_ref) || "unknown"
    shaliach = Map.get(attrs, :shaliach) || "unknown"
    vidui = Map.get(attrs, :vidui) || %{}

    categories =
      Enum.filter(["avonot", "peshaim", "chatot"], fn slot ->
        case Map.get(vidui, slot) do
          list when is_list(list) and list != [] -> true
          _ -> false
        end
      end)

    "case=#{case_ref} shaliach=#{shaliach} categories=#{Enum.join(categories, ",")}"
  end

  # Translate the Python-side JSON shape into Ecto-friendly attrs.
  # The Python AzazelRecord.to_storage_dict() shape is:
  #   {
  #     "case_ref": str,
  #     "vidui": {avonot, peshaim, chatot, confessor, confessed_at},
  #     "routing": "eretz_gezerah",
  #     "authorized_by": str,
  #     "shaliach": str,
  #     "triggering_call": str | null,
  #     "goral_outcome_id": str | null,
  #     "disposed_at": iso8601 string
  #   }
  # We pass `vidui` through as-is (JSONB column) and parse the
  # disposed_at into a UTC DateTime.
  defp normalize_record(raw) when is_map(raw) do
    with {:ok, disposed_at} <- parse_iso8601(Map.get(raw, "disposed_at")) do
      {:ok,
       %{
         case_ref: Map.get(raw, "case_ref"),
         vidui: Map.get(raw, "vidui"),
         routing: Map.get(raw, "routing"),
         authorized_by: Map.get(raw, "authorized_by"),
         shaliach: Map.get(raw, "shaliach"),
         triggering_call: Map.get(raw, "triggering_call"),
         goral_outcome_id: Map.get(raw, "goral_outcome_id"),
         disposed_at: disposed_at
       }}
    end
  end

  defp normalize_record(_), do: {:error, :not_a_map}

  defp parse_iso8601(nil), do: {:error, :missing_disposed_at}

  defp parse_iso8601(s) when is_binary(s) do
    case DateTime.from_iso8601(s) do
      {:ok, dt, _offset} -> {:ok, DateTime.truncate(dt, :second)}
      {:error, _} = err -> err
    end
  end
end