neiam /gitgud

defmodule GitGud.Federation.Workers.DeliverOutbound do
  @moduledoc """
  Signs and POSTs a local Activity to a target inbox.

  Persists per-target state in `fed_outbound_deliveries` so the UI can
  surface delivery health and retries are idempotent.
  """

  use Oban.Worker, queue: :webhooks, max_attempts: 8

  @max_attempts 8

  alias GitGud.Federation
  alias GitGud.Federation.Activity
  alias GitGud.Federation.Actor
  alias GitGud.Federation.HttpSignatures
  alias GitGud.Federation.OutboundDelivery
  alias GitGud.Repo

  def new_job(activity_id, target_inbox_url) do
    __MODULE__.new(%{
      "activity_id" => activity_id,
      "target_inbox_url" => target_inbox_url
    })
  end

  @impl Oban.Worker
  def perform(%Oban.Job{
        args: %{"activity_id" => aid, "target_inbox_url" => target_inbox_url},
        attempt: attempt
      }) do
    case Repo.get(Activity, aid) do
      nil ->
        :ok

      %Activity{} = activity ->
        deliver(activity, target_inbox_url, attempt)
    end
  end

  defp deliver(%Activity{} = activity, target_inbox_url, attempt) do
    actor = Repo.get!(Actor, activity.actor_id)

    case Federation.screen_inbound(target_inbox_url) do
      {:drop, reason} ->
        record(activity, target_inbox_url, attempt, "dropped", nil, reason)
        :ok

      :ok ->
        do_post(activity, actor, target_inbox_url, attempt)
    end
  end

  defp do_post(activity, actor, target_inbox_url, attempt) do
    body = Jason.encode!(activity.raw)
    uri = URI.parse(target_inbox_url)
    host = host_with_port(uri)
    path = path_with_query(uri)

    sig_headers =
      HttpSignatures.sign_request(
        "post",
        path,
        host,
        body,
        actor.private_key_pem,
        actor.public_key_id
      )

    headers =
      [
        {"content-type", "application/activity+json"},
        {"accept", "application/activity+json"},
        {"user-agent", "gitgud-federation"}
      ] ++ sig_headers

    case Req.post(target_inbox_url,
           headers: headers,
           body: body,
           retry: false,
           receive_timeout: 15_000
         ) do
      {:ok, %Req.Response{status: status}} when status in 200..299 ->
        record(activity, target_inbox_url, attempt, "success", status, nil)
        :ok

      {:ok, %Req.Response{status: status, body: body}} ->
        msg = if(is_binary(body), do: String.slice(body, 0, 500), else: inspect(body))
        record(activity, target_inbox_url, attempt, terminal_status(attempt, "failure"), status, msg)
        if attempt >= @max_attempts, do: {:discard, "http #{status}"}, else: {:error, "http #{status}"}

      {:error, reason} ->
        msg = inspect(reason)
        record(activity, target_inbox_url, attempt, terminal_status(attempt, "failure"), nil, msg)
        if attempt >= @max_attempts, do: {:discard, msg}, else: {:error, msg}
    end
  end

  # On the final attempt we mark the row `dead_letter` instead of just
  # `failure` so the UI / admin can filter the truly-undeliverable from
  # the in-flight retries.
  defp terminal_status(attempt, _failure) when attempt >= @max_attempts, do: "dead_letter"
  defp terminal_status(_attempt, status), do: status

  defp host_with_port(%URI{host: host, port: nil}), do: host
  defp host_with_port(%URI{host: host, port: port, scheme: "https"}) when port == 443, do: host
  defp host_with_port(%URI{host: host, port: port, scheme: "http"}) when port == 80, do: host
  defp host_with_port(%URI{host: host, port: port}), do: "#{host}:#{port}"

  defp path_with_query(%URI{path: nil} = uri), do: path_with_query(%{uri | path: "/"})
  defp path_with_query(%URI{path: p, query: nil}), do: p
  defp path_with_query(%URI{path: p, query: q}), do: p <> "?" <> q

  defp record(activity, url, attempt, status, status_code, error) do
    %OutboundDelivery{}
    |> OutboundDelivery.changeset(%{
      activity_id: activity.id,
      target_inbox_url: url,
      status: status,
      status_code: status_code,
      attempts: attempt,
      last_error: error,
      delivered_at: if(status == "success", do: DateTime.utc_now(:second))
    })
    |> Repo.insert!()
  end
end