refactor storage subscription state to support multiple values

2025-11-13 12:20:32 -03:00
parent d4c89458f3
commit aa689ef255
4 changed files with 282 additions and 80 deletions
@@ -1,10 +1,10 @@
-import { dynamicBuilder$ } from "@/state/chains/chain.state"
 import { ActionButton } from "@/components/ActionButton"
+import { dynamicBuilder$ } from "@/state/chains/chain.state"
 import { NOTIN } from "@polkadot-api/react-builder"
 import { state, useStateObservable } from "@react-rxjs/core"
 import { createSignal } from "@react-rxjs/utils"
 import { FC } from "react"
-import { combineLatest, firstValueFrom, map, of } from "rxjs"
+import { combineLatest, firstValueFrom, map } from "rxjs"
 import { addStorageSubscription, selectedEntry$ } from "./storage.state"

 const [valueChange, setValue] = createSignal<string>()
@@ -35,13 +35,12 @@ export const StorageDecode: FC = () => {

  const submit = async () => {
    const entry = await firstValueFrom(selectedEntry$)
-    const stream = of(decoded)

    addStorageSubscription({
      name: `${entry!.pallet}.${entry!.entry}(…)`,
      args: null,
      single: true,
-      stream,
+      value: decoded,
      type: entry!.value,
    })
  }
@@ -4,8 +4,10 @@ import { BinaryEditButton } from "@/components/BinaryEditButton"
 import { CopyText } from "@/components/Copy"
 import SliderToggle from "@/components/Toggle"
 import {
+  chainClient$,
  dynamicBuilder$,
  runtimeCtx$,
+  runtimeCtxAt$,
  unsafeApi$,
 } from "@/state/chains/chain.state"
 import {
@@ -26,6 +28,7 @@ import {
  from,
  map,
  ObservedValueOf,
+  of,
  scan,
  startWith,
  switchMap,
@@ -46,32 +49,50 @@ export const StorageQuery: FC = () => {
  if (!selectedEntry) return null

  const submit = async () => {
-    const [entry, unsafeApi, keyValues, keysEnabled, keyCodec] =
-      await firstValueFrom(
-        combineLatest([
-          selectedEntry$,
-          unsafeApi$,
-          keyValues$,
-          keysEnabled$,
-          keyCodec$,
-        ]),
-      )
+    const [entry, unsafeApi, keyValues, keysEnabled] = await firstValueFrom(
+      combineLatest([selectedEntry$, unsafeApi$, keyValues$, keysEnabled$]),
+    )
    const args = keyValues.slice(0, keysEnabled)
    const storageEntry = unsafeApi.query[entry!.pallet][entry!.entry]
    const single = keyValues.length === keysEnabled
-    const stream = single
-      ? storageEntry.watchValue(...args)
-      : from(storageEntry.getEntries(...args))
-
    const argString = [...args.map(stringifyArg), ...(single ? [] : ["…"])]

    addStorageSubscription({
      name: `${entry!.pallet}.${entry!.entry}(${argString})`,
      args,
      single,
-      stream,
      type: entry!.value,
-      keyCodec: keyCodec!,
+      keyCodec: (hash) =>
+        runtimeCtxAt$(hash).pipe(
+          map(
+            (ctx) =>
+              ctx.dynamicBuilder.buildStorage(entry!.pallet, entry!.entry).keys,
+          ),
+        ),
+      at: (hash) => {
+        const value$ = from(
+          single
+            ? storageEntry.getValue(...args, {
+                at: hash,
+              })
+            : storageEntry.getEntries(...args, {
+                at: hash,
+              }),
+        )
+        const hash$ = single
+          ? chainClient$.pipe(
+              switchMap(({ chainHead }) =>
+                chainHead.storage$(hash, "hash", (ctx) =>
+                  ctx.dynamicBuilder
+                    .buildStorage(entry!.pallet, entry!.entry)
+                    .keys.enc(...args),
+                ),
+              ),
+            )
+          : of(null)
+
+        return combineLatest({ value: value$, hash: hash$ })
+      },
    })
  }

@@ -3,6 +3,7 @@ import { ViewCodec } from "@/codec-components/ViewCodec"
 import { CopyBinary } from "@/codec-components/ViewCodec/CopyBinary"
 import { ButtonGroup } from "@/components/ButtonGroup"
 import { JsonDisplay } from "@/components/JsonDisplay"
+import { cn } from "@/lib/utils"
 import { dynamicBuilder$, metadata$ } from "@/state/chains/chain.state"
 import { ViewValue } from "@/ViewValue"
 import { CodecComponentType, NOTIN } from "@polkadot-api/react-builder"
@@ -19,7 +20,6 @@ import {
  storageSubscriptionKeys$,
  toggleSubscriptionPause,
 } from "./storage.state"
-import { cn } from "@/lib/utils"

 export const StorageSubscriptions: FC = () => {
  const keys = useStateObservable(storageSubscriptionKeys$)
@@ -103,15 +103,35 @@ const DecodedResultDisplay: FC<{
  storageSubscription: StorageSubscription
  subscriptionKey: string
 }> = ({ storageSubscription, subscriptionKey }) => {
-  if ("error" in storageSubscription) {
+  if (storageSubscription.status.type === "loading") {
+    return <div className="text-sm text-foreground/50">Loading…</div>
+  }
+  if (storageSubscription.status.type === "value") {
    return (
-      <div className="text-sm text-foreground/50">
-        {storageSubscription.error}
+      <div className="max-h-[60svh] overflow-auto">
+        <PathsRoot.Provider value={subscriptionKey}>
+          <ValueDisplay
+            mode="decoded"
+            type={storageSubscription.type}
+            value={storageSubscription.status.value}
+            title={"Result"}
+          />
+        </PathsRoot.Provider>
      </div>
    )
  }
-  if (!("result" in storageSubscription)) {
-    return <div className="text-sm text-foreground/50">Loading…</div>
+
+  const lastValue =
+    storageSubscription.status.value[
+      storageSubscription.status.value.length - 1
+    ]
+  if (!lastValue) {
+    throw new Error("nonsense")
+  }
+  if (lastValue.result.type === "error") {
+    return (
+      <div className="text-sm text-foreground/50">{lastValue.result.value}</div>
+    )
  }

  if (storageSubscription.single) {
@@ -121,7 +141,7 @@ const DecodedResultDisplay: FC<{
          <ValueDisplay
            mode="decoded"
            type={storageSubscription.type}
-            value={storageSubscription.result}
+            value={lastValue.result.value.value}
            title={"Result"}
          />
        </PathsRoot.Provider>
@@ -129,7 +149,7 @@ const DecodedResultDisplay: FC<{
    )
  }

-  const values = storageSubscription.result as Array<{
+  const values = lastValue.result.value.value as Array<{
    keyArgs: unknown[]
    value: unknown
  }>
@@ -137,7 +157,7 @@ const DecodedResultDisplay: FC<{
  const renderItem = (keyArgs: unknown[], value: unknown, idx: number) => (
    <div key={idx} className={itemClasses}>
      <PathsRoot.Provider value={`${subscriptionKey}-${idx}`}>
-        <KeyDisplay value={keyArgs} keyCodec={storageSubscription.keyCodec} />
+        <KeyDisplay value={keyArgs} keyCodec={lastValue.keyCodec} />
        <ValueDisplay
          mode="decoded"
          title="Value"
@@ -175,21 +195,26 @@ const DecodedResultDisplay: FC<{
 const JsonResultDisplay: FC<{
  storageSubscription: StorageSubscription
 }> = ({ storageSubscription }) => {
-  if ("error" in storageSubscription) {
+  if (storageSubscription.status.type === "loading") {
+    return <div className="text-sm text-foreground/50">Loading…</div>
+  }
+
+  if (storageSubscription.status.type === "value") {
    return (
-      <div className="text-sm text-foreground/50">
-        {storageSubscription.error}
+      <div className="max-h-[60svh] overflow-auto">
+        <JsonDisplay src={storageSubscription.status.value} />
      </div>
    )
  }

-  if (!("result" in storageSubscription)) {
-    return <div className="text-sm text-foreground/50">Loading…</div>
-  }
+  const lastValue =
+    storageSubscription.status.value[
+      storageSubscription.status.value.length - 1
+    ]

  return (
    <div className="max-h-[60svh] overflow-auto">
-      <JsonDisplay src={storageSubscription.result} />
+      <JsonDisplay src={lastValue.result.value} />
    </div>
  )
 }
@@ -1,6 +1,11 @@
 import { bytesToString } from "@/components/BinaryInput"
 import { getHashParams } from "@/hashParams"
-import { runtimeCtx$, selectedChainChanged$ } from "@/state/chains/chain.state"
+import {
+  client$,
+  runtimeCtx$,
+  selectedChainChanged$,
+} from "@/state/chains/chain.state"
+import { BlockInfo, concatMapEager } from "@polkadot-api/observable-client"
 import { state } from "@react-rxjs/core"
 import {
  createKeyedSignal,
@@ -9,21 +14,25 @@ import {
  partitionByKey,
  toKeySet,
 } from "@react-rxjs/utils"
-import { Binary } from "polkadot-api"
+import { Binary, Enum, HexString } from "polkadot-api"
 import {
  catchError,
  combineLatest,
  concat,
  distinctUntilChanged,
+  EMPTY,
+  filter,
  ignoreElements,
  map,
  merge,
+  mergeMap,
  Observable,
  of,
  scan,
  shareReplay,
  startWith,
  switchMap,
+  take,
  takeUntil,
  withLatestFrom,
 } from "rxjs"
@@ -158,68 +167,216 @@ export const [newStorageSubscription$, addStorageSubscription] = createSignal<{
  name: string
  args: unknown[] | null
  type: number
-  keyCodec?: KeyCodec
  single: boolean
-  stream: Observable<unknown>
+  keyCodec?: (hash: HexString) => Observable<KeyCodec>
+  at?: (hash: HexString) => Observable<{
+    hash: HexString | null
+    value: unknown
+  }>
+  value?: unknown
 }>()
 export const [removeStorageSubscription$, removeStorageSubscription] =
  createKeyedSignal<string>()
 export const [togglePause$, toggleSubscriptionPause] =
  createKeyedSignal<string>()

+type StorageSubscriptionValue = {
+  height: number
+  blockHash: HexString
+  settled: boolean
+  keyCodec?: KeyCodec
+  result: Enum<{
+    success: {
+      hash: string | null
+      value: unknown
+    }
+    error: string
+  }>
+}
 export type StorageSubscription = {
  name: string
  args: unknown[] | null
  type: number
-  keyCodec?: KeyCodec
  single: boolean
  paused: boolean
  completed: boolean
-} & ({ result: unknown } | { error: string } | {})
+  status: Enum<{
+    loading: undefined
+    value: unknown
+    values: Array<StorageSubscriptionValue>
+  }>
+}
+
+const getStatus$ = (
+  at: (
+    hash: HexString,
+  ) => Observable<{ hash: HexString | null; value: unknown }>,
+  single: boolean,
+  keyCodec?: (hash: HexString) => Observable<KeyCodec>,
+): Observable<StorageSubscription["status"]> => {
+  const queryAt$ = (
+    block: BlockInfo,
+    settled: boolean,
+  ): Observable<StorageSubscriptionValue> =>
+    combineLatest([
+      at(block.hash),
+      keyCodec?.(block.hash) ?? of(undefined),
+    ]).pipe(
+      take(1),
+      map(([payload, keyCodec]) => ({
+        height: block.number,
+        blockHash: block.hash,
+        settled,
+        keyCodec,
+        result: Enum("success", payload),
+      })),
+      catchError((ex) => [
+        {
+          height: block.number,
+          blockHash: block.hash,
+          settled,
+          result: Enum("error", String(ex)),
+        },
+      ]),
+    )
+  const finalizedResults$ = client$.pipe(
+    switchMap((client) => client.finalizedBlock$),
+    mergeMap((block) => queryAt$(block, true)),
+    // Not supporting watchEntries for now. In case it's querying entries, we only take one
+    single ? (v) => v : take(1),
+  )
+  const bestResults$ = single
+    ? client$.pipe(
+        switchMap((client) => client.bestBlocks$),
+        filter((v) => v.length > 1),
+        map((blocks) => blocks[blocks.length - 1]),
+        // We need a concatMapEager here to make sure that overrides work properly
+        concatMapEager((block) => queryAt$(block, false)),
+      )
+    : EMPTY
+
+  const getValueHash = (value: StorageSubscriptionValue) =>
+    value.result.type === "success" ? value.result.value.hash : null
+
+  const values$ = merge(finalizedResults$, bestResults$).pipe(
+    scan(
+      (
+        acc: {
+          settled: StorageSubscriptionValue[]
+          settledHashes: Record<string, number>
+          unsettled: StorageSubscriptionValue[]
+        },
+        newValue,
+      ) => {
+        const newAcc = { ...acc }
+
+        if (newValue.settled) {
+          const valueHash = getValueHash(newValue)
+          if (valueHash && newAcc.settledHashes[valueHash] != null) {
+            const prevHeight = newAcc.settledHashes[valueHash]
+            if (prevHeight > newValue.height) {
+              newAcc.settledHashes = {
+                ...newAcc.settledHashes,
+                [valueHash]: newValue.height,
+              }
+              newAcc.settled = newAcc.settled.map((prevSettled) => {
+                const hash = getValueHash(prevSettled)
+                return hash === valueHash ? newValue : prevSettled
+              })
+            }
+          } else {
+            newAcc.settled = [...newAcc.settled, newValue]
+            newAcc.settled.sort((a, b) => a.height - b.height)
+            if (valueHash) {
+              newAcc.settledHashes = {
+                ...newAcc.settledHashes,
+                [valueHash]: newValue.height,
+              }
+            }
+          }
+          // Remove all unsettled blocks behind new finalized
+          newAcc.unsettled = newAcc.unsettled.filter(
+            (u) => u.height > newValue.height,
+          )
+        } else {
+          // Remove all unsettled blocks above the new unsettled one
+          const res = [
+            ...newAcc.unsettled.filter((u) => u.height > newValue.height),
+            newValue,
+          ]
+          // Prune duplicate values by hash
+          newAcc.unsettled = []
+          let prevHash: string | null = null
+          for (let i = 0; i < res.length; i++) {
+            const hash = getValueHash(res[i])
+            if (res[i].result.type === "error" || hash !== prevHash) {
+              newAcc.unsettled.push(res[i])
+              prevHash = hash
+            }
+          }
+        }
+
+        return newAcc
+      },
+      { settled: [], settledHashes: {}, unsettled: [] },
+    ),
+    map(({ settled, settledHashes, unsettled }) => [
+      ...settled,
+      ...unsettled.filter((v) => {
+        const hash = getValueHash(v)
+        // TODO edge case of a finalized value changing in one best block, then resetting on the next one
+        return !hash || settledHashes[hash] === null
+      }),
+    ]),
+  )
+
+  return values$.pipe(map((values) => Enum("values", values)))
+}

 const [getStorageSubscription$, storageSubscriptionKeyChange$] = partitionByKey(
  newStorageSubscription$,
  () => uuid(),
  (src$, id) =>
    src$.pipe(
-      switchMap(({ stream, ...props }): Observable<StorageSubscription> => {
-        const paused$ = togglePause$(id).pipe(
-          scan((v) => !v, false),
-          startWith(false),
-        )
-        const result$ = stream.pipe(
-          map((result) => ({
-            ...props,
-            result,
-          })),
-          startWith(props),
-          shareReplay(1),
-        )
-        const completed$ = concat(
-          of(false),
-          result$.pipe(ignoreElements()),
-          of(true),
-        )
-        return combineLatest([paused$, completed$, result$]).pipe(
-          map(([paused, completed, result]) => ({
-            ...result,
-            paused,
-            completed,
-          })),
-          catchError((ex) => {
-            console.error(ex)
-            return [
-              {
-                ...props,
-                paused: false,
-                completed: true,
-                error:
-                  ex instanceof Error ? `Error: ${ex.message}` : "(Errored)",
-              } satisfies StorageSubscription,
-            ]
-          }),
-        )
-      }),
+      switchMap(
+        ({
+          at,
+          value,
+          keyCodec,
+          ...props
+        }): Observable<StorageSubscription> => {
+          const paused$ = togglePause$(id).pipe(
+            scan((v) => !v, false),
+            startWith(false),
+          )
+          const status$: Observable<StorageSubscription["status"]> = value
+            ? of(Enum("value", value))
+            : getStatus$(at!, props.single, keyCodec)
+          const result$ = status$.pipe(
+            map((status) => ({
+              ...props,
+              status,
+            })),
+            startWith({
+              ...props,
+              status: Enum("loading"),
+            }),
+            shareReplay(1),
+          )
+          const completed$ = concat(
+            of(false),
+            result$.pipe(ignoreElements()),
+            of(true),
+          )
+          return combineLatest([paused$, completed$, result$]).pipe(
+            map(([paused, completed, result]) => ({
+              ...result,
+              paused,
+              completed,
+            })),
+          )
+        },
+      ),
      takeUntil(merge(removeStorageSubscription$(id), selectedChainChanged$)),
    ),
 )