# Phase Recon-3 — File Upload API + Async Job Pipeline

**Status:** ⬜ Pending
**Duration:** Weeks 5–6
**Depends on:** Phase Recon-1, Phase Recon-2
**Goal:** Elixir orchestration layer wires file uploads → matching engine → report generator into a single async job tracked via `AsyncJobStore`. UI will subscribe to PubSub events in Phase 4.

---

## Orchestration Flow

```
User uploads files
      │
      ▼
ReconController.upload/2         ← POST /api/v1/recon/sessions/:id/files
  │  store files as ReconRunFile rows in DB
  │  create async_jobs record
  │
  ▼
MwRecon.Orchestrator.start_run/1  ← spawned async via Task.Supervisor
  │
  ├─[step 1: parsing]──────────────────────────────────────────────────────────
  │   Read file binaries from DB → call AdapterCloudi (file_parser_py)
  │   to validate columns + detect format
  │   PubSub: {session_id, :parsing, %{progress: 1, total: 3}}
  │
  ├─[step 2: matching]─────────────────────────────────────────────────────────
  │   Encode files as base64 → call MwRecon.EngineClient.run_match/1
  │   (CloudI → recon_engine_py)
  │   PubSub: {session_id, :matching, %{}}
  │
  ├─[step 3: reporting]────────────────────────────────────────────────────────
  │   Build report payload → call MwRecon.ReportClient.generate/1
  │   (CloudI → recon_report_py)
  │   PubSub: {session_id, :reporting, %{}}
  │
  └─[step 4: completed]────────────────────────────────────────────────────────
      Store report_data (base64 XLSX) on ReconSession
      Update matched_count, unmatched_count, amounts
      PubSub: {session_id, :completed, %{matched: N, unmatched: M, ...}}
      mw_audit: emit "recon.session.completed" event
```

---

## New API Routes (`gateway_api/router.ex`)

```elixir
# Add inside the authenticated :api scope
scope "/api/v1/recon", GatewayApiWeb do
  pipe_through [:api, :authenticated]

  post   "/sessions",                      ReconController, :create_session
  post   "/sessions/:id/files",            ReconController, :upload_files
  post   "/sessions/:id/run",              ReconController, :start_run
  get    "/sessions/:id/status",           ReconController, :status
  get    "/sessions/:id/report",           ReconController, :download_report
  get    "/sessions",                      ReconController, :list_sessions
end
```

---

## `gateway_api/controllers/recon_controller.ex`

