Updates

messaging-gossip/hyper-p2p-distributed-event-bus/docs/api.md

@@ -0,0 +1,252 @@
+# API: hyper-p2p-distributed-event-bus
+
+**Protocol:** `hyper-p2p-distributed-event-bus/v1` (`EVENT_BUS_PROTOCOL`)
+
+**Export:** `HyperP2PDistributedEventBus` (default class export from `index.js`)
+
+## Overview
+
+`HyperP2PDistributedEventBus` implements a **distributed append-only event log** with **topic-based pub/sub**, **vector clocks** for causal metadata, **deduplication** by `event.id`, optional **Ed25519 signing** (`hypercore-crypto`), and **Hyperbee** persistence for replay. Peers gossip events over Hyperswarm + Protomux using the protocol id above. The wire carries **full event records** (not a `{ type, ... }` envelope).
+
+Designed for Bare/Pear: `bare-events`, `bare-fs`, `bare-crypto`, no Node builtins.
+
+## Constructor
+
+```js
+const HyperP2PDistributedEventBus = require('hyper-p2p-distributed-event-bus')
+const bus = new HyperP2PDistributedEventBus(opts)
+```
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `keyPair` | Hypercore `KeyPair` | `hypercore-crypto.keyPair()` | Identity for signing, swarm, and Hypercore |
+| `topic` | `string` \| `Buffer` \| `null` | `null` → **`'hyper-p2p-event-bus-default'`** at swarm join | Hyperswarm topic; always joined in `_initSwarm` |
+| `storageDir` | `string` | `path.join(process.cwd(), 'hyper-p2p-distributed-event-bus-storage')` | Hypercore directory |
+| `announceInterval` | `number` | `30000` | Ms between `announce` emissions when background timers enabled |
+| `expiry` | `number` | `300000` | Peer map TTL before `peer-expired` |
+| `metadata` | `object` | `{ agent: 'hyper-p2p-distributed-event-bus' }` | Merged into each published event’s `metadata` |
+| `enableSigning` | `boolean` | `true` | Sign/verify events with `hypercore-crypto` |
+| `enableBackgroundTimers` | `boolean` | `false` | Enables announce + cleanup intervals |
+
+### Read-only properties
+
+| Property | Type | Description |
+|----------|------|-------------|
+| `publicKey` | `Buffer` | Signing/swarm identity |
+| `publicKeyHex` | `string` | Hex public key; vector-clock component for this peer |
+
+## Lifecycle
+
+### `async ready() → HyperP2PDistributedEventBus`
+
+Loads vector clock and up to **1000** recent events from Hyperbee (reverse scan), joins Hyperswarm, optionally starts timers, emits `ready` and `swarm-joined`.
+
+### `async close() → void`
+
+Clears timers, destroys swarm, `bee.flush()` if available, closes Hyperbee and Hypercore, emits `closed`.
+
+## Methods
+
+### `async publish(topic, payload, metadata = {}) → Event`
+
+Builds, signs (if enabled), processes locally, gossips to all peers.
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `topic` | `string` | `'default'` if falsy | Logical channel name |
+| `payload` | `object` | `{}` | Application data; also used for subscription filters |
+| `metadata` | `object` | `{}` | Shallow-merged over constructor `metadata` |
+
+**Side effects:**
+
+1. Increments local vector clock entry for `publicKeyHex`
+2. Snapshots `vectorClock` into `event.vectorClock`
+3. `_signEvent` when `enableSigning`
+4. `_processIncomingEvent` (persist, dedup, emit, subscribe notify)
+5. `_propagateEvent` to all peer channels
+6. Emits `published`; increments `_metrics.published`
+
+**Does not throw** module-specific validation errors for empty payload.
+
+### `subscribe(topic, handler, filter = null) → unsubscribeFn`
+
+Registers `{ handler, filter }` in `subscriptions` Map.
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| `topic` | `string` | Only events with `event.topic === topic` |
+| `handler` | `(event) => void` | Called when filter matches |
+| `filter` | `object` \| `null` | Key/value equality against `event.payload` (all keys must match) |
+
+Returns function removing this subscription.
+
+Also listen via `bus.on('event', ...)` or `bus.on('event:' + topic, ...)`.
+
+### `async replay(topic = null, options = {}) → Event[]`
+
+Reads from Hyperbee (not only memory cache).
+
+| Option | Type | Default | Description |
+|--------|------|---------|-------------|
+| `from` | `number` | `0` | Skip first N matching entries |
+| `limit` | `number` | `100` | Max events returned |
+| `handler` | `function` | — | Optional per-event callback during scan |
+
+| `topic` | Hyperbee prefix |
+|---------|-----------------|
+| `null` | `event:` (all topics) |
+| `'orders'` | `event:orders:` |
+
+Emits `replay-complete` with `{ topic, count }`.
+
+### `getVectorClock() → Record<string, number>`
+
+Plain object copy of `vectorClock` Map (`peerHex → logical counter`).
+
+### `getRecentEvents(topic, limit = 50) → Event[]`
+
+Last `limit` events from in-memory `eventLog` for `topic` (may be shorter than Hyperbee history).
+
+### `getStats() → { ops: number, errors: number }`
+
+Returns `_stats` (shallow copy). Internal `_metrics` (`published`, `received`, `signed`, `verified`, `pruned`, `errors`) is updated during operation but **not** exposed via `getStats()` in current code.
+
+## Event record schema
+
+Every stored/gossiped message uses this shape:
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `id` | `string` | 32 hex chars (`randomBytes(16)`) — dedup key |
+| `topic` | `string` | Pub/sub channel |
+| `timestamp` | `number` | `Date.now()` at publish |
+| `vectorClock` | `Record<string, number>` | Snapshot at publish; merged on receive |
+| `payload` | `object` | Application body |
+| `metadata` | `object` | Agent tags + publish-time metadata |
+| `peerId` | `string` | Publisher `publicKeyHex` |
+| `signature` | `string` \| `null` | Base64 Ed25519 signature when signing enabled |
+| `issuer` | `string` \| `null` | Hex public key of signer |
+
+**Signing payload** (canonical JSON string of):
+
+`{ id, topic, timestamp, vectorClock, payload, metadata, peerId }`
+
+Verification uses `issuer` public key and `event.signature` (base64).
+
+## Vector clock behavior
+
+On **publish:**
+
+- `own = vectorClock.get(publicKeyHex) || 0`
+- Set `publicKeyHex` to `own + 1`
+- Attach `Object.fromEntries(vectorClock)` to the event
+
+On **receive** (`_mergeVectorClock`):
+
+- For each `(peer, time)` in incoming map: `local = max(local, time)`
+- Then increment **local** peer: `vectorClock.set(publicKeyHex, own + 1)`
+
+Vector clocks are **metadata for causal hints**, not a total order guarantee. Concurrent events may have incomparable clocks.
+
+## Events (EventEmitter)
+
+| Event | When | Payload |
+|-------|------|---------|
+| `ready` | `ready()` done | — |
+| `swarm-joined` | After swarm join | `topicBuf` (`Buffer`) |
+| `peer-connected` | New connection | `{ peer: peerHex }` |
+| `peer-expired` | Stale peer row | `peerHex` (string) |
+| `event` | Any accepted event | Full `Event` object |
+| `event:{topic}` | Same, namespaced | Full `Event` object |
+| `published` | After local `publish` pipeline | Full `Event` object |
+| `invalid-signature` | Verification failed | Event object (rejected) |
+| `replay-complete` | `replay()` finished | `{ topic, count }` |
+| `announce` | Background announce timer | — |
+| `error` | Subscriber throw, sign/verify failure | `Error` |
+| `closed` | `close()` | — |
+
+## Wire protocol (Protomux)
+
+Channel: **`hyper-p2p-distributed-event-bus/v1`**. Encoding: **JSON** (entire **Event** object per message).
+
+There is **no** `type` discriminator on the wire — the message body **is** the event.
+
+| message | fields | direction | behavior |
+|---------|--------|-----------|----------|
+| `Event` | `id`, `topic`, `timestamp`, `vectorClock`, `payload`, `metadata`, `peerId`, `signature`, `issuer` | bidirectional | `_processIncomingEvent`: verify → dedup → merge VC → persist → memory log → emit → notify subscribers → gossip to other peers (exclude source channel) |
+
+**Example wire payload:**
+
+```json
+{
+  "id": "a1b2c3d4e5f6071829304a5b6c7d8e9f",
+  "topic": "orders",
+  "timestamp": 1710000000000,
+  "vectorClock": {
+    "aa11...": 3,
+    "bb22...": 1
+  },
+  "payload": { "item": "laptop", "qty": 1, "priority": "high" },
+  "metadata": { "agent": "hyper-p2p-distributed-event-bus", "instance": "bus1" },
+  "peerId": "cc33...",
+  "signature": "base64...",
+  "issuer": "cc33..."
+}
+```
+
+## Hyperbee persistence
+
+| Key | Value |
+|-----|-------|
+| `vc:{peerHex}` | `number` — logical time for peer |
+| `event:{topic}:{eventId}` | Full `Event` JSON |
+
+Load path:
+
+1. Stream `vc:*` into `vectorClock` Map
+2. Reverse stream `event:*`, max 1000 entries → `eventLog` + `seenEvents`
+
+`_persistEvent` writes event key; ~10% of writes also flush all vector-clock keys.
+
+## Deduplication and memory bounds
+
+- `seenEvents` `Set` — drop if `event.id` already seen
+- Cleanup timer: if `seenEvents.size > 10000`, rebuild from last 1000 events per topic in `eventLog`
+- `eventLog` per topic: in-memory array; receives `push` on ingest (replay load uses `unshift` for historical order)
+
+## P2P gossip flow
+
+1. `publish` on peer A → local `_processIncomingEvent`
+2. `_propagateEvent` sends to all `peers` with open `msg`, except when excluding source
+3. Peer B `onmessage` → same pipeline
+4. B re-gossips to other peers (excluding channel from A)
+
+Integration tests may call `_processIncomingEvent(published)` directly if swarm propagation is slow.
+
+## Errors
+
+No module-specific error codes. Signing failures emit `invalid-signature` and increment `_metrics.errors`. See [`../_shared/ERROR_CODES.md`](../../_shared/ERROR_CODES.md).
+
+`createSwarm` in shared helper throws `topic is required` — this bus always passes a topic (default string).
+
+## Testing
+
+```bash
+cd modules/messaging-gossip/hyper-p2p-distributed-event-bus
+npm install && npm test
+```
+
+- Unit: `test/test.js` — publish/subscribe, vector clock, replay, filters
+- Integration: [`../../../real_tests/integration/event-bus-two-node.js`](../../../real_tests/integration/event-bus-two-node.js)
+
+Example (two peers):
+
+```bash
+bare examples/basic-usage.js
+```
+
+Disable signing in tests for deterministic gossip:
+
+```js
+new HyperP2PDistributedEventBus({ topic, enableSigning: false })
+```

