refactor: split request option section builders

Author: masterkain

lib/codex_pooler/gateway/payloads/request_options.ex

@@ -0,0 +1,103 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.Continuity do
+  @moduledoc false
+
+  alias CodexPooler.Gateway.Payloads.RequestOptions.Normalization
+
+  @session_header_sources [
+    "x-codex-window-id",
+    "x-codex-session-id",
+    "session-id",
+    "x-session-id",
+    "x-session-affinity",
+    "session_id",
+    "x-codex-conversation-id"
+  ]
+
+  defstruct [
+    :accepted_turn_state,
+    :previous_response_id,
+    :response_id,
+    :session_header,
+    :session_header_source,
+    :session_key,
+    :conversation_key,
+    :owner_instance_id,
+    :bridge_owner_lease_ttl_seconds,
+    :reconnect_window_seconds,
+    :codex_session,
+    :codex_turn_id,
+    :authenticated_owner_attach
+  ]
+
+  @type t :: %__MODULE__{
+          accepted_turn_state: String.t() | nil,
+          previous_response_id: String.t() | nil,
+          response_id: String.t() | nil,
+          session_header: String.t() | nil,
+          session_header_source: String.t() | nil,
+          session_key: String.t() | nil,
+          conversation_key: String.t() | nil,
+          owner_instance_id: String.t() | nil,
+          bridge_owner_lease_ttl_seconds: pos_integer() | nil,
+          reconnect_window_seconds: non_neg_integer() | nil,
+          codex_session: term(),
+          codex_turn_id: Ecto.UUID.t() | nil,
+          authenticated_owner_attach: boolean()
+        }
+
+  @spec build(map() | keyword()) :: t()
+  def build(opts) do
+    opts = Map.new(opts)
+
+    %__MODULE__{
+      accepted_turn_state: Map.get(opts, :accepted_turn_state),
+      previous_response_id: Map.get(opts, :previous_response_id),
+      response_id: Map.get(opts, :response_id),
+      session_header: Map.get(opts, :session_header),
+      session_header_source: session_header_source(Map.get(opts, :session_header_source)),
+      session_key: Map.get(opts, :session_key),
+      conversation_key: Map.get(opts, :conversation_key),
+      owner_instance_id: Map.get(opts, :owner_instance_id),
+      bridge_owner_lease_ttl_seconds:
+        Normalization.optional_positive_integer(Map.get(opts, :bridge_owner_lease_ttl_seconds)),
+      reconnect_window_seconds:
+        Normalization.optional_non_negative_integer(Map.get(opts, :reconnect_window_seconds)),
+      codex_session: Map.get(opts, :codex_session),
+      codex_turn_id: Map.get(opts, :codex_turn_id),
+      authenticated_owner_attach: Map.get(opts, :authenticated_owner_attach, false) == true
+    }
+  end
+
+  @spec update(t(), map() | keyword()) :: t()
+  def update(%__MODULE__{} = continuity, updates) do
+    updates
+    |> Map.new()
+    |> Normalization.normalize_optional_update(
+      :bridge_owner_lease_ttl_seconds,
+      &Normalization.optional_positive_integer/1
+    )
+    |> Normalization.normalize_optional_update(
+      :reconnect_window_seconds,
+      &Normalization.optional_non_negative_integer/1
+    )
+    |> Normalization.normalize_optional_update(:session_header_source, &session_header_source/1)
+    |> then(&struct!(continuity, &1))
+  end
+
+  @spec session_header_source(term()) :: String.t() | nil
+  def session_header_source(value) when is_atom(value) do
+    value
+    |> Atom.to_string()
+    |> session_header_source()
+  end
+
+  def session_header_source(value) when is_binary(value) do
+    value = value |> String.trim() |> String.downcase()
+
+    if value in @session_header_sources do
+      value
+    end
+  end
+
+  def session_header_source(_value), do: nil
+end

lib/codex_pooler/gateway/payloads/request_options/continuity.ex

