From aa689ef2554880ea871f98959dbc06f86a52b6dd Mon Sep 17 00:00:00 2001 From: Victor Oliva Date: Thu, 13 Nov 2025 12:20:32 -0300 Subject: [PATCH] refactor storage subscription state to support multiple values --- src/pages/Storage/StorageDecode.tsx | 7 +- src/pages/Storage/StorageQuery.tsx | 53 +++-- src/pages/Storage/StorageSubscriptions.tsx | 57 +++-- src/pages/Storage/storage.state.ts | 245 +++++++++++++++++---- 4 files changed, 282 insertions(+), 80 deletions(-) diff --git a/src/pages/Storage/StorageDecode.tsx b/src/pages/Storage/StorageDecode.tsx index 6875f61..1ad956f 100644 --- a/src/pages/Storage/StorageDecode.tsx +++ b/src/pages/Storage/StorageDecode.tsx @@ -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() @@ -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, }) } diff --git a/src/pages/Storage/StorageQuery.tsx b/src/pages/Storage/StorageQuery.tsx index 8eb6610..385947e 100644 --- a/src/pages/Storage/StorageQuery.tsx +++ b/src/pages/Storage/StorageQuery.tsx @@ -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$ }) + }, }) } diff --git a/src/pages/Storage/StorageSubscriptions.tsx b/src/pages/Storage/StorageSubscriptions.tsx index c140c10..7c2f41b 100644 --- a/src/pages/Storage/StorageSubscriptions.tsx +++ b/src/pages/Storage/StorageSubscriptions.tsx @@ -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
Loading…
+ } + if (storageSubscription.status.type === "value") { return ( -
- {storageSubscription.error} +
+ + +
) } - if (!("result" in storageSubscription)) { - return
Loading…
+ + const lastValue = + storageSubscription.status.value[ + storageSubscription.status.value.length - 1 + ] + if (!lastValue) { + throw new Error("nonsense") + } + if (lastValue.result.type === "error") { + return ( +
{lastValue.result.value}
+ ) } if (storageSubscription.single) { @@ -121,7 +141,7 @@ const DecodedResultDisplay: FC<{ @@ -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) => (
- + = ({ storageSubscription }) => { - if ("error" in storageSubscription) { + if (storageSubscription.status.type === "loading") { + return
Loading…
+ } + + if (storageSubscription.status.type === "value") { return ( -
- {storageSubscription.error} +
+
) } - if (!("result" in storageSubscription)) { - return
Loading…
- } + const lastValue = + storageSubscription.status.value[ + storageSubscription.status.value.length - 1 + ] return (
- +
) } diff --git a/src/pages/Storage/storage.state.ts b/src/pages/Storage/storage.state.ts index ab27401..23fad99 100644 --- a/src/pages/Storage/storage.state.ts +++ b/src/pages/Storage/storage.state.ts @@ -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 + keyCodec?: (hash: HexString) => Observable + at?: (hash: HexString) => Observable<{ + hash: HexString | null + value: unknown + }> + value?: unknown }>() export const [removeStorageSubscription$, removeStorageSubscription] = createKeyedSignal() export const [togglePause$, toggleSubscriptionPause] = createKeyedSignal() +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 + }> +} + +const getStatus$ = ( + at: ( + hash: HexString, + ) => Observable<{ hash: HexString | null; value: unknown }>, + single: boolean, + keyCodec?: (hash: HexString) => Observable, +): Observable => { + const queryAt$ = ( + block: BlockInfo, + settled: boolean, + ): Observable => + 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 + 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 => { - 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 => { + const paused$ = togglePause$(id).pipe( + scan((v) => !v, false), + startWith(false), + ) + const status$: Observable = 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$)), ), )