gust_web

defmodule GustWeb.RunLive.Index do
  alias Gust.Flows
  alias Gust.Flows.Run
  use GustWeb, :live_view

  @impl true
  def mount(params, _session, socket) do
    %Flows.Dag{runs: runs} = dag = Flows.get_dag_with_runs_and_tasks!(params["name"])

    {:ok, dag_def} = Gust.DAG.Loader.reload_definition(dag.id)

    selected_run = if params["run_id"], do: Flows.get_run!(params["run_id"])

    {selected_task, logs} = maybe_get_task(params["task_name"], params["run_id"]) || {nil, []}

    if connected?(socket), do: subscribe_updates(dag, runs)

    {:ok,
     socket
     |> assign(:dag_def, dag_def)
     |> assign(:dag, dag)
     |> assign(:selected_task, selected_task)
     |> assign(:selected_run, selected_run)
     |> assign(:reload_dag_file, {dag_def.file_path, time()})
     |> stream(:logs, logs)
     |> stream(:runs, runs)}
  end

  defp maybe_get_task(nil, nil), do: nil
  defp maybe_get_task(nil, _run_id), do: nil

  defp maybe_get_task(task_name, run_id) do
    task = Flows.get_task_by_name_run_with_logs(task_name, run_id)
    Gust.PubSub.subscribe_task(task.id)
    {task, task.logs}
  end

  defp subscribe_updates(dag, runs) do
    Enum.each(runs, fn run -> Gust.PubSub.subscribe_run(run.id) end)
    Gust.PubSub.subscribe_runs_for_dag(dag.id)
    Gust.PubSub.subscribe_file(dag.name)
  end

  def time, do: DateTime.utc_now() |> DateTime.to_iso8601()

  defp mermaid_chart(tasks), do: GustWeb.Mermaid.chart(tasks)

  defp read_code({file_path, _reload_time}), do: File.read!(file_path)
  defp reload_time({_file_path, reload_time}), do: reload_time

  @impl true
  def handle_info({:task, :log, %{task_id: _task_id, log_id: log_id}}, socket) do
    log = Flows.get_log!(log_id)

    {:noreply, socket |> stream_insert(:logs, log)}
  end

  @impl true
  def handle_info(
        {:dag, :file_updated, %{action: "reload", dag_name: _name, dag_def: dag_def}},
        socket
      ) do
    {:noreply,
     socket
     |> assign(:dag_def, dag_def)
     |> assign(:reload_dag_file, {dag_def.file_path, time()})}
  end

  @impl true
  def handle_info(
        {:dag, :run_started, %{run_id: run_id}},
        socket
      ) do
    run = Flows.get_run_with_tasks!(run_id)
    Gust.PubSub.subscribe_run(run_id)

    {:noreply, stream_insert(socket, :runs, run)}
  end

  @impl true
  def handle_info(
        {:dag, :run_status, %{run_id: run_id, status: _status}},
        socket
      ) do
    run = Flows.get_run_with_tasks!(run_id)

    {:noreply, stream_insert(socket, :runs, run)}
  end

  @impl true
  def handle_info(:trigger_run, socket) do
    dag = socket.assigns.dag
    dag_def = socket.assigns.dag_def

    {:ok, dag_def} = Gust.DAG.Parser.parse(dag_def.file_path)

    {:ok, run} = Gust.DAG.Starter.start_dag_run(dag.id, dag_def)
    run = Flows.get_run_with_tasks!(run.id)

    {:noreply, socket |> put_flash(:info, "Run #{run.id} triggered")}
  end

  defp pretty_json!(value) do
    Jason.encode_to_iodata!(value, pretty: true, escape_html: true)
  end
end