feat: decode binary/protobuf payloads with messageType (LOE)

src/lib/components/payload/nexus-operation-renderer.svelte

@@ -0,0 +1,30 @@
+<script lang="ts">
+  import PayloadCodeBlock from '$lib/components/payload/payload-code-block.svelte';
+  import type { Payload } from '$lib/types';
+  import {
+    describeNexusOperation,
+    type NexusOperationDescriptor,
+  } from '$lib/utilities/nexus-operation-registry';
+
+  interface Props {
+    payload: Payload;
+    maxHeight?: number;
+  }
+
+  let { payload, maxHeight }: Props = $props();
+
+  const descriptor: NexusOperationDescriptor | null = $derived(
+    describeNexusOperation(payload),
+  );
+</script>
+
+{#if descriptor}
+  <div class="flex flex-col gap-2">
+    <!-- TODO: Future - add clickable workflowId link when descriptor.workflowId is populated -->
+    <!-- TODO: Future - render descriptor.signalName as sub-line for signal-with-start-workflow kind -->
+    <h4 class="text-xs font-medium text-secondary">{descriptor.label}</h4>
+    <PayloadCodeBlock value={descriptor.embeddedInput} {maxHeight} />
+  </div>
+{:else}
+  <PayloadCodeBlock value={payload} {maxHeight} />
+{/if}

src/lib/components/payload/payload-decoder.svelte

@@ -12,29 +12,6 @@
   } from '$lib/utilities/decode-payload';
   import { stringifyWithBigInt } from '$lib/utilities/parse-with-big-int';
 
-  export const decodePayloadValue = async (
-    value: PotentiallyDecodable | PayloadContainingObject,
-  ): Promise<string[]> => {
-    if (isRawPayload(value)) {
-      const decodedPayloadData = await decodePayloadAndParseDataToJSON(value);
-      const stringified = stringifyWithBigInt(decodedPayloadData);
-      onDecode?.([stringified]);
-      return [stringified];
-    } else if (isRawPayloads(value)) {
-      const parsedPayloadsData = await decodePayloadsAndParseDataToJSON(value);
-      const stringified = parsedPayloadsData.map((data) =>
-        stringifyWithBigInt(data),
-      );
-      onDecode?.(stringified);
-      return stringified;
-    } else {
-      const decoded = await decodeEventAttributes(value);
-      const stringified = stringifyWithBigInt(decoded);
-      onDecode?.([stringified]);
-      return [stringified];
-    }
-  };
-
   interface Props {
     value: PotentiallyDecodable | PayloadContainingObject;
     onDecode?: (decodedPayloads: string[]) => void;
@@ -43,10 +20,51 @@
   }
 
   let { value, onDecode, children, loading }: Props = $props();
+
+  let decodedValue = $state<string[]>([]);
+  let isDecoding = $state(true);
+
+  $effect(() => {
+    const controller = new AbortController();
+    isDecoding = true;
+
+    (async () => {
+      try {
+        let result: string[];
+        if (isRawPayload(value)) {
+          const decodedPayloadData =
+            await decodePayloadAndParseDataToJSON(value);
+          if (controller.signal.aborted) return;
+          result = [stringifyWithBigInt(decodedPayloadData)];
+        } else if (isRawPayloads(value)) {
+          const parsedPayloadsData =
+            await decodePayloadsAndParseDataToJSON(value);
+          if (controller.signal.aborted) return;
+          result = parsedPayloadsData.map((data) => stringifyWithBigInt(data));
+        } else {
+          const decoded = await decodeEventAttributes(value);
+          if (controller.signal.aborted) return;
+          result = [stringifyWithBigInt(decoded)];
+        }
+        decodedValue = result;
+        isDecoding = false;
+        onDecode?.(result);
+      } catch {
+        if (!controller.signal.aborted) {
+          isDecoding = false;
+        }
+      }
+    })();
+
+    return () => {
+      controller.abort();
+      isDecoding = false;
+    };
+  });
 </script>
 
