feat(statement-sdk): subscription based statement store

examples/statement-playground/index.ts

@@ -1,13 +1,13 @@
-import { createClient } from "@polkadot-api/substrate-client"
 import {
   createStatementSdk,
+  filterPosted,
   getStatementSigner,
   stringToTopic,
+  topicFilter,
   type Statement,
 } from "@polkadot-api/sdk-statement"
-import { getWsProvider } from "polkadot-api/ws"
-import { sign, getPublicKey } from "@scure/sr25519"
 import { Binary, Blake2256 } from "@polkadot-api/substrate-bindings"
+import { getPublicKey, sign } from "@scure/sr25519"
 import { jsonSerialize, mergeUint8 } from "polkadot-api/utils"
 
 // use any key
@@ -16,33 +16,44 @@ const alice = getStatementSigner(getPublicKey(ALICE_SK), "sr25519", (p) =>
   sign(ALICE_SK, p, Blake2256(mergeUint8([ALICE_SK, p]))),
 )
 
-const client = createClient(getWsProvider("ws://127.0.0.1:9936"))
-
-const sdk = createStatementSdk(client.request)
+const sdk = createStatementSdk("ws://127.0.0.1:9936")
 
 // BUILD STATEMENT
 const stmt1: Statement = {
   decryptionKey: stringToTopic("key"),
-  priority: 1,
+  expiry: {
+    timestampSecs: (Date.now() + 3600000) / 1000,
+    sequence: 0,
+  },
   topics: [stringToTopic("1"), stringToTopic("2")],
   data: Binary.fromText("TEST 1"),
 }
 
 // SIGN AND SUBMIT STATEMENT
+const subscription = sdk
+  .getStatement$()
+  .subscribe((r) => console.log("received new statement", r))
+
 const signed1 = await alice.sign(stmt1)
 console.log(await sdk.submit(signed1))
 
 console.log(
+  "posted to 'key' with topics '1' and '2'",
   JSON.stringify(
-    await sdk.getStatements({
-      dest: stringToTopic("key"),
-      topics: [stringToTopic("1"), stringToTopic("2")],
-    }),
+    (
+      await sdk.getStatements(
+        topicFilter([stringToTopic("1"), stringToTopic("2")]),
+      )
+    ).filter(filterPosted(stringToTopic("key"))),
     jsonSerialize,
   ),
 )
 
 // GET ALL STATEMENTS (i.e. `dump`)
-console.log(JSON.stringify(await sdk.getStatements(), jsonSerialize))
+console.log(
+  "all statements",
+  JSON.stringify(await sdk.getStatements(), jsonSerialize),
+)
 
-client.destroy()
+subscription.unsubscribe()
+sdk.destroy()

examples/statement-playground/package.json

@@ -3,6 +3,10 @@
   "module": "index.ts",
   "type": "module",
   "private": true,
+  "scripts": {
+    "startNode": "zombienet --provider native spawn ./zombienet.toml",
+    "start": "bun run ./index.ts"
+  },
   "dependencies": {
     "@polkadot-api/sdk-statement": "workspace:*",
     "@polkadot-api/substrate-bindings": "^0.19.0",

packages/sdk-statement/CHANGELOG.md

@@ -2,6 +2,16 @@
 
 ## Unreleased
 
+### Changed
+
+- Change API to new subscription-based statement store
+  - `createStatementSdk` takes an endpoint URL rather than a `req` function.
+  - Update `Statement` to new spec: replaces `priority` for `expiry`.
+  - Removes `dump()`
+  - `getStatements` takes a `topicFilter` parameter
+  - Adds filtering functions for broadcasts and posted
+  - Adds `getStatement$(topicFilter)` for subscribing
+
 ## 0.4.1 2026-03-16
 
 ### Fixed

packages/sdk-statement/package.json

@@ -42,7 +42,10 @@
   "license": "MIT",
   "dependencies": {
     "@polkadot-api/substrate-bindings": "0.19.0",
+    "@polkadot-api/substrate-client": "^0.6.0",
     "@polkadot-api/utils": "^0.3.0",
-    "polkadot-api": "^2.0.0-rc.5"
+    "@polkadot-api/ws-provider": "^0.8.1",
+    "polkadot-api": "^2.0.0-rc.5",
+    "rxjs": "^7.8.2"
   }
 }

packages/sdk-statement/src/api.ts

@@ -1,26 +1,61 @@
 import { HexString } from "@polkadot-api/substrate-bindings"
