Add Notifier for running several instances on one storage

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.

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 {

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` |

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)
+	}
+}

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

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=

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].