bikoret/vav_pipeline/test/azazel_ingest_test.exs

defmodule VavPipeline.AzazelIngestTest do
  @moduledoc """
  Tests for the cross-language Azazel ingest GenServer (Pass 4 /
  Finding F, commit 7). These tests verify the file-watcher logic
  and the AzazelRecord changeset validation; they do NOT require
  PostgreSQL because the recorder is injected as a mock.

  L3-side tests (the actual record_disposal write) live in
  l3_store_test.exs tagged :l3.
  """

  use ExUnit.Case, async: false

  alias VavPipeline.AzazelIngest
  alias VavPipeline.L3.AzazelRecord

  # ============================================================
  # Test helpers
  # ============================================================

  setup do
    # Per-test pending directory so tests don't share state.
    dir =
      System.tmp_dir!()
      |> Path.join("azazel-ingest-test-#{:erlang.unique_integer([:positive])}")

    File.mkdir_p!(dir)
    on_exit(fn -> File.rm_rf!(dir) end)
    %{pending_dir: dir}
  end

  defp valid_record_json(case_ref \\ "case-001") do
    %{
      "case_ref" => case_ref,
      "vidui" => %{
        "avonot" => [],
        "peshaim" => [],
        "chatot" => ["empty-vital offering: test"],
        "confessor" => "kohen-gadol",
        "confessed_at" => "2026-04-28T12:00:00Z"
      },
      "routing" => "eretz_gezerah",
      "authorized_by" => "kohen-gadol",
      "shaliach" => "aharon",
      "triggering_call" => "external_offering",
      "goral_outcome_id" => nil,
      "disposed_at" => "2026-04-28T12:00:01Z"
    }
    |> Jason.encode!()
  end

  defp drop_file(dir, name, body) do
    File.write!(Path.join(dir, name), body)
  end

  defp start_ingest(dir, opts \\ []) do
    name = :"ingest_#{:erlang.unique_integer([:positive])}"

    captured = :ets.new(:captured, [:public, :bag])

    recorder = fn attrs ->
      :ets.insert(captured, {:record, attrs})
      {:ok, attrs}
    end

    {:ok, pid} =
      AzazelIngest.start_link(
        Keyword.merge(
          [
            name: name,
            pending_dir: dir,
            poll_interval_ms: 60_000,  # don't auto-poll during tests
            recorder: recorder
          ],
          opts
        )
      )

    on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end)
    %{server: name, captured: captured, recorder: recorder}
  end

  # ============================================================
  # Happy path
  # ============================================================

  test "poll_now ingests a valid file and shreds it", %{pending_dir: dir} do
    drop_file(dir, "case-001.json", valid_record_json())
    %{server: server, captured: captured} = start_ingest(dir)

    AzazelIngest.poll_now(server)

    # File was shredded
    refute File.exists?(Path.join(dir, "case-001.json"))

    # Record was passed to the recorder
    assert [{:record, attrs}] = :ets.tab2list(captured)
    assert attrs.case_ref == "case-001"
    assert attrs.routing == "eretz_gezerah"
    assert attrs.authorized_by == "kohen-gadol"
    assert attrs.shaliach == "aharon"
    assert %DateTime{} = attrs.disposed_at
  end

  test "poll_now ingests multiple files in sorted order", %{pending_dir: dir} do
    drop_file(dir, "case-002.json", valid_record_json("case-002"))
    drop_file(dir, "case-001.json", valid_record_json("case-001"))
    drop_file(dir, "case-003.json", valid_record_json("case-003"))

    %{server: server, captured: captured} = start_ingest(dir)
    AzazelIngest.poll_now(server)

    refs =
      :ets.tab2list(captured)
      |> Enum.map(fn {:record, attrs} -> attrs.case_ref end)

    assert refs == ["case-001", "case-002", "case-003"]
  end

  test "ingest preserves the vidui as a map (jsonb column)", %{pending_dir: dir} do
    drop_file(dir, "case-vidui.json", valid_record_json())
    %{server: server, captured: captured} = start_ingest(dir)

    AzazelIngest.poll_now(server)

    [{:record, attrs}] = :ets.tab2list(captured)
    assert is_map(attrs.vidui)
    assert attrs.vidui["chatot"] == ["empty-vital offering: test"]
    assert attrs.vidui["confessor"] == "kohen-gadol"
  end

  # ============================================================
  # Retry semantics — failed records left in place
  # ============================================================

  test "file is left in place when recorder returns error", %{pending_dir: dir} do
    drop_file(dir, "case-fail.json", valid_record_json())

    name = :"ingest_fail_#{:erlang.unique_integer([:positive])}"

    failing_recorder = fn _ -> {:error, :db_unavailable} end

    {:ok, pid} =
      AzazelIngest.start_link(
        name: name,
        pending_dir: dir,
        poll_interval_ms: 60_000,
        recorder: failing_recorder
      )

    on_exit(fn -> if Process.alive?(pid), do: GenServer.stop(pid) end)

    AzazelIngest.poll_now(name)

    # File still present for retry
    assert File.exists?(Path.join(dir, "case-fail.json"))
    state = AzazelIngest.state(name)
    assert state.errors == 1
    assert state.records_ingested == 0
  end

  test "file is left in place when JSON is malformed", %{pending_dir: dir} do
    drop_file(dir, "case-malformed.json", "not-json{")
    %{server: server} = start_ingest(dir)

    AzazelIngest.poll_now(server)

    # File still present (operator can investigate)
    assert File.exists?(Path.join(dir, "case-malformed.json"))
    state = AzazelIngest.state(server)
    assert state.errors == 1
  end

  test "file is left in place when disposed_at is missing", %{pending_dir: dir} do
    raw =
      %{"case_ref" => "case-no-ts", "vidui" => %{}, "routing" => "eretz_gezerah"}
      |> Jason.encode!()

    drop_file(dir, "case-no-ts.json", raw)
    %{server: server} = start_ingest(dir)

    AzazelIngest.poll_now(server)

    assert File.exists?(Path.join(dir, "case-no-ts.json"))
  end

  # ============================================================
  # State accounting
  # ============================================================

  test "state tracks polls_completed, records_ingested, errors", %{pending_dir: dir} do
    drop_file(dir, "case-001.json", valid_record_json("case-001"))
    drop_file(dir, "case-002.json", valid_record_json("case-002"))
    drop_file(dir, "case-bad.json", "garbage")

    %{server: server} = start_ingest(dir)

    AzazelIngest.poll_now(server)

    state = AzazelIngest.state(server)
    assert state.polls_completed == 1
    assert state.records_ingested == 2
    assert state.errors == 1
  end

  test "state survives multiple polls", %{pending_dir: dir} do
    %{server: server} = start_ingest(dir)
    AzazelIngest.poll_now(server)
    AzazelIngest.poll_now(server)
    AzazelIngest.poll_now(server)

    state = AzazelIngest.state(server)
    assert state.polls_completed == 3
  end

  test "non-json files are ignored", %{pending_dir: dir} do
    File.write!(Path.join(dir, "README.txt"), "not a record")
    drop_file(dir, "case-001.json", valid_record_json())

    %{server: server, captured: captured} = start_ingest(dir)
    AzazelIngest.poll_now(server)

    # JSON file processed; README left alone
    refute File.exists?(Path.join(dir, "case-001.json"))
    assert File.exists?(Path.join(dir, "README.txt"))
    assert length(:ets.tab2list(captured)) == 1
  end

  # ============================================================
  # Property 6 — out-of-band acknowledgment via Chatzotzrot
  # ============================================================

  test "successful ingest fires Chatzotzrot :moadim signal", %{pending_dir: dir} do
    # The supervision-tree-started Chatzotzrot is already running.
    # Subscribe this test process as an :operators recipient and
    # observe the signal AzazelIngest fires.
    VavPipeline.Chatzotzrot.subscribe(:all)

    drop_file(dir, "case-ack-001.json", valid_record_json("case-ack-001"))
    %{server: server} = start_ingest(dir)
    AzazelIngest.poll_now(server)

    # Drain any unrelated signals from prior tests; we want OUR signal.
    receive_until_ours("case-ack-001")
  end

  test "chatzotzrot fire does NOT include vidui item content (privacy)" do
    # Property 9's privacy boundary applies to the chatzotzrot
    # summary too — the operator-side signal channel must not leak
    # the operator-internal vidui items.
    secret_phrase = "must-not-leak-via-chatzotzrot-summary-channel"

    body =
      %{
        "case_ref" => "leak-001",
        "vidui" => %{
          "avonot" => [],
          "peshaim" => [],
          "chatot" => [secret_phrase],
          "confessor" => "kohen",
          "confessed_at" => "2026-04-28T12:00:00Z"
        },
        "routing" => "eretz_gezerah",
        "authorized_by" => "kohen-gadol",
        "shaliach" => "aharon",
        "triggering_call" => "test",
        "goral_outcome_id" => nil,
        "disposed_at" => "2026-04-28T12:00:01Z"
      }
      |> Jason.encode!()

    VavPipeline.Chatzotzrot.subscribe(:all)

    dir =
      System.tmp_dir!() |> Path.join("azazel-leak-#{:erlang.unique_integer([:positive])}")

    File.mkdir_p!(dir)
    on_exit(fn -> File.rm_rf!(dir) end)
    File.write!(Path.join(dir, "leak-001.json"), body)

    %{server: server} = start_ingest(dir)
    AzazelIngest.poll_now(server)

    signal = receive_until_ours("leak-001")
    refute signal.details.reason =~ secret_phrase
  end

  # Drain any unrelated chatzotzrot signals (from other tests sharing
  # the supervision-tree Chatzotzrot) until we see ours, identified by
  # case_ref in the reason. Returns the matching signal.
  defp receive_until_ours(case_ref) do
    receive do
      {:chatzotzrot, signal} ->
        if String.contains?(signal.details.reason || "", case_ref) and
             signal.signal_type == :moadim do
          assert signal.details.reason =~ "Azazel disposal completed"
          signal
        else
          receive_until_ours(case_ref)
        end
    after
      2_000 -> flunk("Did not receive chatzotzrot signal for #{case_ref}")
    end
  end

  # ============================================================
  # Changeset validation (no DB)
  # ============================================================

  describe "AzazelRecord.changeset/2" do
    test "valid attrs produce a valid changeset" do
      attrs = %{
        case_ref: "case-cs-001",
        vidui: %{
          "avonot" => [],
          "peshaim" => [],
          "chatot" => ["something"],
          "confessor" => "kohen",
          "confessed_at" => "2026-04-28T12:00:00Z"
        },
        routing: "eretz_gezerah",
        authorized_by: "kohen-gadol",
        shaliach: "aharon",
        disposed_at: ~U[2026-04-28 12:00:01Z]
      }

      cs = AzazelRecord.changeset(%AzazelRecord{}, attrs)
      assert cs.valid?
    end

    test "rejects routing other than eretz_gezerah" do
      attrs = base_attrs(%{routing: "void"})
      cs = AzazelRecord.changeset(%AzazelRecord{}, attrs)
      refute cs.valid?
      assert {:routing, _} = List.keyfind(cs.errors, :routing, 0)
    end

    test "rejects vidui missing avonot slot" do
      attrs =
        base_attrs(%{
          vidui: %{
            "peshaim" => [],
            "chatot" => ["x"],
            "confessor" => "kohen",
            "confessed_at" => "2026-04-28T12:00:00Z"
          }
        })

      cs = AzazelRecord.changeset(%AzazelRecord{}, attrs)
      refute cs.valid?
      assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0)
      assert msg =~ "avonot"
    end

    test "rejects vidui with all three categories empty" do
      attrs =
        base_attrs(%{
          vidui: %{
            "avonot" => [],
            "peshaim" => [],
            "chatot" => [],
            "confessor" => "kohen",
            "confessed_at" => "2026-04-28T12:00:00Z"
          }
        })

      cs = AzazelRecord.changeset(%AzazelRecord{}, attrs)
      refute cs.valid?
      assert {:vidui, {msg, _}} = List.keyfind(cs.errors, :vidui, 0)
      assert msg =~ "untagged"
    end

    test "rejects missing required fields" do
      cs = AzazelRecord.changeset(%AzazelRecord{}, %{})
      refute cs.valid?

      missing = Enum.map(cs.errors, fn {k, _} -> k end)
      for required <- [:case_ref, :vidui, :routing, :authorized_by, :shaliach, :disposed_at] do
        assert required in missing, "expected #{required} to be missing"
      end
    end

    test "accepts optional triggering_call and goral_outcome_id" do
      attrs = base_attrs(%{
        triggering_call: "external_offering",
        goral_outcome_id: String.duplicate("a", 64)
      })

      cs = AzazelRecord.changeset(%AzazelRecord{}, attrs)
      assert cs.valid?
    end

    defp base_attrs(overrides \\ %{}) do
      Map.merge(
        %{
          case_ref: "case-base-001",
          vidui: %{
            "avonot" => [],
            "peshaim" => [],
            "chatot" => ["something"],
            "confessor" => "kohen",
            "confessed_at" => "2026-04-28T12:00:00Z"
          },
          routing: "eretz_gezerah",
          authorized_by: "kohen-gadol",
          shaliach: "aharon",
          disposed_at: ~U[2026-04-28 12:00:01Z]
        },
        overrides
      )
    end
  end
end