```elixir
defmodule GatewayApiWeb.ReconController do
  use GatewayApiWeb, :controller
  alias InfraRepo.ReconSessions
  alias MwRecon.Orchestrator

  @max_file_size_bytes 20 * 1024 * 1024  # 20 MB

  @doc """
  POST /api/v1/recon/sessions
  Body: { "recon_type": "bank_card_vs_momentspay", "recon_date": "2025-06-09", "config_id": 1 }
  Returns: 201 Created with { session_id: ... }
  """
  def create_session(conn, params) do
    tenant_id   = conn.assigns.current_tenant_id
    user_id     = conn.assigns.current_user_id
    attrs = %{
      tenant_id:    tenant_id,
      recon_type:   params["recon_type"],
      recon_date:   parse_date(params["recon_date"]),
      config_id:    params["config_id"],
      initiated_by_id: user_id,
      status:       "pending"
    }
    case ReconSessions.create_session(attrs) do
      {:ok, session} ->
        conn |> put_status(:created) |> json(%{session_id: session.id, status: session.status})
      {:error, changeset} ->
        conn |> put_status(:unprocessable_entity) |> json(%{errors: format_errors(changeset)})
    end
  end

  @doc """
  POST /api/v1/recon/sessions/:id/files
  Multipart form: file_role=bank_card, file=<binary>
  Also accepts JSON: { "role": "bank_card", "filename": "...", "content_base64": "..." }
  Returns: 200 OK with file metadata
  """
  def upload_files(conn, %{"id" => session_id} = params) do
    session = ReconSessions.get_session!(session_id)
    with :ok <- check_tenant(conn, session),
         {:ok, role, filename, content_b64} <- extract_file(params),
         :ok <- check_file_size(content_b64) do
      case ReconSessions.add_run_file(%{
        session_id:  session.id,
        role:        role,
        filename:    filename,
        file_format: detect_format(filename),
        content:     content_b64
      }) do
        {:ok, run_file} ->
          json(conn, %{file_id: run_file.id, role: role, filename: filename})
        {:error, cs} ->
          conn |> put_status(:unprocessable_entity) |> json(%{errors: format_errors(cs)})
      end
    else
      {:error, reason} ->
        conn |> put_status(:bad_request) |> json(%{error: reason})
    end
  end

  @doc """
  POST /api/v1/recon/sessions/:id/run
  Starts the async reconciliation job.
  Returns: 202 Accepted with { job_id: ... }
  """
  def start_run(conn, %{"id" => session_id}) do
    session = ReconSessions.get_session_with_files(session_id)
    with :ok <- check_tenant(conn, session),
         :ok <- validate_files_complete(session),
         {:ok, job_id} <- Orchestrator.start_run(session) do
      conn |> put_status(:accepted) |> json(%{job_id: job_id, session_id: session.id})
    else
      {:error, reason} ->
        conn |> put_status(:unprocessable_entity) |> json(%{error: reason})
    end
  end

  @doc "GET /api/v1/recon/sessions/:id/status"
  def status(conn, %{"id" => session_id}) do
    session = ReconSessions.get_session!(session_id)
    with :ok <- check_tenant(conn, session) do
      json(conn, %{
        session_id:      session.id,
        status:          session.status,
        matched_count:   session.matched_count,
        unmatched_count: session.unmatched_count,
        total_count:     session.total_count,
        error_reason:    session.error_reason
      })
    end
  end

  @doc "GET /api/v1/recon/sessions/:id/report — returns XLSX"
  def download_report(conn, %{"id" => session_id}) do
    session = ReconSessions.get_session!(session_id)
    with :ok <- check_tenant(conn, session),
         "completed" <- session.status,
         {:ok, xlsx_bytes} <- Base.decode64(session.report_data || "") do
      filename = "recon_#{session.recon_type}_#{session.recon_date}.xlsx"
      conn
      |> put_resp_content_type("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
      |> put_resp_header("content-disposition", ~s(attachment; filename="#{filename}"))
      |> send_resp(200, xlsx_bytes)
    else
      status when is_binary(status) ->
        conn |> put_status(:conflict) |> json(%{error: "Report not ready, status: #{status}"})
      _ ->
        conn |> put_status(:not_found) |> json(%{error: "Report not available"})
    end
  end

  defp parse_date(nil), do: Date.utc_today()
  defp parse_date(s),   do: Date.from_iso8601!(s)
  defp detect_format(filename) when is_binary(filename) do
    filename |> Path.extname() |> String.trim_leading(".") |> String.downcase()
  end
  defp check_file_size(b64) do
    if byte_size(b64) * 3 / 4 > @max_file_size_bytes,
      do: {:error, "File exceeds 20MB limit"}, else: :ok
  end
  defp check_tenant(conn, session) do
    if session.tenant_id == conn.assigns.current_tenant_id, do: :ok, else: {:error, :forbidden}
  end
  defp format_errors(cs), do: Ecto.Changeset.traverse_errors(cs, &elem(&1, 0))
  defp extract_file(%{"role" => role, "filename" => fn_, "content_base64" => b64}),
    do: {:ok, role, fn_, b64}
  defp extract_file(_), do: {:error, "Missing role, filename, or content_base64"}
  defp validate_files_complete(session) do
    required = required_roles(session.recon_type)
    present  = Enum.map(session.run_files, & &1.role)
    missing  = required -- present
    if missing == [], do: :ok, else: {:error, "Missing required files: #{Enum.join(missing, ", ")}"}
  end
  defp required_roles("bank_card_vs_momentspay"),  do: ["bank_card", "momentspay"]
  defp required_roles("bank_upi_vs_momentspay"),   do: ["bank_upi", "momentspay"]
  defp required_roles("his_bank_card"),             do: ["his", "bank_card"]
  defp required_roles("his_bank_upi"),              do: ["his", "bank_upi"]
  defp required_roles("his_momentspay_card"),       do: ["his", "momentspay", "bank_card"]
  defp required_roles("his_momentspay_upi"),        do: ["his", "momentspay", "bank_upi"]
  defp required_roles("amex"),                      do: ["his", "amex"]
  defp required_roles(_),                           do: []
end
```

---

## `apps/mw_recon/lib/mw_recon/orchestrator.ex`

