Channel request detail page (#4541)

.gitignore

@@ -81,6 +81,7 @@ priv/openfn
 .dev.env
 .dev.override.env
 .test.override.env
+.envrc
 
 worktrees
 .docker-cache

CHANGELOG.md

@@ -17,6 +17,16 @@ and this project adheres to
 
 ### Added
 
+- Channel request detail page, reached by clicking a row in the channel history
+  table. Shows a client / destination / timing summary, a nested timing
+  visualization with per-phase breakdown and TTFB marker, foldable request and
+  response headers and body, and humanized transport and credential errors.
+  Captures richer request metadata (query string, body sizes, per-direction
+  durations, Finch phase timings) and attributes both the matched client webhook
+  auth method and the destination project credential on every proxied request.
+  Feature-gated behind experimental features.
+  [#4541](https://github.com/OpenFn/lightning/issues/4541)
+
 ### Changed
 
 ### Fixed
@@ -121,6 +131,11 @@ and this project adheres to
   [#4510](https://github.com/OpenFn/lightning/issues/4510)
 - Worker plan payload now includes `project_id` so workers can scope callbacks
   (e.g. the collections API) to the project that owns the run.
+- bumped local worker to 1.24.0
+- Channel timing fields are now stored in microseconds (previously milliseconds)
+  and request and response headers are stored as native jsonb on
+  `channel_events`. Handler adapted to Philter 0.3.0 timing map.
+  [#4541](https://github.com/OpenFn/lightning/issues/4541)
 - Bumped local worker to 1.24.0
 - Updated the Merge Sandbox UI to be cleaner, clearer, and only include changed
   workflows by default [#4651](https://github.com/OpenFn/lightning/issues/4651)

benchmarking/channels/mock_destination.exs

@@ -143,21 +143,33 @@ defmodule MockDestination.Body do
   """
 
   def generate(body_size) when body_size <= 1024 do
+    # Build the envelope once with an empty padding to measure overhead.
+    envelope =
+      Jason.encode!(%{
+        ok: true,
+        server: "mock_destination",
+        timestamp: DateTime.to_iso8601(DateTime.utc_now()),
+        padding: ""
+      })
+
+    overhead = byte_size(envelope)
+    padding_len = max(body_size - overhead, 0)
+
     json =
       Jason.encode!(%{
         ok: true,
         server: "mock_destination",
         timestamp: DateTime.to_iso8601(DateTime.utc_now()),
-        padding: String.duplicate("x", max(body_size - 80, 0))
+        padding: String.duplicate("x", padding_len)
       })
 
-    # Trim or pad to reach the target size exactly.
-    byte_size = byte_size(json)
+    # Fine-tune to the exact target size.
+    actual = byte_size(json)
 
     cond do
-      byte_size == body_size -> json
-      byte_size > body_size -> binary_part(json, 0, body_size)
-      true -> json <> String.duplicate(" ", body_size - byte_size)
+      actual == body_size -> json
+      actual > body_size -> binary_part(json, 0, body_size)
+      true -> json <> String.duplicate(" ", body_size - actual)
     end
   end
 

lib/lightning/channels.ex

@@ -194,8 +194,11 @@ defmodule Lightning.Channels do
 
     Multi.new()
     |> Multi.insert(:channel, changeset)
-    |> Multi.insert(:audit, fn %{channel: channel} ->
-      Audit.event("created", channel.id, actor, changeset)
+    |> Multi.run(:audit, fn _repo, %{channel: channel} ->
+      case Audit.event("created", channel.id, actor, changeset) do
+        :no_changes -> {:ok, :no_changes}
+        %Ecto.Changeset{} = audit_cs -> Repo.insert(audit_cs)
+      end
     end)
     |> Audit.audit_auth_method_changes(changeset, actor)
     |> Repo.transaction()
@@ -215,8 +218,11 @@ defmodule Lightning.Channels do
 
     Multi.new()
     |> Multi.update(:channel, changeset, stale_error_field: :lock_version)
-    |> Multi.insert(:audit, fn %{channel: updated} ->
-      Audit.event("updated", updated.id, actor, changeset)
+    |> Multi.run(:audit, fn _repo, %{channel: updated} ->
+      case Audit.event("updated", updated.id, actor, changeset) do
+        :no_changes -> {:ok, :no_changes}
+        %Ecto.Changeset{} = audit_cs -> Repo.insert(audit_cs)
+      end
     end)
     |> Audit.audit_auth_method_changes(changeset, actor)
     |> Repo.transaction()
@@ -441,4 +447,38 @@ defmodule Lightning.Channels do
 
     {total, nil}
   end
+
+  @doc """
+  Returns a channel request with preloads, scoped to the given project.
+
+  Returns `nil` if the request doesn't exist, belongs to a different project,
+  or the ID is not a valid UUID.
+
+  Preloads: `channel_events`, `channel`, `channel_snapshot`,
+  `client_webhook_auth_method`, and `destination_credential` (with its
+  `credential` for display).
+  """
+  @spec get_channel_request_for_project(Ecto.UUID.t(), String.t()) ::
+          ChannelRequest.t() | nil
+  def get_channel_request_for_project(project_id, request_id) do
+    case Ecto.UUID.cast(request_id) do
+      {:ok, uuid} ->
+        from(cr in ChannelRequest,
+          join: c in Channel,
+          on: cr.channel_id == c.id,
+          where: cr.id == ^uuid and c.project_id == ^project_id,
+          preload: [
+            :channel_events,
+            :channel,
+            :channel_snapshot,
+            :client_webhook_auth_method,
+            destination_credential: :credential
+          ]
+        )
+        |> Repo.one()
+
+      :error ->
+        nil
+    end
+  end
 end

lib/lightning/channels/channel_event.ex

@@ -23,15 +23,23 @@ defmodule Lightning.Channels.ChannelEvent do
           type: :destination_response | :error,
           request_method: String.t() | nil,
           request_path: String.t() | nil,
-          request_headers: String.t() | nil,
+          request_query_string: String.t() | nil,
+          request_headers: list() | nil,
           request_body_preview: String.t() | nil,
           request_body_hash: String.t() | nil,
+          request_body_size: integer() | nil,
           response_status: integer() | nil,
-          response_headers: String.t() | nil,
+          response_headers: list() | nil,
           response_body_preview: String.t() | nil,
           response_body_hash: String.t() | nil,
-          latency_ms: integer() | nil,
-          ttfb_ms: integer() | nil,
+          response_body_size: integer() | nil,
+          latency_us: integer() | nil,
+          ttfb_us: integer() | nil,
+          request_send_us: integer() | nil,
+          response_duration_us: integer() | nil,
+          queue_us: integer() | nil,
+          connect_us: integer() | nil,
+          reused_connection: boolean() | nil,
           error_message: String.t() | nil,
           inserted_at: DateTime.t()
         }
@@ -41,17 +49,25 @@ defmodule Lightning.Channels.ChannelEvent do
 
     field :request_method, :string
     field :request_path, :string
-    field :request_headers, :string
+    field :request_query_string, :string
+    field :request_headers, {:array, {:array, :string}}
     field :request_body_preview, :string
     field :request_body_hash, :string
+    field :request_body_size, :integer
 
     field :response_status, :integer
-    field :response_headers, :string
+    field :response_headers, {:array, {:array, :string}}
     field :response_body_preview, :string
     field :response_body_hash, :string
+    field :response_body_size, :integer
 
-    field :latency_ms, :integer
-    field :ttfb_ms, :integer
+    field :latency_us, :integer
+    field :ttfb_us, :integer
+    field :request_send_us, :integer
+    field :response_duration_us, :integer
+    field :queue_us, :integer
+    field :connect_us, :integer
+    field :reused_connection, :boolean
     field :error_message, :string
 
     belongs_to :channel_request, ChannelRequest
@@ -61,23 +77,52 @@ defmodule Lightning.Channels.ChannelEvent do
 
   def changeset(event, attrs) do
     event
-    |> cast(attrs, [
-      :channel_request_id,
-      :type,
-      :request_method,
-      :request_path,
-      :request_headers,
-      :request_body_preview,
-      :request_body_hash,
-      :response_status,
-      :response_headers,
-      :response_body_preview,
-      :response_body_hash,
-      :latency_ms,
-      :ttfb_ms,
-      :error_message
-    ])
+    |> cast(
+      attrs,
+      [
+        :channel_request_id,
+        :type,
+        :request_method,
+        :request_path,
+        :request_query_string,
+        :request_headers,
+        :request_body_preview,
+        :request_body_hash,
+        :request_body_size,
+        :response_status,
+        :response_headers,
+        :response_body_preview,
+        :response_body_hash,
+        :response_body_size,
+        :latency_us,
+        :ttfb_us,
+        :request_send_us,
+        :response_duration_us,
+        :queue_us,
+        :connect_us,
+        :reused_connection,
+        :error_message
+      ],
+      empty_values: []
+    )
+    |> drop_non_utf8_preview(:request_body_preview)
+    |> drop_non_utf8_preview(:response_body_preview)
     |> validate_required([:channel_request_id, :type])
     |> assoc_constraint(:channel_request)
   end
+
+  # Body previews are stored as :text (UTF-8 only). Drop the preview when the
+  # upstream returns binary content (gzip-encoded responses, images, PDFs, …)
+  # so the rest of the event — headers, hash, size, timing — still persists.
+  defp drop_non_utf8_preview(changeset, field) do
+    case get_change(changeset, field) do
+      preview when is_binary(preview) and preview != "" ->
+        if String.valid?(preview),
+          do: changeset,
+          else: put_change(changeset, field, nil)
+
+      _ ->
+        changeset
+    end
+  end
 end

lib/lightning/channels/channel_request.ex

@@ -8,13 +8,18 @@ defmodule Lightning.Channels.ChannelRequest do
   alias Lightning.Channels.Channel
   alias Lightning.Channels.ChannelEvent
   alias Lightning.Channels.ChannelSnapshot
+  alias Lightning.Projects.ProjectCredential
+  alias Lightning.Workflows.WebhookAuthMethod
 
   @type t :: %__MODULE__{
           id: Ecto.UUID.t(),
           channel_id: Ecto.UUID.t(),
           channel_snapshot_id: Ecto.UUID.t(),
           request_id: String.t(),
           client_identity: String.t() | nil,
+          client_webhook_auth_method_id: Ecto.UUID.t() | nil,
+          client_auth_type: String.t() | nil,
+          destination_credential_id: Ecto.UUID.t() | nil,
           state: :pending | :success | :failed | :timeout | :error,
           started_at: DateTime.t(),
           completed_at: DateTime.t() | nil
@@ -23,6 +28,7 @@ defmodule Lightning.Channels.ChannelRequest do
   schema "channel_requests" do
     field :request_id, :string
     field :client_identity, :string
+    field :client_auth_type, :string
 
     field :state, Ecto.Enum,
       values: [:pending, :success, :failed, :timeout, :error]
@@ -32,6 +38,8 @@ defmodule Lightning.Channels.ChannelRequest do
 
     belongs_to :channel, Channel
     belongs_to :channel_snapshot, ChannelSnapshot
+    belongs_to :client_webhook_auth_method, WebhookAuthMethod
+    belongs_to :destination_credential, ProjectCredential
 
     has_many :channel_events, ChannelEvent
   end
@@ -43,6 +51,9 @@ defmodule Lightning.Channels.ChannelRequest do
       :channel_snapshot_id,
       :request_id,
       :client_identity,
+      :client_webhook_auth_method_id,
+      :client_auth_type,
+      :destination_credential_id,
       :state,
       :started_at,
       :completed_at

lib/lightning/channels/handler.ex

@@ -60,6 +60,10 @@ defmodule Lightning.Channels.Handler do
       channel_snapshot_id: state.snapshot.id,
       request_id: state.request_id,
       client_identity: state.client_identity,
+      client_webhook_auth_method_id:
+        Map.get(state, :client_webhook_auth_method_id),
+      client_auth_type: Map.get(state, :client_auth_type),
+      destination_credential_id: Map.get(state, :destination_credential_id),
       state: :pending,
       started_at: state.started_at
     }
@@ -105,15 +109,23 @@ defmodule Lightning.Channels.Handler do
       type: event_type,
       request_method: state.request_method,
       request_path: state.request_path,
+      request_query_string: Map.get(state, :query_string),
       request_headers: encode_headers(state.request_headers),
       request_body_preview: get_in(result, [:request_observation, :preview]),
       request_body_hash: get_in(result, [:request_observation, :hash]),
+      request_body_size: get_in(result, [:request_observation, :size]),
       response_status: result.status,
       response_headers: encode_headers(Map.get(state, :response_headers)),
       response_body_preview: get_in(result, [:response_observation, :preview]),
       response_body_hash: get_in(result, [:response_observation, :hash]),
-      latency_ms: div(result.duration_us, 1000),
-      ttfb_ms: state |> Map.get(:ttfb_us) |> maybe_div(1000),
+      response_body_size: get_in(result, [:response_observation, :size]),
+      latency_us: result.timing.total_us,
+      ttfb_us: Map.get(state, :ttfb_us),
+      request_send_us: get_in(result, [:timing, :send_us]),
+      response_duration_us: get_in(result, [:timing, :recv_us]),
+      queue_us: get_in(result, [:timing, :queue_us]),
+      connect_us: get_in(result, [:timing, :connect_us]),
+      reused_connection: get_in(result, [:timing, :reused_connection]),
       error_message: if(result.error, do: classify_error(result.error))
     }
 
@@ -175,12 +187,11 @@ defmodule Lightning.Channels.Handler do
 
   defp encode_headers(nil), do: nil
 
-  # Encodes as array-of-pairs rather than a map because HTTP allows
+  # Returns as array-of-pairs rather than a map because HTTP allows
   # duplicate header keys (e.g. multiple Set-Cookie headers).
+  # Stored as native jsonb in the database.
   defp encode_headers(headers) do
-    headers
-    |> Enum.map(fn {k, v} -> [k, v] end)
-    |> Jason.encode!()
+    Enum.map(headers, fn {k, v} -> [k, v] end)
   end
 
   defp classify_error({:timeout, :connect_timeout}), do: "connect_timeout"
@@ -192,7 +203,4 @@ defmodule Lightning.Channels.Handler do
        do: Atom.to_string(reason)
 
   defp classify_error(error), do: inspect(error)
-
-  defp maybe_div(nil, _), do: nil
-  defp maybe_div(us, divisor), do: div(us, divisor)
 end

lib/lightning_web/live/channel_live/form_component.ex

@@ -132,6 +132,8 @@ defmodule LightningWeb.ChannelLive.FormComponent do
           <div class="space-y-6 bg-white">
             <.input field={f[:name]} label="Name" type="text" phx-debounce="300" />
 
+            <.input field={f[:enabled]} label="Enabled" type="toggle" />
+
             <div>
               <.input
                 field={f[:destination_url]}
@@ -176,8 +178,6 @@ defmodule LightningWeb.ChannelLive.FormComponent do
               </select>
             </div>
 
-            <.input field={f[:enabled]} label="Enabled" type="toggle" />
-
             <div>
               <div class="flex items-baseline gap-2 mb-2">
                 <p class="text-sm/6 font-medium text-slate-800">