@@ -0,0 +1,67 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.FileBridgeContext do
+  @moduledoc false
+
+  alias CodexPooler.Gateway.Payloads.RequestOptions.Normalization
+
+  defstruct [
+    :operation,
+    :endpoint,
+    :route_metadata,
+    :pool_upstream_assignment_id,
+    :upstream_identity_id,
+    :defer_create_request,
+    :finalize_retry_timeout_ms,
+    :finalize_retry_interval_ms,
+    forwarded_headers: []
+  ]
+
+  @type t :: %__MODULE__{
+          operation: atom() | String.t() | nil,
+          endpoint: String.t() | nil,
+          route_metadata: map() | nil,
+          pool_upstream_assignment_id: Ecto.UUID.t() | nil,
+          upstream_identity_id: Ecto.UUID.t() | nil,
+          defer_create_request: boolean() | nil,
+          finalize_retry_timeout_ms: non_neg_integer() | nil,
+          finalize_retry_interval_ms: non_neg_integer() | nil,
+          forwarded_headers: [{String.t(), String.t()}]
+        }
+
+  @spec build(map() | keyword()) :: t()
+  def build(opts) do
+    opts = Map.new(opts)
+
+    %__MODULE__{
+      operation: Map.get(opts, :file_bridge_operation),
+      endpoint: Map.get(opts, :file_bridge_endpoint),
+      route_metadata: Map.get(opts, :file_bridge_route_metadata),
+      forwarded_headers: Normalization.forwarded_headers(Map.get(opts, :forwarded_headers, [])),
+      pool_upstream_assignment_id: Map.get(opts, :pool_upstream_assignment_id),
+      upstream_identity_id: Map.get(opts, :upstream_identity_id),
+      defer_create_request: Map.get(opts, :defer_file_create_request),
+      finalize_retry_timeout_ms:
+        Normalization.optional_non_negative_integer(Map.get(opts, :finalize_retry_timeout_ms)),
+      finalize_retry_interval_ms:
+        Normalization.optional_non_negative_integer(Map.get(opts, :finalize_retry_interval_ms))
+    }
+  end
+
+  @spec update(t(), map() | keyword()) :: t()
+  def update(%__MODULE__{} = file_bridge, updates) do
+    updates
+    |> Map.new()
+    |> Normalization.normalize_optional_update(
+      :forwarded_headers,
+      &Normalization.forwarded_headers_update/1
+    )
+    |> Normalization.normalize_optional_update(
+      :finalize_retry_timeout_ms,
+      &Normalization.optional_non_negative_integer/1
+    )
+    |> Normalization.normalize_optional_update(
+      :finalize_retry_interval_ms,
+      &Normalization.optional_non_negative_integer/1
+    )
+    |> then(&struct!(file_bridge, &1))
+  end
+end

lib/codex_pooler/gateway/payloads/request_options/file_bridge_context.ex

@@ -0,0 +1,54 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.Normalization do
+  @moduledoc false
+
+  @spec optional_positive_integer(term()) :: pos_integer() | nil
+  def optional_positive_integer(value) when is_integer(value) and value > 0, do: value
+  def optional_positive_integer(_value), do: nil
+
+  @spec optional_non_negative_integer(term()) :: non_neg_integer() | nil
+  def optional_non_negative_integer(value) when is_integer(value) and value >= 0, do: value
+  def optional_non_negative_integer(_value), do: nil
+
+  @spec forwarded_headers(term()) :: [{String.t(), String.t()}]
+  def forwarded_headers(headers) when is_list(headers) do
+    Enum.filter(headers, fn
+      {name, value} -> is_binary(name) and is_binary(value)
+      _other -> false
+    end)
+  end
+
+  def forwarded_headers(_headers), do: []
+
+  @spec forwarded_headers_update(term()) :: [{String.t(), String.t()}] | nil
+  def forwarded_headers_update(headers) do
+    case forwarded_headers(headers) do
+      [] -> nil
+      headers -> headers
+    end
+  end
+
+  @spec safe_endpoint(term()) :: String.t() | nil
+  def safe_endpoint(value) when is_binary(value) do
+    value = value |> String.trim() |> String.slice(0, 160)
+
+    if String.starts_with?(value, "/") and value != "/" do
+      value
+    else
+      nil
+    end
+  end
+
+  def safe_endpoint(_value), do: nil
+
+  @spec normalize_optional_update(map(), atom(), (term() -> term())) :: map()
+  def normalize_optional_update(updates, key, normalizer) when is_map(updates) and is_atom(key) do
+    if Map.has_key?(updates, key) do
+      case normalizer.(Map.fetch!(updates, key)) do
+        nil -> Map.delete(updates, key)
+        value -> Map.put(updates, key, value)
+      end
+    else
+      updates
+    end
+  end
+end

lib/codex_pooler/gateway/payloads/request_options/normalization.ex