```elixir
defmodule MwRecon.Orchestrator do
  @moduledoc """
  Runs a complete reconciliation job asynchronously.

  Lifecycle:
    pending → parsing → matching → reporting → completed
                                              → failed (any step)

  Broadcasts status updates to Phoenix.PubSub topic: "recon:session:{session_id}"
  """
  require Logger
  alias InfraRepo.{ReconSessions, Repo}
  alias InfraRepo.Schemas.{ReconConfig, ReconLocationMap}
  alias MwRecon.{EngineClient, ReportClient}
  alias InfraQueue.AsyncJobStore
  alias MwAudit

  @pubsub MwCore.PubSub

  @spec start_run(map()) :: {:ok, String.t()} | {:error, String.t()}
  def start_run(session) do
    # Prevent duplicate runs
    if active_job_exists?(session.tenant_id) do
      {:error, "A reconciliation job is already running for this hospital"}
    else
      {:ok, job} = AsyncJobStore.create(%{type: "recon_match"})
      {:ok, _} = ReconSessions.update_session(session, %{job_id: job.id, status: "pending"})

      Task.Supervisor.start_child(MwCore.TaskSupervisor, fn ->
        run_job(session.id, job.id)
      end)

      {:ok, job.id}
    end
  end

  defp run_job(session_id, job_id) do
    session = ReconSessions.get_session_with_files(session_id)
    config  = load_config(session)

    MwAudit.log("recon.session.started", %{
      session_id: session_id,
      tenant_id:  session.tenant_id,
      recon_type: session.recon_type
    })

    with {:ok, files_b64}    <- step_parse(session),
         {:ok, match_result} <- step_match(session, files_b64, config),
         {:ok, report_b64}   <- step_report(session, match_result, config) do

      summary = match_result["summary"] || %{}
      ReconSessions.update_session(session, %{
        status:          "completed",
        matched_count:   summary["matched"],
        unmatched_count: summary["unmatched"],
        total_count:     summary["total"],
        matched_amount:  Decimal.new(to_string(summary["matched_amount"] || "0")),
        unmatched_amount: Decimal.new(to_string(summary["unmatched_amount"] || "0")),
        report_data:     report_b64
      })
      AsyncJobStore.complete(job_id, %{session_id: session_id})
      broadcast(session_id, :completed, summary)
      MwAudit.log("recon.session.completed", %{session_id: session_id, summary: summary})
    else
      {:error, reason} ->
        ReconSessions.update_session(session, %{status: "failed", error_reason: reason})
        AsyncJobStore.fail(job_id, reason)
        broadcast(session_id, :failed, %{error: reason})
        MwAudit.log("recon.session.failed", %{session_id: session_id, reason: reason})
        Logger.error("Recon session #{session_id} failed: #{reason}")
    end
  end

  defp step_parse(session) do
    broadcast(session.id, :parsing, %{})
    set_status(session.id, "parsing")
    # Files are already stored as base64 strings in DB — no re-parsing needed.
    # Build the files map keyed by role.
    files = Map.new(session.run_files, fn f -> {f.role, f.content} end)
    {:ok, files}
  end

  defp step_match(session, files_b64, config) do
    broadcast(session.id, :matching, %{})
    set_status(session.id, "matching")

    payload = EngineClient.build_payload(session, files_b64, session.tenant_id)
    EngineClient.run_match(payload)
  end

  defp step_report(session, match_result, config) do
    broadcast(session.id, :reporting, %{})
    set_status(session.id, "reporting")

    location_map = load_location_map(config)
    payload = ReportClient.build_payload(
      session.tenant_id,
      match_result,
      location_map
    )
    case ReportClient.generate(payload) do
      {:ok, b64} -> {:ok, b64}
      {:error, r} -> {:error, r}
    end
  end

  defp set_status(session_id, status) do
    ReconSessions.set_status(session_id, status)
  end

  defp broadcast(session_id, event, payload) do
    Phoenix.PubSub.broadcast(@pubsub, "recon:session:#{session_id}", {event, payload})
  end

  defp load_config(session) do
    if session.config_id do
      Repo.get(ReconConfig, session.config_id) |> Repo.preload(:location_maps)
    else
      nil
    end
  end

  defp load_location_map(nil), do: %{}
  defp load_location_map(config) do
    Map.new(config.location_maps, fn lm -> {lm.full_name, lm.short_code} end)
  end

  defp active_job_exists?(tenant_id) do
    import Ecto.Query
    alias InfraRepo.Schemas.ReconSession
    InfraRepo.Repo.exists?(
      from s in ReconSession,
      where: s.tenant_id == ^tenant_id and s.status in ["parsing", "matching", "reporting"]
    )
  end
end
```

---

## Acceptance Criteria

- [ ] `POST /api/v1/recon/sessions` creates a session and returns `session_id`
- [ ] `POST /api/v1/recon/sessions/:id/files` stores files; returns 413 for files > 20MB
- [ ] `POST /api/v1/recon/sessions/:id/run` returns 202 with `job_id`
- [ ] Job status transitions fire PubSub events: `parsing` → `matching` → `reporting` → `completed`
- [ ] `GET /api/v1/recon/sessions/:id/report` returns XLSX when status is `completed`
- [ ] Second run on same tenant while one is active returns error "already running"
- [ ] Failed job sets `error_reason` on the session and `async_jobs.error`