feat: add components for pubsub

README.md

@@ -36,6 +36,7 @@ console.log(chunk.address.toHex())
 | `@ethersphere/core-sdk/mantaray` | `MantarayNode` — the manifest trie used for directory/collection uploads |
 | `@ethersphere/core-sdk/erasure-coding` | Reed-Solomon parity, redundancy-level tables, and the batching logic used to add parity chunks to a stream |
 | `@ethersphere/core-sdk/encryption` | Chunk-level stream cipher and XOR helpers |
+| `@ethersphere/core-sdk/bps` | BPS-lite live pub/sub (SWIP-74/87): session topics and chunk ids, signed DATA/AUTH/ROSTER/EOS chunks, WebSocket frame codecs, roster payloads, and `BpsPosition` for resuming a subscription |
 | `@ethersphere/core-sdk/stamper` | Postage stamp signing (`Stamper`, `stamp()`) and effective-capacity math |
 
 Every subpath is also re-exported from the package root (`@ethersphere/core-sdk`), so `import { Bytes } from '@ethersphere/core-sdk'` works too.

package.json

@@ -104,6 +104,16 @@
         "default": "./dist/stamper/index.cjs"
       }
     },
+    "./bps": {
+      "import": {
+        "types": "./dist/bps/index.d.ts",
+        "default": "./dist/bps/index.js"
+      },
+      "require": {
+        "types": "./dist/bps/index.d.cts",
+        "default": "./dist/bps/index.cjs"
+      }
+    },
     "./package.json": "./package.json"
   },
   "scripts": {

src/bps/chunk.ts

@@ -0,0 +1,126 @@
+import { Bytes } from '../bytes/bytes.js'
+import { numberToUint64 } from '../bytes/encoding.js'
+import { EthAddress } from '../bytes/eth-address.js'
+import { Identifier } from '../bytes/identifier.js'
+import { PrivateKey } from '../bytes/private-key.js'
+import { Span } from '../bytes/span.js'
+import { Topic } from '../bytes/topic.js'
+import { makeChunkUnchecked, MAX_PAYLOAD_SIZE } from '../chunk/cac.js'
+import { makeSingleOwnerChunk, makeSOCAddress, SingleOwnerChunk, unmarshalSingleOwnerChunk } from '../chunk/soc.js'
+import { keccak256 } from '../crypto/keccak.js'
+import { decodeBpsRoster } from './roster.js'
+
+export enum BpsKind {
+  Data = 1,
+  Auth = 2,
+  Eos = 3,
+  Roster = 4,
+}
+
+export const BPS_CHALLENGE_LENGTH = 32
+
+const SERVICE_PREFIX = new TextEncoder().encode('bps-service:v1')
+const MAX_INDEX = 2n ** 64n - 1n
+const SOC_HEADER_LENGTH = Identifier.LENGTH + 65 + Span.LENGTH
+
+function isEmptyKind(kind: BpsKind): boolean {
+  return kind === BpsKind.Auth || kind === BpsKind.Eos
+}
+
+function toIndex(index: bigint | number): bigint {
+  const value = BigInt(index)
+
+  if (value < 0n || value >= MAX_INDEX) {
+    throw new RangeError(`bps index ${value} out of range`)
+  }
+
+  return value
+}
+
+function assertEmptyKindAtZero(kind: BpsKind, index: bigint): void {
+  if (isEmptyKind(kind) && index !== 0n) {
+    throw new RangeError(`bps ${BpsKind[kind]} chunk must be at index 0, got ${index}`)
+  }
+}
+
+export function bpsSessionTopic(
+  kind: BpsKind,
+  topic: Topic | Uint8Array | string,
+  challenge: Bytes | Uint8Array | string,
+): Topic {
+  const prefix = kind === BpsKind.Data ? new Uint8Array() : Bytes.concat(SERVICE_PREFIX, new Uint8Array([kind]))
+
+  return new Topic(keccak256(Bytes.concat(prefix, new Topic(topic), new Bytes(challenge, BPS_CHALLENGE_LENGTH))))
+}
+
+export function bpsIdentifier(
+  kind: BpsKind,
+  topic: Topic | Uint8Array | string,
+  challenge: Bytes | Uint8Array | string,
+  index: bigint | number,
+): Identifier {
+  const sessionTopic = bpsSessionTopic(kind, topic, challenge)
+
+  return new Identifier(keccak256(Bytes.concat(sessionTopic, numberToUint64(toIndex(index), 'BE'))))
+}
+
+export function makeBpsChunk(
+  kind: BpsKind,
+  topic: Topic | Uint8Array | string,
+  challenge: Bytes | Uint8Array | string,
+  index: bigint | number,
+  privateKey: PrivateKey | Uint8Array | string,
+  payload: Bytes | Uint8Array | string = new Uint8Array(),
+): SingleOwnerChunk {
+  const data = typeof payload === 'string' ? Bytes.fromUtf8(payload) : new Bytes(payload)
+  const value = toIndex(index)
+  assertEmptyKindAtZero(kind, value)
+
+  if (isEmptyKind(kind) && data.length > 0) {
+    throw new RangeError(`bps ${BpsKind[kind]} chunk must have no payload`)
+  }
+
+  if (data.length > MAX_PAYLOAD_SIZE) {
+    throw new RangeError(`payload size ${data.length} exceeds maximum ${MAX_PAYLOAD_SIZE}`)
+  }
+
+  if (kind === BpsKind.Roster) {
+    decodeBpsRoster(data)
+  }
+
+  const chunk = makeChunkUnchecked(data, Span.fromBigInt(BigInt(data.length)))
+  const identifier = bpsIdentifier(kind, topic, challenge, value)
+
+  return makeSingleOwnerChunk(chunk, identifier, new PrivateKey(privateKey).toBigInt())
+}
+
+export function verifyBpsChunk(
+  frame: {
+    kind: BpsKind
+    owner: EthAddress | Uint8Array | string
+    challenge: Bytes | Uint8Array | string
+    index: bigint | number
+    soc: Bytes | Uint8Array
+  },
+  topic: Topic | Uint8Array | string,
+): SingleOwnerChunk {
+  const data = new Bytes(frame.soc).toUint8Array()
+  const index = toIndex(frame.index)
+  assertEmptyKindAtZero(frame.kind, index)
+
+  if (data.length < SOC_HEADER_LENGTH || data.length > SOC_HEADER_LENGTH + MAX_PAYLOAD_SIZE) {
+    throw new Error(`bps chunk length ${data.length} out of range`)
+  }
+
+  if (isEmptyKind(frame.kind)) {
+    const span = Span.fromSlice(data, SOC_HEADER_LENGTH - Span.LENGTH).toBigInt()
+
+    if (data.length !== SOC_HEADER_LENGTH || span !== 0n) {
+      throw new Error(`bps ${BpsKind[frame.kind]} chunk is not empty`)
+    }
+  }
+
+  const identifier = bpsIdentifier(frame.kind, topic, frame.challenge, index)
+
+  return unmarshalSingleOwnerChunk(data, makeSOCAddress(identifier, frame.owner))
+}

src/bps/frame.ts

@@ -0,0 +1,61 @@
+import { Bytes } from '../bytes/bytes.js'
+import { numberToUint64, uint64ToNumber } from '../bytes/encoding.js'
+import { EthAddress } from '../bytes/eth-address.js'
+import { Identifier } from '../bytes/identifier.js'
+import { Span } from '../bytes/span.js'
+import type { SingleOwnerChunk } from '../chunk/soc.js'
+import { BPS_CHALLENGE_LENGTH, BpsKind } from './chunk.js'
+
+const INDEX_LENGTH = 8
+const DELIVERY_HEADER_LENGTH = 1 + EthAddress.LENGTH + BPS_CHALLENGE_LENGTH + INDEX_LENGTH
+const SOC_PAYLOAD_OFFSET = Identifier.LENGTH + 65 + Span.LENGTH
+
+export interface BpsDelivery {
+  kind: BpsKind.Data | BpsKind.Eos | BpsKind.Roster
+  owner: EthAddress
+  challenge: Bytes
+  index: bigint
+  soc: Uint8Array
+  payload: Bytes
+}
+
+export function encodeBpsPublishFrame(
+  kind: BpsKind,
+  index: bigint | number,
+  soc: SingleOwnerChunk | Bytes | Uint8Array,
+): Uint8Array {
+  if (!(kind in BpsKind)) {
+    throw new Error(`unknown bps kind ${kind}`)
+  }
+
+  const data = 'data' in soc ? soc.data : new Bytes(soc).toUint8Array()
+
+  return Bytes.concat(new Uint8Array([kind]), numberToUint64(BigInt(index), 'BE'), data)
+}
+
+export function decodeBpsDeliveryFrame(frame: Bytes | Uint8Array): BpsDelivery {
+  const bytes = new Bytes(frame).toUint8Array()
+
+  if (bytes.length < DELIVERY_HEADER_LENGTH + SOC_PAYLOAD_OFFSET) {
+    throw new Error(`bps delivery frame too short: ${bytes.length} bytes`)
+  }
+
+  const kind = bytes[0]!
+
+  if (kind !== BpsKind.Data && kind !== BpsKind.Eos && kind !== BpsKind.Roster) {
+    throw new Error(`unexpected bps delivery kind ${kind}`)
+  }
+
+  const challengeOffset = 1 + EthAddress.LENGTH
+  const indexOffset = challengeOffset + BPS_CHALLENGE_LENGTH
+  const soc = bytes.slice(DELIVERY_HEADER_LENGTH)
+
+  return {
+    kind,
+    owner: new EthAddress(bytes.subarray(1, challengeOffset)),
+    challenge: new Bytes(bytes.subarray(challengeOffset, indexOffset)),
+    index: uint64ToNumber(bytes.subarray(indexOffset, DELIVERY_HEADER_LENGTH), 'BE'),
+    soc,
+    payload: new Bytes(soc.subarray(SOC_PAYLOAD_OFFSET)),
+  }
+}

src/bps/index.ts

@@ -0,0 +1,6 @@
+export { BPS_CHALLENGE_LENGTH, BpsKind, bpsIdentifier, bpsSessionTopic, makeBpsChunk, verifyBpsChunk } from './chunk.js'
+export { decodeBpsDeliveryFrame, encodeBpsPublishFrame } from './frame.js'
+export type { BpsDelivery } from './frame.js'
+export { BPS_MAX_ROSTER_SIZE, decodeBpsRoster, encodeBpsRoster } from './roster.js'
+export { BpsPosition } from './position.js'
+export type { BpsPositionOptions } from './position.js'

src/bps/position.ts

@@ -0,0 +1,103 @@
+import { Bytes } from '../bytes/bytes.js'
+import { EthAddress } from '../bytes/eth-address.js'
+import { BpsKind } from './chunk.js'
+import { decodeBpsRoster } from './roster.js'
+
+export interface BpsPositionOptions {
+  cursor?: bigint
+  roster?: Record<string, bigint>
+  rosterCursor?: bigint
+  ended?: boolean
+}
+
+export class BpsPosition {
+  private readonly owner: string
+  private readonly cursors = new Map<string, bigint>()
+  private publishers: string[] = []
+  private nextRosterIndex: bigint
+  private channelEnded: boolean
+
+  constructor(owner: EthAddress | Uint8Array | string, options: BpsPositionOptions = {}) {
+    this.owner = new EthAddress(owner).toHex()
+    this.cursors.set(this.owner, options.cursor ?? 0n)
+
+    for (const [address, cursor] of Object.entries(options.roster ?? {})) {
+      const publisher = new EthAddress(address).toHex()
+      this.publishers.push(publisher)
+      this.cursors.set(publisher, cursor)
+    }
+
+    this.nextRosterIndex = options.rosterCursor ?? 0n
+    this.channelEnded = options.ended ?? false
+  }
+
+  get ended(): boolean {
+    return this.channelEnded
+  }
+
+  get roster(): EthAddress[] {
+    return this.publishers.map(publisher => new EthAddress(publisher))
+  }
+
+  get rosterCursor(): bigint {
+    return this.nextRosterIndex
+  }
+
+  cursor(publisher: EthAddress | Uint8Array | string = this.owner): bigint {
+    return this.cursors.get(new EthAddress(publisher).toHex()) ?? 0n
+  }
+
+  accept(delivery: {
+    kind: BpsKind
+    owner: EthAddress | Uint8Array | string
+    index: bigint
+    payload: Bytes | Uint8Array
+  }): boolean {
+    switch (delivery.kind) {
+      case BpsKind.Data: {
+        const publisher = new EthAddress(delivery.owner).toHex()
+
+        if (!this.admits(publisher) || delivery.index < this.cursor(publisher)) {
+          return false
+        }
+        this.cursors.set(publisher, delivery.index + 1n)
+
+        return true
+      }
+      case BpsKind.Roster:
+        if (delivery.index < this.nextRosterIndex) {
+          return false
+        }
+        this.publishers = decodeBpsRoster(delivery.payload).map(publisher => publisher.toHex())
+        this.nextRosterIndex = delivery.index + 1n
+
+        return true
+      case BpsKind.Eos:
+        if (this.channelEnded) {
+          return false
+        }
+        this.channelEnded = true
+
+        return true
+      default:
+        return false
+    }
+  }
+
+  toQuery(): Record<string, string> {
+    const query: Record<string, string> = {
+      cursor: this.cursor().toString(),
+      rosterCursor: this.nextRosterIndex.toString(),
+    }
+
+    if (this.publishers.length > 0) {
+      query.roster = this.publishers.map(publisher => `${publisher}:${this.cursor(publisher)}`).join(',')
+    }
+
+    return query
+  }
+
+  private admits(publisher: string): boolean {
+    return publisher === this.owner || this.publishers.includes(publisher)
+  }
+}