-import { fromHex } from "@polkadot-api/utils"
-import { SubmitResult } from "./types"
-
-export type RequestFn = <Reply = any, Params extends Array<any> = any[]>(
-  method: string,
-  params: Params,
-) => Promise<Reply>
-
-export const getApi = (req: RequestFn) => ({
-  submit: (stmt: HexString) =>
-    req<SubmitResult | undefined, [HexString]>("statement_submit", [stmt]),
-
-  dump: () => req<HexString[], []>("statement_dump", []),
-
-  broadcasts: (matchAllTopics: HexString[]) =>
-    req<HexString[], [number[][]]>("statement_broadcastsStatement", [
-      matchAllTopics.map((v) => [...fromHex(v)]),
-    ]),
-
-  posted: (matchAllTopics: HexString[], dest: HexString) =>
-    req<HexString[], [number[][], number[]]>("statement_postedStatement", [
-      matchAllTopics.map((v) => [...fromHex(v)]),
-      [...fromHex(dest)],
-    ]),
-})
+import { createClient, UnsubscribeFn } from "@polkadot-api/substrate-client"
+import { getWsProvider } from "@polkadot-api/ws-provider"
+import { Observable } from "rxjs"
+import { StatementEvent, SubmitResult, TopicFilter } from "./types"
+
+export const getApi = (endpoint: string) => {
+  const client = createClient(getWsProvider(endpoint))
+
+  const subscribe = <T>(
+    method: string,
+    unsubscribeMethod: string,
+    params: any[],
+  ) =>
+    new Observable<T>((obs) => {
+      let unsubInner: UnsubscribeFn | null = null
+      let subId: string | null = null
+
+      const sendUnsubscribe = () => {
+        // Fire-and-forget
+        subId != null &&
+          client.request(unsubscribeMethod, [subId]).catch(() => {})
+      }
+
+      client._request(method, params, {
+        onSuccess: (res: string, follow) => {
+          subId = res
+          if (obs.closed) {
+            sendUnsubscribe()
+            return
+          }
+          unsubInner = follow(subId, {
+            next: (data: any) => obs.next(data),
+            error: (e) => obs.error(e),
+          })
+        },
+        onError(e) {
+          obs.error(e)
+        },
+      })
+
+      return () => {
+        sendUnsubscribe()
+        unsubInner?.()
+      }
+    })
+
+  return {
+    submit: (stmt: HexString) =>
+      client.request<SubmitResult>("statement_submit", [stmt]),
+
+    subscribeStatement: (topicFilter: TopicFilter) =>
+      subscribe<StatementEvent>(
+        "statement_subscribeStatement",
+        "statement_unsubscribeStatement",
+        [topicFilter],
+      ),
+
+    destroy: () => client.destroy(),
+  }
+}