@@ -0,0 +1,69 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.OpenAICompatibility do
+  @moduledoc false
+
+  alias CodexPooler.Gateway.Payloads.RequestOptions.Normalization
+
+  defstruct public_openai_responses_stream: false,
+            public_openai_chat_stream: false,
+            collect_openai_response_stream: false,
+            collect_openai_image_stream: false,
+            openai_chat_payload: nil,
+            source_endpoint: nil,
+            translated_endpoint: nil
+
+  @type t :: %__MODULE__{
+          public_openai_responses_stream: boolean(),
+          public_openai_chat_stream: boolean(),
+          collect_openai_response_stream: boolean(),
+          collect_openai_image_stream: boolean(),
+          openai_chat_payload: map() | nil,
+          source_endpoint: String.t() | nil,
+          translated_endpoint: String.t() | nil
+        }
+
+  @spec build(map() | keyword()) :: t()
+  def build(opts) do
+    opts = Map.new(opts)
+
+    %__MODULE__{
+      public_openai_responses_stream: Map.get(opts, :public_openai_responses_stream, false),
+      public_openai_chat_stream: Map.get(opts, :public_openai_chat_stream, false),
+      collect_openai_response_stream: Map.get(opts, :collect_openai_response_stream, false),
+      collect_openai_image_stream: Map.get(opts, :collect_openai_image_stream, false),
+      openai_chat_payload: Map.get(opts, :openai_chat_payload),
+      source_endpoint: Normalization.safe_endpoint(Map.get(opts, :openai_source_endpoint)),
+      translated_endpoint: Normalization.safe_endpoint(Map.get(opts, :openai_translated_endpoint))
+    }
+  end
+
+  @spec mark_origin(t(), String.t(), String.t()) :: t()
+  def mark_origin(%__MODULE__{} = compatibility, source_endpoint, translated_endpoint)
+      when is_binary(source_endpoint) and is_binary(translated_endpoint) do
+    %__MODULE__{
+      compatibility
+      | source_endpoint: Normalization.safe_endpoint(source_endpoint),
+        translated_endpoint: Normalization.safe_endpoint(translated_endpoint)
+    }
+  end
+
+  @spec metadata(t()) :: map()
+  def metadata(%__MODULE__{} = compatibility) do
+    metadata =
+      %{
+        "surface" => surface(compatibility),
+        "source_endpoint" => compatibility.source_endpoint,
+        "translated_endpoint" => compatibility.translated_endpoint
+      }
+      |> Enum.reject(fn {_key, value} -> is_nil(value) end)
+      |> Map.new()
+
+    if map_size(metadata) == 0 do
+      %{}
+    else
+      %{"openai_compatibility" => metadata}
+    end
+  end
+
+  defp surface(%__MODULE__{source_endpoint: endpoint}) when is_binary(endpoint), do: "openai_v1"
+  defp surface(_compatibility), do: nil
+end

lib/codex_pooler/gateway/payloads/request_options/openai_compatibility.ex

@@ -0,0 +1,20 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.PayloadContext do
+  @moduledoc false
+
+  defstruct [:media_upload, :forced_transcription_model]
+
+  @type t :: %__MODULE__{
+          media_upload: map() | nil,
+          forced_transcription_model: String.t() | nil
+        }
+
+  @spec build(map() | keyword()) :: t()
+  def build(opts) do
+    opts = Map.new(opts)
+
+    %__MODULE__{
+      media_upload: Map.get(opts, :media_upload),
+      forced_transcription_model: Map.get(opts, :forced_transcription_model)
+    }
+  end
+end

lib/codex_pooler/gateway/payloads/request_options/payload_context.ex

@@ -0,0 +1,39 @@
+defmodule CodexPooler.Gateway.Payloads.RequestOptions.RuntimeContext do
+  @moduledoc false
+
+  alias CodexPooler.Gateway.Payloads.RequestOptions.Normalization
+  alias CodexPooler.Gateway.RequestCompression.Metadata, as: RequestCompressionMetadata
+
+  defstruct [:now, :interrupt_reason, :gateway_debug_payload, :payload_compression]
+
+  @type t :: %__MODULE__{
+          now: DateTime.t() | nil,
+          interrupt_reason: String.t() | nil,
+          gateway_debug_payload: map() | nil,
+          payload_compression: map() | nil
+        }
+
+  @spec build(map() | keyword()) :: t()
+  def build(opts) do
+    opts = Map.new(opts)
+
+    %__MODULE__{
+      now: Map.get(opts, :now),
+      interrupt_reason: Map.get(opts, :interrupt_reason) || Map.get(opts, :reason),
+      gateway_debug_payload: Map.get(opts, :gateway_debug_payload),
+      payload_compression:
+        RequestCompressionMetadata.runtime_metadata(Map.get(opts, :payload_compression))
+    }
+  end
+
+  @spec update(t(), map() | keyword()) :: t()
+  def update(%__MODULE__{} = runtime, updates) do
+    updates
+    |> Map.new()
+    |> Normalization.normalize_optional_update(
+      :payload_compression,
+      &RequestCompressionMetadata.runtime_metadata/1
+    )
+    |> then(&struct!(runtime, &1))
+  end
+end