feat(referenda-sdk): watch referenda

packages/sdk-governance/src/bounties/bounties-sdk.ts

@@ -1,12 +1,7 @@
-import { combineKeys, partitionByKey, toKeySet } from "@react-rxjs/utils"
+import { partitionEntries } from "@/util/watchEntries"
+import { combineKeys, toKeySet } from "@react-rxjs/utils"
 import { Binary, TxEvent } from "polkadot-api"
-import {
-  combineLatest,
-  distinctUntilChanged,
-  map,
-  mergeMap,
-  takeWhile,
-} from "rxjs"
+import { combineLatest, distinctUntilChanged, map } from "rxjs"
 import { getBountyDescriptions$ } from "./bounty-descriptions"
 import { BountiesSdkTypedApi, BountyWithoutDescription } from "./descriptors"
 import {
@@ -138,32 +133,8 @@ export function createBountiesSdk(typedApi: BountiesSdkTypedApi): BountiesSdk {
   }
 
   function watchBounties() {
-    const [getBountyById$, bountyKeyChanges$] = partitionByKey(
-      typedApi.query.Bounties.Bounties.watchEntries().pipe(
-        mergeMap((v) =>
-          v.deltas
-            ? [
-                ...v.deltas.deleted.map((d) => ({
-                  id: d.args[0],
-                  value: undefined,
-                })),
-                ...v.deltas.upserted.map((d) => ({
-                  id: d.args[0],
-                  value: d.value,
-                })),
-              ].sort((a, b) => a.id - b.id)
-            : [],
-        ),
-      ),
-      (res) => res.id,
-      (group$, id) =>
-        group$.pipe(
-          takeWhile(({ value }) => Boolean(value), false),
-          map((bounty) => ({
-            id,
-            bounty: bounty.value!,
-          })),
-        ),
+    const [getBountyById$, bountyKeyChanges$] = partitionEntries(
+      typedApi.query.Bounties.Bounties.watchEntries(),
     )
     const descriptions$ = getBountyDescriptions$(
       typedApi.query.Bounties.BountyDescriptions.getEntries,
@@ -184,13 +155,11 @@ export function createBountiesSdk(typedApi: BountiesSdkTypedApi): BountiesSdk {
           distinctUntilChanged(),
         ),
       ]).pipe(
-        map(([{ id, bounty }, description]) =>
-          enhanceBounty(bounty, description, id),
-        ),
+        map(([bounty, description]) => enhanceBounty(bounty, description, id)),
       )
 
     return {
-      bounties$: combineKeys(bountyIds$, getEnhancedBountyById$),
+      bounties$: combineKeys(bountyKeyChanges$, getEnhancedBountyById$),
       getBountyById$: getEnhancedBountyById$,
       bountyIds$,
     }

packages/sdk-governance/src/bounties/child-bounties-sdk.ts

@@ -1,11 +1,7 @@
-import { combineKeys, partitionByKey, toKeySet } from "@react-rxjs/utils"
-import {
-  combineLatest,
-  distinctUntilChanged,
-  map,
-  mergeMap,
-  takeWhile,
-} from "rxjs"
+import { keyedMemo } from "@/util/memo"
+import { partitionEntries } from "@/util/watchEntries"
+import { combineKeys, toKeySet } from "@react-rxjs/utils"
+import { combineLatest, distinctUntilChanged, map } from "rxjs"
 import { getBountyDescriptions$ } from "./bounty-descriptions"
 import {
   ChildBountiesSdkTypedApi,
@@ -16,7 +12,6 @@ import {
   ChildBounty,
   GenericChildBounty,
 } from "./child-sdk-types"
-import { keyedMemo } from "@/util/memo"
 
 export function createChildBountiesSdk(
   typedApi: ChildBountiesSdkTypedApi,
@@ -111,32 +106,8 @@ export function createChildBountiesSdk(
   }
 
   function watchChildBounties(parentId: number) {
-    const [getBountyById$, bountyKeyChanges$] = partitionByKey(
-      typedApi.query.ChildBounties.ChildBounties.watchEntries(parentId).pipe(
-        mergeMap((v) =>
-          v.deltas
-            ? [
-                ...v.deltas.deleted.map((d) => ({
-                  id: d.args[1],
-                  value: undefined,
-                })),
-                ...v.deltas.upserted.map((d) => ({
-                  id: d.args[1],
-                  value: d.value,
-                })),
-              ].sort((a, b) => a.id - b.id)
-            : [],
-        ),
-      ),
-      (res) => res.id,
-      (group$, id) =>
-        group$.pipe(
-          takeWhile(({ value }) => Boolean(value), false),
-          map((bounty) => ({
-            id,
-            bounty: bounty.value!,
-          })),
-        ),
+    const [getBountyById$, bountyKeyChanges$] = partitionEntries(
+      typedApi.query.ChildBounties.ChildBounties.watchEntries(parentId),
     )
     const descriptions$ = getBountyDescriptions$(
       typedApi.query.ChildBounties.ChildBountyDescriptions.getEntries,
@@ -157,9 +128,7 @@ export function createChildBountiesSdk(
           distinctUntilChanged(),
         ),
       ]).pipe(
-        map(([{ id, bounty }, description]) =>
-          enhanceBounty(bounty, description, id),
-        ),
+        map(([bounty, description]) => enhanceBounty(bounty, description, id)),
       )
 
     return {

packages/sdk-governance/src/referenda/referenda-sdk.ts

@@ -1,5 +1,8 @@
+import { Deltas, partitionEntries } from "@/util/watchEntries"
 import { blake2b } from "@noble/hashes/blake2b"
+import { combineKeys, toKeySet } from "@react-rxjs/utils"
 import { Binary, TxEvent } from "polkadot-api"
+import { map, scan } from "rxjs"
 import { getPreimageResolver } from "../preimages"
 import { originToTrack, polkadotSpenderOrigin } from "./chainConfig"
 import {
@@ -134,6 +137,62 @@ export function createReferendaSdk(
       .filter((v) => !!v)
   }
 
+  const [rawReferendumById$, referendaKeyChange$] = partitionEntries(
+    typedApi.query.Referenda.ReferendumInfoFor.watchEntries().pipe(
+      scan(
+        (acc, v) => {
+          if (!v.deltas) return { ...acc, deltas: null }
+          const deleted = v.deltas.deleted.map((v) => ({
+            ...v,
+            value: v.value.value as RawOngoingReferendum,
+          }))
+          const upserted = v.deltas.upserted
+            .map((v) => {
+              if (v.value.type === "Ongoing") {
+                acc.referendums[v.args[0]] = v.value.value
+                return {
+                  args: v.args,
+                  value: v.value.value,
+                }
+              }
+              if (v.args[0] in acc.referendums) {
+                // An Ongoing has become closed, remove from list
+                deleted.push({
+                  args: v.args,
+                  value: acc.referendums[v.args[0]],
+                })
+                delete acc.referendums[v.args[0]]
+              }
+              return null!
+            })
+            .filter(Boolean)
+
+          return {
+            referendums: acc.referendums,
+            deltas: { deleted, upserted },
+          }
+        },
+        {
+          referendums: {} as Record<number, RawOngoingReferendum>,
+          deltas: null as Deltas<RawOngoingReferendum> | null,
+        },
+      ),
+    ),
+  )
+
+  const getOngoingReferendumById$ = (id: number) =>
+    rawReferendumById$(id).pipe(
+      map((entry) => enhanceOngoingReferendum(id, entry)),
+    )
+  const ongoingReferenda$ = combineKeys(
+    referendaKeyChange$,
+    getOngoingReferendumById$,
+  )
+  const ongoingReferendaIds$ = referendaKeyChange$.pipe(
+    toKeySet(),
+    map((set) => [...set]),
+  )
+
   const getSpenderTrack: ReferendaSdk["getSpenderTrack"] = (value) => {
     const spenderOriginType = spenderOrigin(value)
     const origin: PolkadotRuntimeOriginCaller = spenderOriginType
@@ -223,6 +282,11 @@ export function createReferendaSdk(
       : null
 
   return {
+    watch: {
+      ongoingReferenda$,
+      getOngoingReferendumById$,
+      ongoingReferendaIds$,
+    },
     getOngoingReferenda,
     getSpenderTrack,
     getTrack,

packages/sdk-governance/src/referenda/sdk-types.ts

@@ -8,6 +8,7 @@ import {
   ReferendumInfo,
   TraitsScheduleDispatchTime,
 } from "./descriptors"
+import { Observable } from "rxjs"
 
 type RawOngoingReferendum = (ReferendumInfo & { type: "Ongoing" })["value"]
 
@@ -60,6 +61,11 @@ export type ReferendaTrack = Omit<
 
 export interface ReferendaSdk {
   getOngoingReferenda(): Promise<OngoingReferendum[]>
+  watch: {
+    ongoingReferenda$: Observable<Map<number, OngoingReferendum>>
+    ongoingReferendaIds$: Observable<number[]>
+    getOngoingReferendumById$: (key: number) => Observable<OngoingReferendum>
+  }
   getSpenderTrack(value: bigint): {
     origin: PolkadotRuntimeOriginCaller
     track: Promise<ReferendaTrack>

packages/sdk-governance/src/util/watchEntries.ts

@@ -0,0 +1,43 @@
+import { partitionByKey } from "@react-rxjs/utils"
+import { map, mergeMap, Observable, takeWhile } from "rxjs"
+
+export type Deltas<T> = {
+  deleted: {
+    args: Array<number>
+    value: T
+  }[]
+  upserted: {
+    args: Array<number>
+    value: T
+  }[]
+}
+export function partitionEntries<T>(
+  entries$: Observable<{
+    deltas: Deltas<T> | null
+  }>,
+) {
+  return partitionByKey(
+    entries$.pipe(
+      mergeMap((v) =>
+        v.deltas
+          ? [
+              ...v.deltas.deleted.map((d) => ({
+                id: d.args.at(-1)!,
+                value: undefined,
+              })),
+              ...v.deltas.upserted.map((d) => ({
+                id: d.args.at(-1)!,
+                value: d.value,
+              })),
+            ].sort((a, b) => a.id - b.id)
+          : [],
+      ),
+    ),
+    (res) => res.id,
+    (group$) =>
+      group$.pipe(
+        takeWhile(({ value }) => Boolean(value), false),
+        map(({ value }) => value!),
+      ),
+  )
+}