defmodule OpenAgents.Forge.GitHTTP do
  @moduledoc """
  Git smart-HTTP v0, wrapping the stock git binary ("Spokes got that exactly
  right" — standard packfiles, upstream clients, no custom object format).

  Serves under the canonical `/:owner/:repo.git` mount point:

      GET  /:repo.git/info/refs?service=git-upload-pack|git-receive-pack
      POST /:repo.git/git-upload-pack
      POST /:repo.git/git-receive-pack

  Reads (`upload-pack`) run against the local bare repo after a WAL
  freshness check. Writes (`receive-pack`) go through `OpenAgents.Forge.Pushes`:
  applied locally, then persisted to the WAL — the push is acked only after
  the WAL accepts it (Continuity rule), otherwise refs are rolled back and
  the client sees a failed push.

  Request bodies are bounded and gunzipped when the client says so. The
  `Git-Protocol` header is passed through so protocol v2 works.
  """

  @behaviour Plug

  import Ecto.Query
  import Plug.Conn

  require Logger

  alias OpenAgents.{Audit, Repositories}
  alias OpenAgents.Forge.{Pushes, Repos, Sync}

  @max_body_bytes 512 * 1024 * 1024
  @read_chunk 8 * 1024 * 1024

  @impl true
  def init(opts), do: opts

  @impl true
  def call(conn, _opts) do
    case {conn.method, split_repo(conn)} do
      {"GET", {:ok, owner, name, ["info", "refs"]}} ->
        advertise(conn, owner, name, first_query(conn, "service"))

      {"POST", {:ok, owner, name, ["git-upload-pack"]}} ->
        upload_pack(conn, owner, name)

      {"POST", {:ok, owner, name, ["git-receive-pack"]}} ->
        receive_pack(conn, owner, name)

      _other ->
        send_resp(conn, 404, "not found") |> halt()
    end
  end

  # ── routes ──────────────────────────────────────────────────────────────

  defp advertise(conn, owner, name, service)
       when service in ["git-upload-pack", "git-receive-pack"] do
    operation = if service == "git-upload-pack", do: :read, else: :write

    with {:ok, repository} <- resolve_repository(conn, owner, name),
         :ok <- authorize(conn, repository, operation),
         :ok <- Sync.ensure_fresh(repository.storage_key, repository.default_branch) do
      command = String.trim_leading(service, "git-")
      path = Repos.ensure_repo!(repository.storage_key, repository.default_branch)

      {output, 0} =
        run_git_service(command, ["--advertise-refs", path], "", git_protocol(conn))

      header = pkt_line("# service=#{service}\n") <> "0000"

      conn
      |> put_resp_content_type("application/x-#{service}-advertisement")
      |> put_resp_header("cache-control", "no-cache")
      |> send_resp(200, header <> output)
      |> halt()
    else
      error -> send_git_error(conn, error)
    end
  end

  defp advertise(conn, _owner, _name, _service),
    do: send_resp(conn, 400, "dumb http protocol is not supported") |> halt()

  defp upload_pack(conn, owner, name) do
    with {:ok, repository} <- resolve_repository(conn, owner, name),
         :ok <- authorize(conn, repository, :read),
         {:ok, body, conn} <- read_git_body(conn),
         :ok <- Sync.ensure_fresh(repository.storage_key, repository.default_branch) do
      path = Repos.ensure_repo!(repository.storage_key, repository.default_branch)
      {output, _status} = run_git_service("upload-pack", [path], body, git_protocol(conn))

      conn
      |> put_resp_content_type("application/x-git-upload-pack-result")
      |> put_resp_header("cache-control", "no-cache")
      |> send_resp(200, output)
      |> halt()
    else
      error -> send_git_error(conn, error)
    end
  end

  defp receive_pack(conn, owner, name) do
    with {:ok, repository} <- resolve_repository(conn, owner, name),
         :ok <- authorize(conn, repository, :write),
         {:ok, body, conn} <- read_git_body(conn),
         :ok <- authorize_receive_pack(conn, repository, body) do
      case Pushes.handle_receive_pack(
             repository.storage_key,
             body,
             principal(conn),
             git_protocol(conn)
           ) do
        {:ok, output, receipt} ->
          Audit.record!(
            "repository.git.write",
            audit_actor(conn),
            "repository",
            repository.id || repository.storage_key,
            repository_id: repository.id,
            metadata: %{"operation" => "receive_pack"}
          )

          conn
          |> put_resp_content_type("application/x-git-receive-pack-result")
          |> put_resp_header("cache-control", "no-cache")
          |> send_resp(200, with_wal_receipt(output, repository, receipt))
          |> halt()

        {:error, :wal_persist_failed} ->
          send_resp(conn, 503, "push not persisted; refs rolled back — retry") |> halt()

        {:error, %OpenAgents.Forge.SyncError{}} ->
          send_resp(conn, 503, "repository cache unavailable; retry") |> halt()

        {:error, _reason} ->
          send_resp(conn, 500, "push failed") |> halt()
      end
    else
      error -> send_git_error(conn, error)
    end
  end

  # ── helpers ─────────────────────────────────────────────────────────────

  defp split_repo(%Plug.Conn{
         path_params: %{"owner" => owner, "repo" => segment},
         path_info: [_owner, _repo | rest]
       }) do
    case strip_git_suffix(segment) do
      {:ok, name} -> {:ok, owner, name, rest}
      :error -> :error
    end
  end

  defp split_repo(%Plug.Conn{path_info: path_info}), do: split_legacy_repo(path_info)

  defp split_legacy_repo([segment | rest]) do
    case strip_git_suffix(segment) do
      {:ok, "openagents.com"} -> {:ok, "OpenAgentsInc", "openagents.com", rest}
      {:ok, name} -> {:ok, nil, name, rest}
      :error -> split_namespaced_repo(segment, rest)
    end
  end

  defp split_legacy_repo(_), do: :error

  defp split_namespaced_repo(owner, [segment | rest]) do
    case strip_git_suffix(segment) do
      {:ok, name} -> {:ok, owner, name, rest}
      :error -> :error
    end
  end

  defp split_namespaced_repo(_owner, _rest), do: :error

  defp strip_git_suffix(segment) do
    if String.ends_with?(segment, ".git") and byte_size(segment) > 4 do
      {:ok, String.trim_trailing(segment, ".git")}
    else
      :error
    end
  end

  defp resolve_repository(conn, owner, name) when is_binary(owner) do
    case OpenAgents.Repo.one(repository_query(owner, name)) do
      nil -> repository_not_found(conn)
      repository -> {:ok, repository}
    end
  end

  defp resolve_repository(conn, nil, name) do
    case conn.assigns[:forge_principal] do
      %{kind: kind} when kind in [:operator, :machine] ->
        if Repos.valid_name?(name) do
          {:ok,
           %OpenAgents.Repositories.Repository{
             owner: name,
             name: name,
             storage_key: name,
             default_branch: "main",
             visibility: "private",
             lifecycle_state: "ready"
           }}
        else
          repository_not_found(conn)
        end

      _principal ->
        repository_not_found(conn)
    end
  end

  defp repository_query(owner, name) do
    owner_key = String.downcase(owner)
    name_key = String.downcase(name)

    from repository in OpenAgents.Repositories.Repository,
      join: namespace in assoc(repository, :namespace),
      left_join: namespace_alias in OpenAgents.Repositories.NamespaceAlias,
      on: namespace_alias.namespace_id == namespace.id and namespace_alias.slug_key == ^owner_key,
      where:
        repository.name_key == ^name_key and repository.lifecycle_state == "ready" and
          namespace.state == "active" and
          (namespace.slug_key == ^owner_key or not is_nil(namespace_alias.id)),
      preload: [namespace: namespace]
  end

  defp authorize(conn, %{visibility: "public"} = repository, :read) do
    case conn.assigns[:forge_principal] do
      %{kind: :assignment, repository_id: repository_id} ->
        if repository.id == repository_id, do: :ok, else: {:error, 404, "unknown repository"}

      _principal ->
        :ok
    end
  end

  defp authorize(conn, repository, :read) do
    case conn.assigns[:forge_principal] do
      nil ->
        authentication_required()

      %{kind: :user, user: user} ->
        member_read(repository, user)

      %{kind: :operator} ->
        operational_access(repository)

      %{kind: :machine, id: machine_id} ->
        machine_access(repository, machine_id, "read")

      %{kind: :assignment, repository_id: repository_id} ->
        if repository.id == repository_id, do: :ok, else: {:error, 404, "unknown repository"}
    end
  end

  # An upstream mirror is one-way by construction. There is no push to the
  # upstream it names, so accepting a push here would produce a copy that
  # claims an origin it has silently diverged from — worse than a copy that
  # will not move.
  #
  # The refusal sits on the Git plane rather than in a controller because this
  # is where a push actually lands: `advertise/4` and `receive_pack/3` are the
  # only two callers, both reach it through this clause, and the clause is
  # ahead of every principal. An operator token, a paired computer's grant,
  # and an assignment credential are refused the same way an account is, so
  # there is no principal for which the mirror is writable.
  defp authorize(_conn, %{upstream_url: upstream_url}, :write) when is_binary(upstream_url) do
    {:error, 403,
     "this repository is an upstream mirror of #{upstream_url} and accepts no pushes. " <>
       "A mirror is one-way: it carries the upstream's history and its license, " <>
       "and nothing here can be pushed back to the upstream. " <>
       "Create your own repository if you want to push."}
  end

  defp authorize(conn, repository, :write) do
    case conn.assigns[:forge_principal] do
      nil ->
        authentication_required()

      %{kind: :user, user: user} ->
        if Repositories.writable?(repository, user) do
          :ok
        else
          case Repositories.membership_role(repository, user) do
            nil -> {:error, 404, "unknown repository"}
            _read_only -> {:error, 403, "repository is read only"}
          end
        end

      %{kind: :operator} ->
        operational_access(repository)

      %{kind: :machine, id: machine_id} ->
        machine_access(repository, machine_id, "write")

      %{kind: :assignment, repository_id: repository_id} ->
        if repository.id == repository_id, do: :ok, else: {:error, 404, "unknown repository"}
    end
  end

  defp authorize_receive_pack(conn, repository, body) do
    case conn.assigns[:forge_principal] do
      %{kind: :assignment, repository_id: repository_id, branch: branch}
      when repository.id == repository_id ->
        with {:ok, refs} <- OpenAgents.Forge.GitReceivePack.refs(body),
             true <-
               refs != [] and Enum.all?(refs, &allowed_assignment_ref?(&1, branch, repository)) do
          :ok
        else
          _ -> {:error, 403, "assignment branch is not authorized"}
        end

      %{kind: :assignment} ->
        {:error, 404, "unknown repository"}

      _principal ->
        :ok
    end
  end

  defp allowed_assignment_ref?(ref, branch, repository) do
    ref == "refs/heads/" <> branch and
      branch != repository.default_branch and
      branch not in (repository.protected_branches || []) and
      branch not in ["main", "master"] and
      not String.starts_with?(branch, "protected/")
  end

  defp member_read(repository, user) do
    if Repositories.membership_role(repository, user),
      do: :ok,
      else: {:error, 404, "unknown repository"}
  end

  defp operational_access(repository) do
    owner =
      case repository.namespace do
        %{slug: slug} when is_binary(slug) -> slug
        _not_loaded -> Application.get_env(:openagents, :forge_url_owner, "OpenAgentsInc")
      end

    configured? =
      Enum.any?(Repos.allowed_repos(), fn allowed ->
        allowed in [
          repository.storage_key,
          repository.name,
          "#{owner}/#{repository.name}"
        ]
      end)

    if configured?,
      do: :ok,
      else: {:error, 404, "unknown repository"}
  end

  defp machine_access(%{id: nil}, _machine_id, _operation),
    do: {:error, 404, "unknown repository"}

  defp machine_access(repository, machine_id, operation) do
    if Repositories.machine_access?(repository, machine_id, operation),
      do: :ok,
      else: {:error, 404, "unknown repository"}
  end

  defp repository_not_found(conn) do
    if conn.assigns[:forge_principal],
      do: {:error, 404, "unknown repository"},
      else: authentication_required()
  end

  defp authentication_required,
    do:
      {:error, 401, "authentication required",
       [{"www-authenticate", ~s(Basic realm="openagents-forge")}]}

  defp send_git_error(conn, {:error, status, message}) do
    conn |> put_resp_content_type("text/plain") |> send_resp(status, message) |> halt()
  end

  defp send_git_error(conn, {:error, %OpenAgents.Forge.SyncError{}}) do
    conn |> send_resp(503, "repository cache unavailable; retry") |> halt()
  end

  defp send_git_error(conn, {:error, status, message, headers}) do
    conn =
      Enum.reduce(headers, conn, fn {name, value}, acc -> put_resp_header(acc, name, value) end)

    conn |> send_resp(status, message) |> halt()
  end

  defp principal(conn) do
    case conn.assigns[:forge_principal] do
      %{kind: kind, id: id} -> "#{kind}:#{id}"
      %{kind: kind} -> to_string(kind)
      _ -> "unauthenticated"
    end
  end

  # The principal kinds a Git-plane audit event can name. Every one of them has
  # to be an actor kind `OpenAgents.Audit` accepts, or a push by that principal
  # raises after the pack has already been written. `OpenAgents.AuditTest`
  # proves the containment from this list rather than from a reading of the
  # call sites, because the actor here is a variable and a source scan for
  # `{:machine, …}` finds nothing. CANON-002.
  @audit_actor_kinds [:user, :machine, :operator]

  @doc false
  @spec audit_actor_kinds() :: [atom()]
  def audit_actor_kinds, do: @audit_actor_kinds

  defp audit_actor(conn) do
    case conn.assigns[:forge_principal] do
      %{kind: kind, id: id} when kind in @audit_actor_kinds -> {kind, id}
      _principal -> :system
    end
  end

  defp git_protocol(conn) do
    case get_req_header(conn, "git-protocol") do
      [value | _] -> value
      [] -> nil
    end
  end

  defp read_git_body(conn) do
    case read_all_body(conn, []) do
      {:ok, body, conn} ->
        case get_req_header(conn, "content-encoding") do
          ["gzip" | _] -> {:ok, safe_gunzip(body), conn}
          _ -> {:ok, body, conn}
        end

      {:error, _} ->
        {:error, 413, "request body too large"}
    end
  end

  defp read_all_body(conn, acc) do
    case read_body(conn, length: @read_chunk, read_length: @read_chunk) do
      {:ok, chunk, conn} ->
        {:ok, IO.iodata_to_binary(Enum.reverse([chunk | acc])), conn}

      {:more, chunk, conn} ->
        acc = [chunk | acc]

        if IO.iodata_length(acc) > @max_body_bytes do
          {:error, :too_large}
        else
          read_all_body(conn, acc)
        end

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp safe_gunzip(body) do
    :zlib.gunzip(body)
  rescue
    _ -> body
  end

  # ── the WAL receipt returned to the pusher (#167) ────────────────────────

  # Nothing here may fail a push. The WAL has already accepted the entry and
  # the client is about to be told so; a receipt that cannot be formatted is
  # dropped, exactly as `4651b3a` dropped a closing reference that could not be
  # applied. Refusing now would ask a client to retry a push the forge has
  # taken.
  defp with_wal_receipt(output, repository, %{seq: seq, link: link})
       when is_integer(seq) and is_binary(link) do
    append_side_band(
      output,
      "openagents wal-receipt seq=#{seq} link=#{link}" <>
        " (GET /api/v1/repos/#{repository.owner}/#{repository.name}/pushes/#{seq})"
    )
  rescue
    error ->
      Logger.warning(
        "forge_push_receipt_line_failed code=#{OpenAgents.OperationalLog.code(error)}"
      )

      output
  catch
    kind, reason ->
      Logger.warning(
        "forge_push_receipt_line_failed code=#{OpenAgents.OperationalLog.code({kind, reason})}"
      )

      output
  end

  defp with_wal_receipt(output, _repository, _no_receipt), do: output

  @doc """
  Append one informational message to a side-band-framed `receive-pack`
  response, so `git push` prints it to the pusher as a `remote:` line.

  The stream is only touched when it is demonstrably safe to touch: the
  response must parse as a pkt-line stream whose first packet carries a
  side-band designator, and it must end in a flush packet. `git` treats an
  unparseable report-status as a failed push, so a response that is not
  side-band framed — an old client, or one that did not ask for
  `side-band-64k` — is returned exactly as `git` produced it and the pusher
  gets no line rather than a broken push.

  The message rides band 2, which is progress output. It is delivered after
  the report-status the client parses, so nothing about the push's success or
  failure depends on it.
  """
  @spec append_side_band(binary(), String.t()) :: binary()
  def append_side_band(output, message) when is_binary(output) and is_binary(message) do
    text = sanitize_side_band(message)

    if text != "" and side_band_framed?(output) and String.ends_with?(output, "0000") do
      body = binary_part(output, 0, byte_size(output) - 4)
      body <> pkt_line(<<2>> <> text <> "\n") <> "0000"
    else
      output
    end
  end

  # A side-band-framed response begins with a complete pkt-line whose first
  # payload byte is a band designator. A response that begins with a flush, or
  # with `unpack ok`, is a bare report-status and must not be touched.
  defp side_band_framed?(<<hex::binary-size(4), rest::binary>>) do
    case Integer.parse(hex, 16) do
      {length, ""} when length > 4 ->
        case rest do
          <<band, _remainder::binary>> when band in 1..3 -> byte_size(rest) >= length - 4
          _short -> false
        end

      _unparseable ->
        false
    end
  end

  defp side_band_framed?(_output), do: false

  # One line, printable, and short enough that the packet cannot exceed the
  # side-band-64k ceiling with room to spare.
  defp sanitize_side_band(message) do
    message
    |> String.replace(~r/[[:cntrl:]]/u, " ")
    |> String.trim()
    |> String.slice(0, 512)
  end

  @doc false
  def pkt_line(data) do
    length = byte_size(data) + 4
    hex = length |> Integer.to_string(16) |> String.downcase() |> String.pad_leading(4, "0")
    hex <> data
  end

  @doc """
  Run a git service (`upload-pack` / `receive-pack`) in stateless-rpc mode
  with `input` on stdin, via a temp file so stdin EOF semantics are exact.
  Argv-only; request data never touches a shell string.
  """
  def run_git_service(command, args, input, git_protocol) do
    input_path =
      Path.join(
        System.tmp_dir!(),
        "forge-rpc-#{System.unique_integer([:positive])}-#{:erlang.phash2(self())}"
      )

    File.write!(input_path, input)

    env = if git_protocol, do: [{"GIT_PROTOCOL", git_protocol}], else: []

    try do
      # `sh` is used ONLY for stdin redirection of a server-generated temp
      # path; every request-derived value rides argv ("$@"), never the string.
      System.cmd(
        "sh",
        [
          "-c",
          ~s(exec git "$@" < "$FORGE_RPC_INPUT"),
          "sh",
          command,
          "--stateless-rpc"
        ] ++ args,
        env: env ++ [{"FORGE_RPC_INPUT", input_path}]
      )
    after
      File.rm(input_path)
    end
  end

  defp first_query(conn, key) do
    conn.query_string
    |> URI.decode_query()
    |> Map.get(key)
  end
end