-
@ hzrd149
2025-05-16 20:52:42Streams are the key to nostr
Loading events from a nostr relay is probably the most inconsistent way of loading data I've had to work with, and that's only loading from a single relay. the problem gets exponentially more complicated once you try to load events from multiple relays
Unlike HTTP nostr does not have a simple flow with timeouts built in. events are sent back one at a time and can fail at any point or have massive (10s) gaps between them
The key is to use streams. something that starts, emits any number of results, then maybe errors or completes. luckily it just so happens that JavaScript / TypeScript has a great observable stream library called RxJS
What is an observable
An
Observablein RxJS is stream a of data that are initialized lazily, which means the stream is inactive and not running until something subscribes to it```ts let stream = new Observable((observer) => { observer.next(1) observer.next(2) observer.next(3) observer.complete() })
// The stream method isn't run until its subscribed to stream.subscribe(v => console.log(v)) ```
This is super powerful and perfect for nostr because it means we don't need to manage the life-cycle of the stream. it will run when something subscribes to it and stop when unsubscribed.
Its helpful to think of this as "pulling" data. once we have created an observable we can request the data from it at any point in the future.
Pulling data from relays
We can use the lazy nature of observables to only start fetching events from a nostr relay when we need them
For example we can create an observable that will load kind 1 events from the damus relay and stream them back as they are returned from the relay
```typescript let req = new Observable((observer) => { // Create a new websocket connection when the observable is start let ws = new WebSocket('wss://relay.damus.io')
ws.onopen = () => { // Start a REQ ws.send(JSON.stringify(['REQ', 'test', {kinds: [1], limit: 20}])) }
ws.onmessage = (event) => { let message = JSON.parse(event.data) // Get the event from the message and pass it along to the subscribers if(message[0] === 'EVENT') observer.next(message[1]) }
// Cleanup subscription return () => { ws.send(JSON.stringify(['CLOSE', 'test'])) ws.close() } }) ```
But creating the observable wont do anything. we need to subscribe to it to get any events.
ts let sub = req.subscribe(event => { console.log('we got an event' event) })Cool now we are pulling events from a relay. once we are done we can stop listening to it by unsubscribing from it
ts sub.unsubscribe()This will call the cleanup method on the observable, which in turn closes the connection to the relay.
Hopefully you can see how this work, we don't have any
open,connect, ordisconnectmethods. we simply subscribe to a stream of events and it handles all the messy logic of connecting to a relayComposing and chaining observables
I've shown you how we can create a simple stream of events from a relay, but what if we want to pull from two relays?
Easy, lets make the previous example into a function that takes a relay URL
```ts function getNoteFromRelay(relay: string){ return new Observable((observer) => { let ws = new WebSocket(relay)
// ...rest of the observable...}) } ```
Then we can "merge" two of these observables into a single observable using the
mergemethod from RxJSThe
mergemethod will create a single observable that subscribes to both upstream observables and sends all the events back. Think of it as pulling events from both relays at once```ts import { merge } from 'rxjs'
const notes = merge( getNoteFromRelay('wss://relay.damus.io'), getNoteFromRelay('wss://nos.lol') )
// Subscribe to the observable to start getting data from it const sub = notes.subscribe(event => { console.log(event) })
// later unsubscribe setTimeout(() => { sub.unsubscribe() }, 10_000) ```
But now we have a problem, because we are pulling events from two relays we are getting duplicate events. to solve this we can use the
.pipemethod and thedistinctoperator from RxJS to modify our single observable to only return one version of each eventThe
.pipemethod will create a chain of observables that will each subscribe to the previous one and modify the returned values in some wayThe
distinctoperator takes a method that returns a unique identifier and filters out any duplicate values```ts import { merge, distinct } from 'rxjs'
const notes = merge( getNoteFromRelay('wss://relay.damus.io'), getNoteFromRelay('wss://nos.lol') ).pipe( // filter out events we have seen before based on the event id distinct(event => event.id) ) ```
Now we have an observable that when subscribed to will connect to two relays and return a stream of events without duplicates...
As you can see things can start getting complicated fast. but its also very powerful because we aren't managing any life-cycle code, we just subscribe and unsubscribe from an observable
Taking it to an extreme
Hopefully at this point you can see how powerful this is, we can think of almost any data loading pattern as a series of observables that pull data from upstream observables and stream it back to the original subscriber.
Here is a quick sketch of what it could look like to load user profiles. each node is an observable that "pulls" data from its child node ending with the "connect websocket" or "load from database" nodes which do the work of making a relay connection
Conclusion
All this might seem pretty simple and straight forward, but its been a long six month of learning for me. I've had to completely rethink how data and nostr events should be handled in a client and how to avoid screwing up and shooting myself in the foot with these powerful tools.
If you want to give RxJS a try I would encourage you to checkout the nostr sdk I've been building called applesauce
Its uses RxJS for pretty much everything and has the simplest and most flexible relay connection API I've seen so far (mainly no life-cycle management)