packages/sdk-statement/src/codec/codec.spec.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from "vitest"
-import { statementCodec } from "./codec"
-import { fromHex } from "@polkadot-api/utils"
+import { Statement, statementCodec } from "./codec"
+import { fromHex, toHex } from "@polkadot-api/utils"
 
 describe("statement codec", () => {
   it("encodes and decodes empty statement", () => {
@@ -13,15 +13,15 @@ describe("statement codec", () => {
 
   it("throws on repeated and/or unordered keys", () => {
     expect(() => {
-      // [{ priority: 1 }, { priority: 3 }]
-      statementCodec.dec(fromHex("0x0802010000000203000000"))
+      // [{ expiry: 1 }, { expiry: 3 }]
+      statementCodec.dec(fromHex("0x0802010000000000000002030000000000000000"))
     }).toThrow("entries order")
 
     expect(() => {
       statementCodec.dec(
         fromHex(
-          // [{ channel: [0; 32] }, { priority: 1 }]
-          "0x080300000000000000000000000000000000000000000000000000000000000000000201000000",
+          // [{ channel: [0; 32] }, { expiry: 1 }]
+          "0x0803000000000000000000000000000000000000000000000000000000000000000002010000000000000000",
         ),
       )
     }).toThrow("entries order")
@@ -37,4 +37,27 @@ describe("statement codec", () => {
       )
     }).toThrow("Unexpected topic")
   })
+
+  it("encodes and decodes statement with expiry", () => {
+    const expiry: Statement["expiry"] = {
+      sequence: 0x12345678,
+      timestampSecs: 0x76543231,
+    }
+    const stmt: Statement = { expiry }
+    const encoded = statementCodec.enc(stmt)
+    const decoded = statementCodec.dec(encoded)
+    expect(decoded.expiry).toEqual(expiry)
+  })
+
+  it("encodes expiry as u64 little-endian", () => {
+    const stmt: Statement = {
+      expiry: {
+        sequence: 1,
+        timestampSecs: 0,
+      },
+    }
+    const encoded = statementCodec.enc(stmt)
+    // [compact length=1 (0x04)][variant=0x02][u64 LE = 01 00 00 00 00 00 00 00]
+    expect(toHex(encoded)).toBe("0x04020100000000000000")
+  })
 })

packages/sdk-statement/src/codec/codec.ts

@@ -7,7 +7,6 @@ import {
   SizedBytes,
   SizedHex,
   Struct,
-  u32,
   u64,
   Variant,
   Vector,
@@ -26,8 +25,15 @@ export type Proof = Enum<{
 
 export type Statement = Partial<{
   proof: Proof
+  /**
+   * @deprecated Experimental feature, may be removed/changed in future
+   *             releases.
+   */
   decryptionKey: SizedHex<32>
-  priority: number
+  expiry: {
+    timestampSecs: number
+    sequence: number
+  }
   channel: SizedHex<32>
   topics: Array<SizedHex<32>>
   data: Uint8Array
@@ -36,7 +42,7 @@ export type Statement = Partial<{
 const sortIdxs = {
   proof: 0,
   decryptionKey: 1,
-  priority: 2,
+  expiry: 2,
   channel: 3,
   topics: 4,
   topic1: 4,
@@ -60,7 +66,7 @@ const field = Variant({
     onChain: Struct({ who: bin32, blockHash: bin32, event: u64 }),
   }),
   decryptionKey: bin32,
-  priority: u32,
+  expiry: u64,
   channel: bin32,
   topic1: bin32,
   topic2: bin32,
@@ -88,6 +94,13 @@ export const statementCodec = enhanceCodec<
           stmt[k]!.forEach((v, i) => {
             statement.push(Enum(`topic${i + 1}` as `topic${1 | 2 | 3 | 4}`, v))
           })
+        } else if (k === "expiry") {
+          statement.push(
+            Enum(
+              "expiry",
+              createExpiry(stmt.expiry!.timestampSecs, stmt.expiry!.sequence),
+            ),
+          )
         } else {
           statement.push(Enum(k, stmt[k]!))
         }
@@ -104,7 +117,9 @@ export const statementCodec = enhanceCodec<
       if (idx <= maxIdx) throw new Error("Unexpected entries order")
       maxIdx = idx
 
-      if (!v.type.startsWith("topic")) {
+      if (v.type === "expiry") {
+        statement.expiry = parseExpiry(v.value)
+      } else if (!v.type.startsWith("topic")) {
         ;(statement as any)[v.type] = v.value
       } else if (v.type !== `topic${++maxTopicChecked}`) {
         throw new Error(`Unexpected ${v.type}`)
@@ -116,3 +131,16 @@ export const statementCodec = enhanceCodec<
     return statement
   },
 )
+
+const MAX_SEQ_NUMBER = 0xffffffff
+const createExpiry = (timestamp: number, sequenceNumber: number = 0) => {
+  if (sequenceNumber < 0 || sequenceNumber > MAX_SEQ_NUMBER) {
+    throw new RangeError(`sequenceNumber must be 0-${MAX_SEQ_NUMBER}`)
+  }
+  return (BigInt(Math.floor(timestamp)) << 32n) | BigInt(sequenceNumber)
+}
+
+const parseExpiry = (expiry: bigint) => ({
+  timestampSecs: Number(expiry >> 32n),
+  sequence: Number(expiry & 0xffffffffn),
+})

packages/sdk-statement/src/filters.ts

@@ -0,0 +1,12 @@
+import { SizedHex } from "polkadot-api"
+import { Statement } from "./codec"
+import { TopicFilter } from "./types"
+
+export const topicFilter = (topics: Array<SizedHex<32>>): TopicFilter => ({
+  matchAll: topics,
+})
+
+export const filterBroadcasts = () => (stmt: Statement) =>
+  stmt.decryptionKey === undefined
+export const filterPosted = (dest: SizedHex<32>) => (stmt: Statement) =>
+  stmt.decryptionKey === dest

packages/sdk-statement/src/index.ts

@@ -1,5 +1,6 @@
 export * from "./codec"
-export * from "./statement-sdk"
+export * from "./filters"
 export * from "./signer"
-export type { SubmitResult } from "./types"
+export * from "./statement-sdk"
+export type { StatementEvent, SubmitResult, TopicFilter } from "./types"
 export { stringToTopic } from "./utils"