From 1d95aa56000370f0cafe8d2cc5b15567e54a7931 Mon Sep 17 00:00:00 2001 From: Yasuhiro Matsumoto Date: Tue, 15 Sep 2026 00:23:47 +0900 Subject: [PATCH] Add Notifier for running several instances on one storage Live subscriptions are tracked in memory, so an event published to one relay process was only pushed to the clients of that process. Notifier propagates accepted events between instances sharing a storage: Notify is called for every accepted event (ephemeral ones included) and Notifications delivers the events accepted by any instance. When a Notifier is present, AddEvent hands events over to Notify and delivery to local subscribers happens only through Notifications, so there is a single path and no duplicates. It is looked up on the Relay first and on the Store second, so a backend with native change notifications can provide one while any other backend can be combined with a transport of the user's choice. Add an example running several processes on one SQLite file, kept in sync over Redis pub/sub. --- README.md | 23 ++++ add-event.go | 11 ++ examples/multi-instance/README.md | 37 +++++ examples/multi-instance/main.go | 127 +++++++++++++++++ go.mod | 3 + go.sum | 12 +- interface.go | 32 +++++ notifier_test.go | 219 ++++++++++++++++++++++++++++++ start.go | 27 +++- 9 files changed, 488 insertions(+), 3 deletions(-) create mode 100644 examples/multi-instance/README.md create mode 100644 examples/multi-instance/main.go create mode 100644 notifier_test.go diff --git a/README.md b/README.md index f35fcae..90890d7 100644 --- a/README.md +++ b/README.md @@ -32,3 +32,26 @@ The default NIP-11 document reports exactly this set. A relay implementing Note that the numbers 12, 15, 16, 20 and 33 still appear in that document for the benefit of older clients; those NIPs were merged into NIP-01 and no longer exist on their own. + +## Running several instances on one storage + +Live subscriptions are tracked in memory, so out of the box an event published +to one instance is only pushed to the clients of that instance. To run several +instances (behind a load balancer, say) against one shared storage, implement +`Notifier` on your relay or on your storage: + +```go +type Notifier interface { + Notify(context.Context, *nostr.Event) error + Notifications(context.Context) (<-chan *nostr.Event, error) +} +``` + +`Notify` is called for every accepted event; `Notifications` must deliver the +events accepted by every instance, this one included. With a `Notifier` present, +events reach local subscribers only through `Notifications`, so there is a single +delivery path and no duplicates. + +The transport is up to you — Redis pub/sub, NATS, or something native to the +storage such as PostgreSQL `LISTEN`/`NOTIFY`. It only needs to carry events from +`Notify` on one instance to `Notifications` on all of them. diff --git a/add-event.go b/add-event.go index a10ae49..3c6ff74 100644 --- a/add-event.go +++ b/add-event.go @@ -53,6 +53,17 @@ func AddEvent(ctx context.Context, relay Relay, evt *nostr.Event) (accepted bool } } + // with a Notifier, local delivery happens when the event comes back through + // Notifier.Notifications, so the event is handed over there and nowhere else. + if n := resolveNotifier(relay, store); n != nil { + if err := n.Notify(ctx, evt); err != nil { + if srv, ok := getServer(ctx); ok { + srv.Log.Errorf("failed to notify event %s: %v", evt.ID, err) + } + } + return true, "" + } + if srv, ok := getServer(ctx); ok { srv.notifyListeners(evt) } else { diff --git a/examples/multi-instance/README.md b/examples/multi-instance/README.md new file mode 100644 index 0000000..c02eed6 --- /dev/null +++ b/examples/multi-instance/README.md @@ -0,0 +1,37 @@ +# multi-instance + +Several relay processes sharing one SQLite database, kept in sync with Redis +pub/sub so that a subscription on any instance receives the events published to +any other instance. + +Each process implements `relayer.Notifier`: `Notify` publishes accepted events +on a Redis channel and `Notifications` subscribes to it. The framework then +delivers only what comes back from Redis, so every instance — including the one +the event was published to — sees each event exactly once. + +## Running + +Start Redis, then a few instances on different ports: + +``` +go build -o multi-instance . +PORT=7447 ./multi-instance & +PORT=7448 ./multi-instance & +PORT=7449 ./multi-instance & +``` + +Subscribe on one and publish on another: + +``` +nak req -k 1 --stream ws://localhost:7448 & +nak event -k 1 -c 'hello' ws://localhost:7447 +``` + +Environment: + +| variable | default | +| --- | --- | +| `SQLITE_DATABASE` | `relay.db` | +| `REDIS_URL` | `redis://127.0.0.1:6379` | +| `REDIS_CHANNEL` | `nostr:events` | +| `PORT` | `7447` | diff --git a/examples/multi-instance/main.go b/examples/multi-instance/main.go new file mode 100644 index 0000000..9373b13 --- /dev/null +++ b/examples/multi-instance/main.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + + "github.com/fiatjaf/eventstore" + "github.com/fiatjaf/eventstore/sqlite3" + "github.com/fiatjaf/relayer/v2" + "github.com/kelseyhightower/envconfig" + "github.com/nbd-wtf/go-nostr" + "github.com/redis/go-redis/v9" +) + +// Relay is a relay meant to be run as several processes sharing one SQLite +// database file. SQLite has no way to tell the other processes that a row was +// inserted, so accepted events are also published on a Redis channel and every +// instance delivers what it receives from there to its own subscribers. +type Relay struct { + SQLiteDatabase string `envconfig:"SQLITE_DATABASE" default:"relay.db"` + RedisURL string `envconfig:"REDIS_URL" default:"redis://127.0.0.1:6379"` + RedisChannel string `envconfig:"REDIS_CHANNEL" default:"nostr:events"` + Port int `envconfig:"PORT" default:"7447"` + + storage *sqlite3.SQLite3Backend + rdb *redis.Client +} + +func (r *Relay) Name() string { + return "MultiInstanceRelay" +} + +func (r *Relay) Storage(ctx context.Context) eventstore.Store { + return r.storage +} + +func (r *Relay) Init() error { + return r.rdb.Ping(context.Background()).Err() +} + +func (r *Relay) AcceptEvent(ctx context.Context, evt *nostr.Event) (bool, string) { + return true, "" +} + +// Notify publishes every accepted event, so that all instances (this one +// included) pick it up in Notifications. +func (r *Relay) Notify(ctx context.Context, evt *nostr.Event) error { + b, err := json.Marshal(evt) + if err != nil { + return err + } + return r.rdb.Publish(ctx, r.RedisChannel, b).Err() +} + +// Notifications subscribes to the channel and hands the events over to the +// server, which matches them against the subscriptions of its own clients. +func (r *Relay) Notifications(ctx context.Context) (<-chan *nostr.Event, error) { + sub := r.rdb.Subscribe(ctx, r.RedisChannel) + // wait for the subscription to be confirmed before returning, so that + // events published from now on are not missed + if _, err := sub.Receive(ctx); err != nil { + sub.Close() + return nil, err + } + + ch := make(chan *nostr.Event) + go func() { + defer close(ch) + defer sub.Close() + // sub.Channel reconnects by itself if the connection drops + msgs := sub.Channel() + for { + select { + case msg, ok := <-msgs: + if !ok { + return + } + var evt nostr.Event + if err := json.Unmarshal([]byte(msg.Payload), &evt); err != nil { + log.Printf("dropping malformed notification: %v", err) + continue + } + select { + case ch <- &evt: + case <-ctx.Done(): + return + } + case <-ctx.Done(): + return + } + } + }() + return ch, nil +} + +func (r *Relay) OnShutdown(ctx context.Context) { + r.rdb.Close() +} + +func main() { + r := Relay{} + if err := envconfig.Process("", &r); err != nil { + log.Fatalf("failed to read from env: %v", err) + } + + opts, err := redis.ParseURL(r.RedisURL) + if err != nil { + log.Fatalf("invalid REDIS_URL: %v", err) + } + r.rdb = redis.NewClient(opts) + + // WAL mode lets several processes read and write the same file, and the + // busy timeout makes writers wait for each other instead of failing. + r.storage = &sqlite3.SQLite3Backend{ + DatabaseURL: fmt.Sprintf("file:%s?_journal_mode=WAL&_busy_timeout=5000", r.SQLiteDatabase), + } + + server, err := relayer.NewServer(&r) + if err != nil { + log.Fatalf("failed to create server: %v", err) + } + if err := server.Start("0.0.0.0", r.Port); err != nil { + log.Fatalf("server terminated: %v", err) + } +} diff --git a/go.mod b/go.mod index a98acdd..09ec5c5 100644 --- a/go.mod +++ b/go.mod @@ -13,6 +13,7 @@ require ( github.com/lib/pq v1.10.9 github.com/mmcdole/gofeed v1.3.0 github.com/nbd-wtf/go-nostr v0.52.3 + github.com/redis/go-redis/v9 v9.22.0 github.com/rif/cache2go v1.0.0 github.com/rs/cors v1.11.1 github.com/stevelacy/daz v0.1.4 @@ -93,6 +94,7 @@ require ( github.com/lightningnetwork/lnd/tlv v1.3.0 // indirect github.com/lightningnetwork/lnd/tor v1.1.4 // indirect github.com/mailru/easyjson v0.9.0 // indirect + github.com/mattn/go-sqlite3 v1.14.24 // indirect github.com/miekg/dns v1.1.63 // indirect github.com/mmcdole/goxpp v1.1.1 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -119,6 +121,7 @@ require ( go.opentelemetry.io/otel v1.34.0 // indirect go.opentelemetry.io/otel/metric v1.34.0 // indirect go.opentelemetry.io/otel/trace v1.34.0 // indirect + go.uber.org/atomic v1.11.0 // indirect golang.org/x/arch v0.15.0 // indirect golang.org/x/crypto v0.55.0 // indirect golang.org/x/mod v0.40.0 // indirect diff --git a/go.sum b/go.sum index f3509be..e6e7db5 100644 --- a/go.sum +++ b/go.sum @@ -26,6 +26,10 @@ github.com/aquasecurity/esquery v0.2.0 h1:9WWXve95TE8hbm3736WB7nS6Owl8UGDeu+0jiy github.com/aquasecurity/esquery v0.2.0/go.mod h1:VU+CIFR6C+H142HHZf9RUkp4Eedpo9UrEKeCQHWf9ao= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/btcsuite/btcd v0.20.1-beta/go.mod h1:wVuoA8VJLEcwgqHBwHmzLRazpKxTv13Px/pDuV7OomQ= github.com/btcsuite/btcd v0.22.0-beta.0.20220111032746-97732e52810c/go.mod h1:tjmYdS6MLJ5/s0Fj4DbLgSbDHbEqLJrtnHecBFkdz5M= github.com/btcsuite/btcd v0.23.5-0.20231215221805-96c9fd8078fd/go.mod h1:nm3Bko6zh6bWP60UxwoT5LzdGJsQJaPo6HjduXq9p6A= @@ -372,6 +376,8 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk= github.com/puzpuzpuz/xsync/v3 v3.5.1 h1:GJYJZwO6IdxN/IKbneznS6yPkVC+c3zyY/j19c++5Fg= github.com/puzpuzpuz/xsync/v3 v3.5.1/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA= +github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= +github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rif/cache2go v1.0.0 h1:DhvZcxXvsuD9ExQ6ZO6f/sOE66OaAQIwB8Mfumap4w4= @@ -437,6 +443,8 @@ github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3i github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.etcd.io/bbolt v1.3.11 h1:yGEzV1wPz2yVCLsD8ZAiGHhHVlczyC9d1rP43/VCRJ0= go.etcd.io/bbolt v1.3.11/go.mod h1:dksAq7YMXoljX0xu6VF5DMZGbhYYoLUalEiSySYAS4I= go.etcd.io/etcd/api/v3 v3.5.7 h1:sbcmosSVesNrWOJ58ZQFitHMdncusIifYcrBfwrlJSY= @@ -471,8 +479,8 @@ go.opentelemetry.io/otel/trace v1.34.0 h1:+ouXS2V8Rd4hp4580a8q23bg0azF2nI8cqLYnC go.opentelemetry.io/otel/trace v1.34.0/go.mod h1:Svm7lSjQD7kG7KJ/MUHPVXSDGz2OX4h0M2jHBhmSfRE= go.opentelemetry.io/proto/otlp v0.9.0 h1:C0g6TWmQYvjKRnljRULLWUVJGy8Uvu0NEL/5frY2/t4= go.opentelemetry.io/proto/otlp v0.9.0/go.mod h1:1vKfU9rv61e9EVGthD1zNvUbiwPcimSsOPU9brfSHJg= -go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= -go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= diff --git a/interface.go b/interface.go index 2e3ec7a..dfa326e 100644 --- a/interface.go +++ b/interface.go @@ -45,6 +45,38 @@ type Injector interface { InjectEvents() chan nostr.Event } +// Notifier propagates accepted events between multiple relay instances that +// share the same storage, so that clients subscribed on one instance receive +// events published on another. +// +// It is looked up first on the [Relay] and then on the [eventstore.Store] +// returned by [Relay.Storage]; the first one found is used. When a Notifier is +// present, [AddEvent] no longer delivers events to local subscribers directly: +// every accepted event is passed to Notify, and only events arriving from +// Notifications are delivered. An implementation must therefore also send back +// the events it was asked to Notify about, including to the instance that +// published them. +type Notifier interface { + // Notify is called for every accepted event, after it has been saved. + // Ephemeral events are passed as well, even though they are not saved. + Notify(context.Context, *nostr.Event) error + // Notifications returns a channel delivering events accepted by any + // instance, including this one. The channel must stay open until ctx is + // done, reconnecting to the underlying transport as needed. + Notifications(context.Context) (<-chan *nostr.Event, error) +} + +// resolveNotifier returns the Notifier to use for relay, if any. +func resolveNotifier(relay Relay, store eventstore.Store) Notifier { + if n, ok := relay.(Notifier); ok { + return n + } + if n, ok := store.(Notifier); ok { + return n + } + return nil +} + // Informationer is called to compose NIP-11 response to an HTTP request // with application/nostr+json mime type. // See also [Relay.Name]. diff --git a/notifier_test.go b/notifier_test.go new file mode 100644 index 0000000..0852d11 --- /dev/null +++ b/notifier_test.go @@ -0,0 +1,219 @@ +package relayer + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/fasthttp/websocket" + "github.com/fiatjaf/eventstore/slicestore" + "github.com/nbd-wtf/go-nostr" +) + +// testBus is an in-process stand-in for a pub/sub transport (Redis, NATS, ...) +// shared by several relay instances. +type testBus struct { + mu sync.Mutex + subs []chan *nostr.Event +} + +func (b *testBus) publish(evt *nostr.Event) { + b.mu.Lock() + defer b.mu.Unlock() + for _, ch := range b.subs { + ch <- evt + } +} + +func (b *testBus) subscribe(ctx context.Context) <-chan *nostr.Event { + ch := make(chan *nostr.Event, 16) + b.mu.Lock() + b.subs = append(b.subs, ch) + b.mu.Unlock() + go func() { + <-ctx.Done() + b.mu.Lock() + defer b.mu.Unlock() + for i, c := range b.subs { + if c == ch { + b.subs = append(b.subs[:i], b.subs[i+1:]...) + break + } + } + close(ch) + }() + return ch +} + +// testNotifierRelay is a relay that propagates events over a testBus. +type testNotifierRelay struct { + testRelay + bus *testBus + notified []*nostr.Event +} + +func (r *testNotifierRelay) Notify(ctx context.Context, evt *nostr.Event) error { + r.notified = append(r.notified, evt) + r.bus.publish(evt) + return nil +} + +func (r *testNotifierRelay) Notifications(ctx context.Context) (<-chan *nostr.Event, error) { + return r.bus.subscribe(ctx), nil +} + +// testNotifierStorage is a store that implements Notifier itself, the way a +// backend with native change notifications would. +type testNotifierStorage struct { + slicestore.SliceStore + bus *testBus +} + +func (s *testNotifierStorage) Notify(ctx context.Context, evt *nostr.Event) error { + s.bus.publish(evt) + return nil +} + +func (s *testNotifierStorage) Notifications(ctx context.Context) (<-chan *nostr.Event, error) { + return s.bus.subscribe(ctx), nil +} + +func startServer(t *testing.T, relay Relay) *Server { + t.Helper() + srv, err := NewServer(relay) + if err != nil { + t.Fatal(err) + } + started := make(chan bool) + go srv.Start("127.0.0.1", 0, started) + <-started + t.Cleanup(func() { srv.Shutdown(context.TODO()) }) + return srv +} + +func subscribe(t *testing.T, conn *websocket.Conn, subID string, filter nostr.Filter) { + t.Helper() + sendJSON(t, conn, []interface{}{"REQ", subID, filter}) + if typ, _ := recvMessage(t, conn); typ != "EOSE" { + t.Fatalf("expected EOSE for %s, got %s", subID, typ) + } +} + +func expectEvent(t *testing.T, conn *websocket.Conn, subID, eventID string) { + t.Helper() + typ, raw := recvMessage(t, conn) + if typ != "EVENT" { + t.Fatalf("expected EVENT, got %s", typ) + } + var gotSub string + json.Unmarshal(raw[1], &gotSub) + if gotSub != subID { + t.Errorf("expected sub %q, got %q", subID, gotSub) + } + var evt nostr.Event + json.Unmarshal(raw[2], &evt) + if evt.ID != eventID { + t.Errorf("expected event %s, got %s", eventID, evt.ID) + } +} + +func TestNotifier_PropagatesAcrossServers(t *testing.T) { + bus := &testBus{} + store := &slicestore.SliceStore{} + + a := &testNotifierRelay{testRelay: testRelay{name: "a", storage: store}, bus: bus} + b := &testNotifierRelay{testRelay: testRelay{name: "b", storage: store}, bus: bus} + srvA := startServer(t, a) + srvB := startServer(t, b) + + connA := dialWS(t, srvA.Addr) + connB := dialWS(t, srvB.Addr) + subscribe(t, connA, "subA", nostr.Filter{Kinds: []int{1}}) + subscribe(t, connB, "subB", nostr.Filter{Kinds: []int{1}}) + + // publish on a + sk := nostr.GeneratePrivateKey() + evt := signedEvent(sk, 1, "hello from a", nostr.Tags{}) + sendJSON(t, connA, []interface{}{"EVENT", evt}) + if id, ok, reason := recvOK(t, connA); !ok || id != evt.ID { + t.Fatalf("expected OK for %s, got ok=%v id=%s reason=%q", evt.ID, ok, id, reason) + } + + // both a's and b's subscribers get it, a's exactly once + expectEvent(t, connA, "subA", evt.ID) + expectEvent(t, connB, "subB", evt.ID) + + connA.SetReadDeadline(time.Now().Add(200 * time.Millisecond)) + if _, msg, err := connA.ReadMessage(); err == nil { + t.Fatalf("a received an extra message: %s", msg) + } + + if len(a.notified) != 1 || a.notified[0].ID != evt.ID { + t.Errorf("expected a.Notify to be called once with %s, got %v", evt.ID, a.notified) + } + if len(b.notified) != 0 { + t.Errorf("expected b.Notify not to be called, got %v", b.notified) + } +} + +func TestNotifier_OnStorage(t *testing.T) { + bus := &testBus{} + store := &testNotifierStorage{bus: bus} + + srvA := startServer(t, &testRelay{name: "a", storage: store}) + srvB := startServer(t, &testRelay{name: "b", storage: store}) + + connB := dialWS(t, srvB.Addr) + subscribe(t, connB, "subB", nostr.Filter{Kinds: []int{1}}) + + connA := dialWS(t, srvA.Addr) + sk := nostr.GeneratePrivateKey() + evt := signedEvent(sk, 1, "hello from a", nostr.Tags{}) + sendJSON(t, connA, []interface{}{"EVENT", evt}) + recvOK(t, connA) + + expectEvent(t, connB, "subB", evt.ID) +} + +func TestNotifier_EphemeralEventsAreNotified(t *testing.T) { + bus := &testBus{} + store := &slicestore.SliceStore{} + a := &testNotifierRelay{testRelay: testRelay{name: "a", storage: store}, bus: bus} + srvA := startServer(t, a) + + conn := dialWS(t, srvA.Addr) + subscribe(t, conn, "sub", nostr.Filter{Kinds: []int{20001}}) + + sk := nostr.GeneratePrivateKey() + evt := signedEvent(sk, 20001, "ephemeral", nostr.Tags{}) + sendJSON(t, conn, []interface{}{"EVENT", evt}) + recvOK(t, conn) + + expectEvent(t, conn, "sub", evt.ID) + if len(a.notified) != 1 { + t.Errorf("expected Notify once, got %d", len(a.notified)) + } +} + +func TestNotifier_NotificationsErrorFailsNewServer(t *testing.T) { + r := &testFailingNotifierRelay{testRelay: testRelay{name: "a", storage: &slicestore.SliceStore{}}} + if _, err := NewServer(r); err == nil { + t.Fatal("expected NewServer to fail") + } + serversMutex.RLock() + defer serversMutex.RUnlock() + for srv := range servers { + if srv.relay == r { + t.Error("failed server left registered") + } + } +} + +type testFailingNotifierRelay struct{ testRelay } + +func (r *testFailingNotifierRelay) Notify(context.Context, *nostr.Event) error { return nil } +func (r *testFailingNotifierRelay) Notifications(context.Context) (<-chan *nostr.Event, error) { + return nil, context.Canceled +} diff --git a/start.go b/start.go index ee458e4..5397b4d 100644 --- a/start.go +++ b/start.go @@ -21,7 +21,8 @@ import ( // It can serve HTTP requests and websockets, passing control over to a relay implementation. // // To implement a relay, it is enough to satisfy [Relay] interface. Other interfaces are -// [Informationer], [CustomWebSocketHandler], [ShutdownAware] and AdvancedXxx types. +// [Informationer], [CustomWebSocketHandler], [ShutdownAware], [Injector], [Notifier] +// and AdvancedXxx types. // See their respective doc comments. // // The basic usage is to call Start or StartConf, which starts serving immediately. @@ -101,6 +102,30 @@ func NewServer(relay Relay, opts ...Option) (*Server, error) { return nil, fmt.Errorf("relay init: %w", err) } + // start listening for events accepted by other instances sharing the storage, if any + if n := resolveNotifier(relay, relay.Storage(context.Background())); n != nil { + ctx, cancel := context.WithCancel(context.Background()) + ch, err := n.Notifications(ctx) + if err != nil { + cancel() + return nil, fmt.Errorf("storage notifications: %w", err) + } + go func() { + defer cancel() + for { + select { + case event, ok := <-ch: + if !ok { + return + } + srv.notifyListeners(event) + case <-srv.done: + return + } + } + }() + } + serversMutex.Lock() servers[srv] = struct{}{} serversMutex.Unlock()