-{#await decodePayloadValue(value)}
+{#if isDecoding}
   {@render loading?.()}
-{:then decodedValue}
+{:else}
   {@render children(decodedValue)}
-{/await}
+{/if}

src/lib/components/workflow/input-and-results-payload.svelte

@@ -1,15 +1,18 @@
 <script lang="ts">
   import type { Snippet } from 'svelte';
 
+  import NexusOperationRenderer from '$lib/components/payload/nexus-operation-renderer.svelte';
   import PayloadCodeBlock from '$lib/components/payload/payload-code-block.svelte';
   import CodeBlock from '$lib/holocene/code-block.svelte';
-  import type { Payloads } from '$lib/types';
+  import type { Payload, Payloads } from '$lib/types';
+  import { isRawPayload } from '$lib/utilities/decode-payload';
   import type { CompletionEventAttributes } from '$lib/utilities/get-started-completed-and-task-failed-events';
+  import { describeNexusOperation } from '$lib/utilities/nexus-operation-registry';
 
   type Props = {
     title: string;
     titleSnippet?: Snippet;
-    content: Payloads | CompletionEventAttributes;
+    content: Payloads | CompletionEventAttributes | Payload;
     isPending?: boolean;
   };
   let {
@@ -20,6 +23,11 @@
   }: Props = $props();
 
   const MAX_HEIGHT = 300;
+
+  const isNexusPayload = $derived(
+    isRawPayload(content) &&
+      describeNexusOperation(content as Payload) !== null,
+  );
 </script>
 
 {#snippet defaultTitleSnippet()}
@@ -30,7 +38,12 @@
 
 <div class="flex w-full grow flex-col gap-2">
   {@render titleSnippet()}
-  {#if content}
+  {#if content && isNexusPayload}
+    <NexusOperationRenderer
+      payload={content as Payload}
+      maxHeight={MAX_HEIGHT}
+    />
+  {:else if content}
     <PayloadCodeBlock maxHeight={MAX_HEIGHT} value={content} />
   {:else}
     <CodeBlock

src/lib/services/data-encoder.ts

@@ -18,12 +18,28 @@ import { stringifyWithBigInt } from '$lib/utilities/parse-with-big-int';
 
 export type PotentialPayloads = { payloads: unknown[] };
 
+const delay = (ms: number, signal?: AbortSignal): Promise<void> => {
+  return new Promise((resolve, reject) => {
+    if (signal?.aborted) {
+      reject(signal.reason);
+      return;
+    }
+    const timer = setTimeout(resolve, ms);
+    signal?.addEventListener('abort', () => {
+      clearTimeout(timer);
+      reject(signal.reason);
+    });
+  });
+};
+
 export async function codeServerRequest({
   type,
   payloads,
+  signal,
 }: {
   type: 'decode' | 'encode';
   payloads: PotentialPayloads;
+  signal?: AbortSignal;
 }): Promise<Payloads> {
   const settings = page.data.settings;
   const namespace = page.params.namespace;
@@ -37,7 +53,7 @@ export async function codeServerRequest({
   const passAccessToken = getCodecPassAccessToken(settings);
   const includeCredentials = getCodecIncludeCredentials(settings);
 
-  const headers = {
+  const headers: Record<string, string> = {
     'Content-Type': 'application/json',
     'X-Namespace': namespace,
   };
@@ -64,44 +80,70 @@ export async function codeServerRequest({
         credentials: 'include' as RequestCredentials,
         method: 'POST',
         body: stringifyWithBigInt(payloads),
+        signal,
       }
     : {
         headers,
         method: 'POST',
         body: stringifyWithBigInt(payloads),
+        signal,
       };
 
-  const decoderResponse: Promise<PotentialPayloads> = fetch(
-    endpoint + `/${type}`,
-    requestOptions,
-  )
-    .then((response) => {
+  const delays = [0, 500, 1000];
+  let lastErr: unknown;
+
+  for (let attempt = 0; attempt < delays.length; attempt++) {
+    if (attempt > 0) {
+      try {
+        await delay(delays[attempt], signal);
+      } catch {
+        break;
+      }
+    }
+    if (signal?.aborted) break;
+
+    try {
+      const response = await fetch(endpoint + `/${type}`, requestOptions);
+
       if (response.ok === false) {
-        throw {
+        const err = {
           statusCode: response.status,
           statusText: response.statusText,
           response,
           message: translate(`common.${type}-failed`),
         } as NetworkError;
-      } else {
-        return response.json();
+
+        if (response.status >= 400 && response.status < 500) {
+          setLastDataEncoderFailure(err);
+          if (type === 'decode') return payloads;
+          throw err;
+        }
+
+        lastErr = err;
+        continue;
       }
-    })
-    .then((response) => {
-      setLastDataEncoderSuccess();
 
-      return response;
-    })
-    .catch((err: unknown) => {
-      setLastDataEncoderFailure(err);
-      if (type === 'decode') {
-        return payloads;
-      } else {
+      const data = await response.json();
+      setLastDataEncoderSuccess();
+      return data;
+    } catch (err: unknown) {
+      if (
+        err &&
+        typeof err === 'object' &&
+        'statusCode' in err &&
+        (err as { statusCode: number }).statusCode >= 400 &&
+        (err as { statusCode: number }).statusCode < 500
+      ) {
         throw err;
       }
-    });
+      if (signal?.aborted) break;
+      lastErr = err;
+    }
+  }
 
-  return decoderResponse;
+  setLastDataEncoderFailure(lastErr);
+  if (type === 'decode') return payloads;
+  throw lastErr;
 }
 
 export async function decodePayloadsWithCodec({