messaging-gossip/hyper-p2p-distributed-event-bus/docs/architecture.md

@@ -0,0 +1,241 @@
+# Architecture: hyper-p2p-distributed-event-bus
+
+**Category:** Messaging & gossip
+
+**Protocol:** `hyper-p2p-distributed-event-bus/v1` (`EVENT_BUS_PROTOCOL`)
+
+**Composes with:** `hyper-p2p-gossip-mesh`, `hyper-p2p-dedup-filter`, `hyper-p2p-vector-clock`
+
+## Problem and approach
+
+Event-driven and CQRS-style decentralized apps need **append-only facts** propagated across peers with **durability** and **idempotent delivery**. This module provides:
+
+1. **Topic pub/sub** — application channels with optional payload filters
+2. **Vector clocks** — per-peer logical counters carried on every event
+3. **Gossip fan-out** — Protomux JSON messages to all connected peers
+4. **Hyperbee log** — `event:{topic}:{id}` keys for offline replay
+5. **Optional signing** — tamper detection on gossip payloads
+
+Wire format is the **event record itself** (not a typed envelope), simplifying handlers at the cost of versioning via metadata conventions.
+
+## Layer diagram
+
+```mermaid
+flowchart TB
+  subgraph app [Application]
+    Pub[publish]
+    Sub[subscribe / on event]
+  end
+  subgraph bus [HyperP2PDistributedEventBus]
+    PIPE[_processIncomingEvent]
+    VC[vectorClock Map]
+    DEDUP[seenEvents Set]
+    LOG[eventLog Map]
+    SUBS[subscriptions Map]
+    SIG[sign / verify]
+  end
+  subgraph persist [Persistence]
+    CORE[Hypercore]
+    BEE[Hyperbee]
+  end
+  subgraph net [P2P]
+    PM[Protomux EVENT_BUS_PROTOCOL]
+    HS[Hyperswarm]
+  end
+  Pub --> PIPE
+  PIPE --> VC
+  PIPE --> DEDUP
+  PIPE --> LOG
+  PIPE --> BEE
+  BEE --> CORE
+  PIPE --> SUBS
+  Sub --> SUBS
+  PIPE --> SIG
+  PIPE --> PM
+  PM --> HS
+  HS --> PM
+  PM --> PIPE
+```
+
+## Primary sequence: publish → local → gossip
+
+```mermaid
+sequenceDiagram
+  participant App
+  participant A as EventBus A
+  participant HB as Hyperbee
+  participant B as EventBus B
+
+  App->>A: publish(topic, payload)
+  A->>A: increment vectorClock[self]
+  A->>A: _signEvent (optional)
+  A->>A: _processIncomingEvent
+  A->>HB: put event:topic:id
+  A->>A: seenEvents.add(id)
+  A-->>App: event / published
+
+  A->>B: Protomux send Event JSON
+  B->>B: _verifyEvent
+  B->>B: dedup by id
+  B->>B: _mergeVectorClock
+  B->>HB: persist
+  B-->>App: event / event:topic
+  B->>B: _propagateEvent (other peers)
+```
+
+## Distributed event model
+
+Events are **immutable facts**. The log is **append-only** at the API level (no `delete` or `retract` in `index.js`). Ordering:
+
+| Mechanism | What it provides |
+|-----------|------------------|
+| `timestamp` | Wall-clock hint (publisher `Date.now()`) |
+| `vectorClock` | Causal metadata; merged with component-wise max |
+| `id` | Global dedup within a bus instance / swarm session |
+| `topic` | Partition for subscribers and replay scans |
+
+**Not provided:** strict total order across topics; conflict resolution between duplicate payloads (same `id` is impossible by construction).
+
+## Vector clock state
+
+```
+vectorClock: Map<peerPublicKeyHex, non-negative integer>
+```
+
+| Operation | Effect |
+|-----------|--------|
+| `publish` | Increment self counter before snapshot |
+| `_mergeVectorClock(incoming)` | `max(local[peer], incoming[peer])` for each peer; then increment self |
+| `_persistEvent` (10% samples) | Write `vc:{peer}` keys to Hyperbee |
+| `ready()` load | Restore all `vc:*` keys |
+
+**Interpretation:** If `VC(a) < VC(b)` component-wise, `a` may have happened-before `b`. Incomparable clocks indicate concurrent publishes. The bus does not reorder delivery based on VC — subscribers receive in network arrival order.
+
+## In-memory structures
+
+| Structure | Type | Purpose |
+|-----------|------|---------|
+| `vectorClock` | `Map<string, number>` | Causal metadata |
+| `seenEvents` | `Set<string>` | Dedup by `event.id` |
+| `eventLog` | `Map<topic, Event[]>` | Hot cache for `getRecentEvents` |
+| `subscriptions` | `Map<topic, Set<{handler, filter}>>` | Topic handlers |
+| `peers` | `Map<peerHex, { lastSeen, metadata, channel, msg }>` | Gossip targets |
+
+## Wire messages (detailed)
+
+| message | fields | direction | behavior |
+|---------|--------|-----------|----------|
+| **Event** (root object) | `id`, `topic`, `timestamp`, `vectorClock`, `payload`, `metadata`, `peerId`, `signature`, `issuer` | peer ↔ peer | `onmessage` → `_processIncomingEvent(event, channel)` |
+
+### Processing pipeline (`_processIncomingEvent`)
+
+```mermaid
+flowchart LR
+  IN[Incoming Event] --> V{enableSigning?}
+  V -->|invalid| X[invalid-signature emit return]
+  V -->|ok| D{id in seenEvents?}
+  D -->|yes| STOP[drop]
+  D -->|no| M[merge vector clock]
+  M --> P[persist Hyperbee]
+  P --> L[push eventLog]
+  L --> E[emit event + event:topic]
+  E --> N[notify subscribers]
+  N --> G[gossip to other peers]
+```
+
+| Step | Failure mode |
+|------|----------------|
+| Verify | Return early; `invalid-signature` |
+| Dedup | Silent return |
+| Persist | Await `bee.put` |
+| Subscribe | Handler exceptions → `error` event |
+| Gossip | Skip `excludeChannel`; ignore send errors |
+
+### Protomux setup
+
+```js
+protocolChannel(mux, {
+  protocol: 'hyper-p2p-distributed-event-bus/v1',
+  onopen(channel, msg) { peers.set(peerHex, { ..., channel, msg }) },
+  onmessage(event) { _processIncomingEvent(event, channel) }
+})
+```
+
+Encoding defaults to `compact-encoding` **json** in `p2p-bare.js`.
+
+## Hyperbee / Hypercore layout
+
+```
+storageDir/
+  Hypercore(keyPair, valueEncoding: json)
+    Hyperbee(keyEncoding: utf-8, valueEncoding: json)
+      vc:{peerHex}              → number
+      event:{topic}:{eventId}   → Event
+```
+
+**Replay** scans lexicographic range:
+
+- All topics: `gte: 'event:'`, `lt: 'event:\xff'`
+- One topic: `gte: 'event:orders:'`, `lt: 'event:orders:\xff'`
+
+Startup reverse load caps at **1000** events to bound memory; full history remains on disk for `replay()`.
+
+Constants: `EVENT_DB_NAME = 'event-bus'`, `VECTOR_CLOCK_PREFIX = 'vc:'`, `EVENT_PREFIX = 'event:'`, `META_PREFIX = 'meta:'` (unused), `DEFAULT_ANNOUNCE_INTERVAL = 30000`, `DEFAULT_EXPIRY = 300000`.
+
+## Subscription filter model
+
+Filter is plain object equality on **top-level** `payload` keys:
+
+```js
+subscribe('orders', handler, { priority: 'high' })
+// delivers only if event.payload.priority === 'high'
+```
+
+Missing keys in payload fail the match. No operators (AND across nested paths, regex, etc.) in core code.
+
+## Signing architecture
+
+When `enableSigning !== false`:
+
+1. **Sign** on publish with `hypercore-crypto.sign(JSON.stringify(coreFields), secretKey)`
+2. Store `signature` (base64) and `issuer` (hex public key)
+3. **Verify** on ingest with `hypercore-crypto.verify(...)`
+
+Disabling signing (`enableSigning: false`) skips verification branch (`_verifyEvent` returns true when disabled or missing signature fields).
+
+## Timers
+
+| Timer | Default | Action |
+|-------|---------|--------|
+| `announceTimer` | 30s | Emit `announce` (placeholder for heartbeat / re-announce) |
+| `cleanupTimer` | 60s | Expire stale peers; trim `seenEvents` if > 10000 |
+
+Enable with `enableBackgroundTimers: true`.
+
+## Composition
+
+| Module | Relationship |
+|--------|----------------|
+| `hyper-p2p-gossip-mesh` | Alternative fan-out topologies for large meshes |
+| `hyper-p2p-dedup-filter` | Additional dedup policies beyond `seenEvents` |
+| `hyper-p2p-vector-clock` | Standalone VC utilities; example `with-event-bus.js` |
+| `hyper-p2p-reactive-state` | Materialized view / snapshot while bus stores facts |
+
+Default swarm topic when unset: **`hyper-p2p-event-bus-default`** (hashed if string via `topicToBuffer`).
+
+Network stack ordering: [`../_shared/WAVE6_NETWORK_STACK.md`](../../_shared/WAVE6_NETWORK_STACK.md).
+
+## Scale and operations
+
+- **Fan-out:** O(peers) per event per hop; multi-hop delivery requires mesh connectivity
+- **Storage:** unbounded `event:*` keys; plan compaction externally
+- **ID space:** 128-bit random hex — collision risk negligible
+- **Trust:** signing validates publisher identity, not semantic correctness of `payload`
+
+## Testing architecture
+
+| Layer | File |
+|-------|------|
+| Unit | `test/test.js` |
+| Integration | `real_tests/integration/event-bus-two-node.js` (10s swarm wait; may inject via `_processIncomingEvent`) |
+| Example | `examples/basic-usage.js` — two buses, filter subscribe, replay |