Add Notifier, implemented for PostgreSQL with LISTEN/NOTIFY

postgresql/notify.go

@@ -0,0 +1,111 @@
+package postgresql
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/fiatjaf/eventstore"
+	"github.com/lib/pq"
+	"github.com/nbd-wtf/go-nostr"
+)
+
+const (
+	defaultNotifyChannel = "nostr_events"
+
+	// pg_notify refuses payloads above this size, so bigger events are
+	// announced by id and fetched back from the table by the receiver.
+	maxNotifyPayload = 8000
+)
+
+var _ eventstore.Notifier = (*PostgresBackend)(nil)
+
+func (b *PostgresBackend) notifyChannel() string {
+	if b.NotifyChannel != "" {
+		return b.NotifyChannel
+	}
+	return defaultNotifyChannel
+}
+
+// Notify sends evt on the NotifyChannel with NOTIFY, so that every process
+// listening through Notifications receives it.
+func (b *PostgresBackend) Notify(ctx context.Context, evt *nostr.Event) error {
+	payload, err := json.Marshal(evt)
+	if err != nil {
+		return err
+	}
+	if len(payload) > maxNotifyPayload {
+		payload = []byte(evt.ID)
+	}
+	_, err = b.DB.ExecContext(ctx, `SELECT pg_notify($1, $2)`, b.notifyChannel(), string(payload))
+	return err
+}
+
+// Notifications listens on the NotifyChannel and delivers every event sent
+// with Notify by any process connected to the same database. It needs
+// DatabaseURL, since LISTEN takes over a dedicated connection.
+func (b *PostgresBackend) Notifications(ctx context.Context) (<-chan *nostr.Event, error) {
+	if b.DatabaseURL == "" {
+		return nil, fmt.Errorf("DatabaseURL is required to listen for notifications")
+	}
+
+	listener := pq.NewListener(b.DatabaseURL, 10*time.Second, time.Minute, nil)
+	if err := listener.Listen(b.notifyChannel()); err != nil {
+		listener.Close()
+		return nil, err
+	}
+
+	ch := make(chan *nostr.Event)
+	go func() {
+		defer close(ch)
+		defer listener.Close()
+		for {
+			select {
+			case n, ok := <-listener.Notify:
+				if !ok {
+					return
+				}
+				if n == nil {
+					// the listener reconnected; notifications sent meanwhile are lost
+					continue
+				}
+				evt := b.decodeNotification(ctx, n.Extra)
+				if evt == nil {
+					continue
+				}
+				select {
+				case ch <- evt:
+				case <-ctx.Done():
+					return
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return ch, nil
+}
+
+// decodeNotification turns a payload sent by Notify back into an event: either
+// the event itself or, when it was too big for pg_notify, its id.
+func (b *PostgresBackend) decodeNotification(ctx context.Context, payload string) *nostr.Event {
+	if len(payload) > 0 && payload[0] == '{' {
+		var evt nostr.Event
+		if err := json.Unmarshal([]byte(payload), &evt); err != nil {
+			return nil
+		}
+		return &evt
+	}
+
+	// too big to be sent inline; ephemeral events are not stored, so those are lost
+	ch, err := b.QueryEvents(ctx, nostr.Filter{IDs: []string{payload}, Limit: 1})
+	if err != nil {
+		return nil
+	}
+	var evt *nostr.Event
+	for e := range ch {
+		evt = e
+	}
+	return evt
+}

postgresql/postgresql.go

@@ -9,17 +9,18 @@ import (
 type PostgresBackend struct {
 	sync.Mutex
 	*sqlx.DB
-	DatabaseURL              string
-	QueryLimit               int
-	QueryIDsLimit            int
-	QueryAuthorsLimit        int
-	QueryKindsLimit          int
-	QueryTagsLimit           int
-	KeepRecentEvents         bool
-	FullTextSearchConfig     string // text search configuration for to_tsvector/to_tsquery, defaults to "simple"
-	FullTextSearchMaxLength  int    // maximum content length for full-text search, 0 means no limit
-	FullTextSearchColumn     string // column to search in, defaults to "content"
-	SubstringSearch          bool   // match NIP-50 search as an ILIKE '%q%' substring instead of tsvector full-text; see queryEventsSql
+	DatabaseURL             string
+	QueryLimit              int
+	QueryIDsLimit           int
+	QueryAuthorsLimit       int
+	QueryKindsLimit         int
+	QueryTagsLimit          int
+	KeepRecentEvents        bool
+	FullTextSearchConfig    string // text search configuration for to_tsvector/to_tsquery, defaults to "simple"
+	FullTextSearchMaxLength int    // maximum content length for full-text search, 0 means no limit
+	FullTextSearchColumn    string // column to search in, defaults to "content"
+	SubstringSearch         bool   // match NIP-50 search as an ILIKE '%q%' substring instead of tsvector full-text; see queryEventsSql
+	NotifyChannel           string // LISTEN/NOTIFY channel used by Notify and Notifications, defaults to "nostr_events"
 }
 
 func (b *PostgresBackend) Close() {

store.go

@@ -30,3 +30,18 @@ type Store interface {
 type Counter interface {
 	CountEvents(context.Context, nostr.Filter) (int64, error)
 }
+
+// Notifier is implemented by stores that can propagate events between several
+// processes sharing the same storage, so that a relay running as multiple
+// instances can push events published to one instance to the subscribers of
+// all the others.
+//
+// The relay calls Notify for every accepted event, after saving it; ephemeral
+// events are passed too even though they are never saved. Notifications must
+// deliver every event passed to Notify by any process, including the calling
+// one, and keep its channel open until ctx is done, reconnecting to the
+// underlying transport as needed.
+type Notifier interface {
+	Notify(context.Context, *nostr.Event) error
+	Notifications(context.Context) (<-chan *nostr.Event, error)
+}

test/notify_test.go

@@ -0,0 +1,109 @@
+package test
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	embeddedpostgres "github.com/fergusstrange/embedded-postgres"
+	"github.com/fiatjaf/eventstore/postgresql"
+	"github.com/nbd-wtf/go-nostr"
+)
+
+func TestPostgresNotifier(t *testing.T) {
+	postgres := embeddedpostgres.NewDatabase()
+	if err := postgres.Start(); err != nil {
+		t.Fatalf("failed to start embedded postgres: %s", err)
+	}
+	defer postgres.Stop()
+
+	url := "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
+
+	// two backends on the same database stand in for two relay processes
+	a := &postgresql.PostgresBackend{DatabaseURL: url}
+	b := &postgresql.PostgresBackend{DatabaseURL: url}
+	for _, db := range []*postgresql.PostgresBackend{a, b} {
+		if err := db.Init(); err != nil {
+			t.Fatal(err)
+		}
+		defer db.Close()
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	fromA, err := a.Notifications(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	fromB, err := b.Notifications(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	recv := func(ch <-chan *nostr.Event) *nostr.Event {
+		t.Helper()
+		select {
+		case evt := <-ch:
+			return evt
+		case <-time.After(5 * time.Second):
+			t.Fatal("no notification received")
+			return nil
+		}
+	}
+
+	// a small event travels inline
+	small := &nostr.Event{Kind: 1, Content: "hello", CreatedAt: nostr.Now(), Tags: nostr.Tags{}}
+	small.Sign(sk3)
+	if err := a.SaveEvent(ctx, small); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.Notify(ctx, small); err != nil {
+		t.Fatal(err)
+	}
+	for _, ch := range []<-chan *nostr.Event{fromA, fromB} {
+		if got := recv(ch); got.ID != small.ID || got.Content != small.Content {
+			t.Errorf("expected %s, got %s", small.ID, got.ID)
+		}
+	}
+
+	// an event above the pg_notify limit is announced by id and read back
+	big := &nostr.Event{Kind: 1, Content: strings.Repeat("x", 9000), CreatedAt: nostr.Now(), Tags: nostr.Tags{}}
+	big.Sign(sk3)
+	if err := a.SaveEvent(ctx, big); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.Notify(ctx, big); err != nil {
+		t.Fatal(err)
+	}
+	for _, ch := range []<-chan *nostr.Event{fromA, fromB} {
+		if got := recv(ch); got.ID != big.ID || got.Content != big.Content {
+			t.Errorf("expected %s with full content, got %s (%d bytes)", big.ID, got.ID, len(got.Content))
+		}
+	}
+
+	// an ephemeral event is not saved but still travels
+	eph := &nostr.Event{Kind: 20001, Content: "ephemeral", CreatedAt: nostr.Now(), Tags: nostr.Tags{}}
+	eph.Sign(sk3)
+	if err := a.Notify(ctx, eph); err != nil {
+		t.Fatal(err)
+	}
+	for _, ch := range []<-chan *nostr.Event{fromA, fromB} {
+		if got := recv(ch); got.ID != eph.ID {
+			t.Errorf("expected %s, got %s", eph.ID, got.ID)
+		}
+	}
+
+	// cancelling the context closes the channels
+	cancel()
+	for _, ch := range []<-chan *nostr.Event{fromA, fromB} {
+		select {
+		case _, ok := <-ch:
+			if ok {
+				t.Error("expected channel to be closed")
+			}
+		case <-time.After(5 * time.Second):
+			t.Error("channel not closed after cancel")
+		}
+	}
+}