initial public relay build repo

internal/federation/federation.go

@@ -0,0 +1,406 @@
+// Package federation lets the relay mirror NIP-29 chat with peer relays, so a
+// channel can live on more than one relay and survive any single one going down.
+// It is OPTIONAL and chat-only by design: only kind-9/10 messages cross the
+// boundary. Membership, admins, bans and metadata stay local to each relay (each
+// operator moderates their own instance) — this sidesteps cross-relay moderation
+// conflicts entirely.
+//
+// Two per-channel modes:
+//
+//   - mirror  bidirectional. Chat posted on either relay is forwarded to the
+//     other, so the channel is fully replicated.
+//   - ingest  read-only. This relay pulls chat FROM the peer but never pushes
+//     back (announcement feeds, syndication, public-room mirrors).
+//
+// Loop prevention is structural rather than tag-based: an incoming federated
+// event is injected through the relay's normal add pipeline, whose eventstore
+// rejects duplicates (ErrDupEvent) BEFORE the egress hook runs. So an event that
+// has already been seen is never re-forwarded, and A↔B↔A cycles die on the
+// second sight. No custom "seen" bookkeeping to get wrong.
+package federation
+
+import (
+	"context"
+	"log"
+	"strings"
+	"sync"
+	"time"
+
+	"fiatjaf.com/nostr"
+
+	"github.com/nosterm/relay/internal/group"
+)
+
+// Mode is a peer's federation direction.
+type Mode int
+
+const (
+	// ModeMirror is bidirectional federation.
+	ModeMirror Mode = iota
+	// ModeIngest is read-only (pull-only) federation.
+	ModeIngest
+)
+
+func (m Mode) String() string {
+	if m == ModeIngest {
+		return "ingest"
+	}
+	return "mirror"
+}
+
+// Peer is one configured peer relay and the channels federated with it.
+type Peer struct {
+	URL      string
+	Mode     Mode
+	Channels map[string]bool // group ids; empty map with AllChans=true means "all"
+	AllChans bool
+}
+
+// wants reports whether events for the given group id federate with this peer.
+func (p *Peer) wants(groupID string) bool {
+	if p.AllChans {
+		return true
+	}
+	return p.Channels[groupID]
+}
+
+// Config is the parsed set of peers.
+type Config struct {
+	Peers []Peer
+}
+
+// Enabled reports whether any peer is configured.
+func (c Config) Enabled() bool {
+	return len(c.Peers) > 0
+}
+
+// FederatesChannel reports whether any peer federates the given group id. Used
+// by group evaluation to accept federated chat regardless of local membership
+// (a federated channel is effectively open on every participating relay).
+func (c Config) FederatesChannel(id string) bool {
+	for i := range c.Peers {
+		if c.Peers[i].wants(id) {
+			return true
+		}
+	}
+	return false
+}
+
+// ParseConfig parses the RELAY_FEDERATION_PEERS specification.
+//
+// Grammar (peers separated by ';', ignoring surrounding whitespace):
+//
+//	<url>|<mode>:<channels>
+//
+// where <mode> is "mirror" or "ingest" and <channels> is a comma-separated list
+// of group ids, or "*" for every channel. The "|<mode>:<channels>" suffix is
+// optional and defaults to "mirror:*". Blank entries are skipped.
+//
+// Examples:
+//
+//	wss://a.example
+//	wss://a.example|mirror:general,dev
+//	wss://a.example|mirror:general;wss://b.example|ingest:announcements
+//	wss://a.example|ingest:*
+func ParseConfig(spec string) (Config, []string) {
+	var cfg Config
+	var warnings []string
+
+	for _, raw := range strings.Split(spec, ";") {
+		entry := strings.TrimSpace(raw)
+		if entry == "" {
+			continue
+		}
+
+		urlPart := entry
+		modePart := ""
+		if i := strings.Index(entry, "|"); i >= 0 {
+			urlPart = strings.TrimSpace(entry[:i])
+			modePart = strings.TrimSpace(entry[i+1:])
+		}
+
+		// Require an explicit ws/wss scheme on the RAW input: NormalizeURL would
+		// silently rewrite http→ws and bare hosts→wss, masking config mistakes.
+		lower := strings.ToLower(urlPart)
+		if !(strings.HasPrefix(lower, "ws://") || strings.HasPrefix(lower, "wss://")) {
+			warnings = append(warnings, "ignoring peer with non-ws(s) URL: "+urlPart)
+			continue
+		}
+		url := nostr.NormalizeURL(urlPart)
+		if url == "" {
+			warnings = append(warnings, "ignoring peer with invalid URL: "+urlPart)
+			continue
+		}
+
+		peer := Peer{
+			URL:      url,
+			Mode:     ModeMirror,
+			Channels: map[string]bool{},
+			AllChans: true, // default: all channels
+		}
+
+		if modePart != "" {
+			modeName := modePart
+			chanList := "*"
+			if j := strings.Index(modePart, ":"); j >= 0 {
+				modeName = strings.TrimSpace(modePart[:j])
+				chanList = strings.TrimSpace(modePart[j+1:])
+			}
+			switch strings.ToLower(modeName) {
+			case "mirror", "":
+				peer.Mode = ModeMirror
+			case "ingest", "pull", "read", "readonly":
+				peer.Mode = ModeIngest
+			default:
+				warnings = append(warnings, "unknown federation mode "+modeName+" for "+url+"; using mirror")
+				peer.Mode = ModeMirror
+			}
+			if chanList != "" && chanList != "*" {
+				peer.AllChans = false
+				for _, c := range strings.Split(chanList, ",") {
+					id := strings.TrimSpace(strings.TrimPrefix(c, "#"))
+					if id != "" {
+						peer.Channels[id] = true
+					}
+				}
+				// A mode with an explicit but empty channel list federates nothing.
+				if len(peer.Channels) == 0 {
+					warnings = append(warnings, "peer "+url+" lists no valid channels; skipping")
+					continue
+				}
+			}
+		}
+
+		cfg.Peers = append(cfg.Peers, peer)
+	}
+
+	return cfg, warnings
+}
+
+// EventInjector accepts a federated event into the local relay pipeline. It
+// returns whether the event was newly stored (false for duplicates/rejects), so
+// tests can assert loop prevention without a live relay. In production this is
+// backed by khatru's Relay.AddEvent.
+type EventInjector func(ctx context.Context, evt nostr.Event) (stored bool)
+
+// Federator owns the live peer connections and the forwarding logic.
+type Federator struct {
+	cfg     Config
+	inject  EventInjector
+	dialCtx context.Context
+	cancel  context.CancelFunc
+
+	mu        sync.Mutex
+	connected map[string]*nostr.Relay // url -> live client connection
+
+	wg sync.WaitGroup
+}
+
+// federationKinds are the only kinds we ever subscribe to / forward: chat only.
+var federationKinds = []nostr.Kind{
+	nostr.KindSimpleGroupChatMessage,   // 9
+	nostr.KindSimpleGroupThreadedReply, // 10
+}
+
+// NewFederator creates a federator for the given config. inject feeds received
+// peer events into the local relay pipeline.
+func NewFederator(cfg Config, inject EventInjector) *Federator {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &Federator{
+		cfg:       cfg,
+		inject:    inject,
+		dialCtx:   ctx,
+		cancel:    cancel,
+		connected: map[string]*nostr.Relay{},
+	}
+}
+
+// Start dials every peer and begins pulling their chat. Non-blocking: each peer
+// runs in its own goroutine that reconnects with backoff until Stop is called.
+func (f *Federator) Start() {
+	for i := range f.cfg.Peers {
+		peer := f.cfg.Peers[i]
+		f.wg.Add(1)
+		go func() {
+			defer f.wg.Done()
+			f.runPeer(peer)
+		}()
+	}
+	log.Printf("federation: started with %d peer(s)", len(f.cfg.Peers))
+}
+
+// Stop tears down all peer connections and waits for their goroutines to exit.
+func (f *Federator) Stop() {
+	f.cancel()
+	f.mu.Lock()
+	for _, r := range f.connected {
+		_ = r.Close()
+	}
+	f.mu.Unlock()
+	f.wg.Wait()
+}
+
+// runPeer maintains one peer connection, reconnecting with capped backoff. Each
+// (re)connection opens a fresh subscription for the federated chat kinds and
+// injects received events into the local pipeline (dedup handles loops).
+func (f *Federator) runPeer(peer Peer) {
+	backoff := time.Second
+	const maxBackoff = 30 * time.Second
+
+	for f.dialCtx.Err() == nil {
+		if err := f.connectAndPull(peer); err != nil {
+			if f.dialCtx.Err() != nil {
+				return
+			}
+			log.Printf("federation: peer %s (%s) error: %v; retrying in %s",
+				peer.URL, peer.Mode, err, backoff)
+			select {
+			case <-f.dialCtx.Done():
+				return
+			case <-time.After(backoff):
+			}
+			backoff *= 2
+			if backoff > maxBackoff {
+				backoff = maxBackoff
+			}
+			continue
+		}
+		// Clean subscription end (peer closed / context canceled) — reset backoff
+		// and try to re-establish unless we're shutting down.
+		backoff = time.Second
+	}
+}
+
+// connectAndPull opens a connection + subscription and blocks pumping events
+// until the subscription ends or the connection drops. Returns nil on a clean
+// end (so the caller re-establishes), or an error to trigger backoff.
+func (f *Federator) connectAndPull(peer Peer) error {
+	relay, err := nostr.RelayConnect(f.dialCtx, peer.URL, nostr.RelayOptions{})
+	if err != nil {
+		return err
+	}
+	f.mu.Lock()
+	f.connected[peer.URL] = relay
+	f.mu.Unlock()
+	defer func() {
+		f.mu.Lock()
+		delete(f.connected, peer.URL)
+		f.mu.Unlock()
+		_ = relay.Close()
+	}()
+
+	filter := nostr.Filter{Kinds: federationKinds}
+	// Scope the pull to the configured channels when not "all", so we don't drag
+	// down chat for channels this peer isn't federating.
+	if !peer.AllChans {
+		ids := make([]string, 0, len(peer.Channels))
+		for id := range peer.Channels {
+			ids = append(ids, id)
+		}
+		filter.Tags = nostr.TagMap{"h": ids}
+	}
+
+	sub, err := relay.Subscribe(f.dialCtx, filter, nostr.SubscriptionOptions{Label: "federation"})
+	if err != nil {
+		return err
+	}
+	defer sub.Unsub()
+
+	log.Printf("federation: pulling %s from %s", peer.Mode, peer.URL)
+
+	for {
+		select {
+		case <-f.dialCtx.Done():
+			return nil
+		case <-sub.EndOfStoredEvents:
+			// Historical backfill done; keep receiving live events.
+		case reason := <-sub.ClosedReason:
+			log.Printf("federation: peer %s closed subscription: %s", peer.URL, reason)
+			return nil
+		case evt, ok := <-sub.Events:
+			if !ok {
+				return nil
+			}
+			f.receive(peer, evt)
+		}
+	}
+}
+
+// receive validates and injects an event pulled from a peer. Signature is
+// re-verified defensively (never trust a peer to have done so), the channel
+// scope is re-checked, and the event is fed through the local pipeline where
+// the eventstore dedups it — that dedup is what prevents forwarding loops.
+func (f *Federator) receive(peer Peer, evt nostr.Event) {
+	id := group.GroupIDFromEvent(evt)
+	if id == "" || !peer.wants(id) {
+		return
+	}
+	if !evt.VerifySignature() {
+		log.Printf("federation: dropped event %s from %s (bad signature)", evt.ID.Hex(), peer.URL)
+		return
+	}
+	f.inject(f.dialCtx, evt)
+}
+
+// Forward pushes a locally-saved chat event out to every peer that federates its
+// channel in a writable (mirror) mode. Ingest peers are pull-only and are
+// skipped. Called from the relay's OnEventSaved hook for chat kinds; a duplicate
+// re-injected from a peer never reaches here because AddEvent stops on dedup
+// before OnEventSaved fires.
+func (f *Federator) Forward(evt nostr.Event) {
+	id := group.GroupIDFromEvent(evt)
+	if id == "" {
+		return
+	}
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for i := range f.cfg.Peers {
+		peer := f.cfg.Peers[i]
+		if peer.Mode != ModeMirror || !peer.wants(id) {
+			continue
+		}
+		relay := f.connected[peer.URL]
+		if relay == nil || !relay.IsConnected() {
+			continue // reconnect logic will backfill via subscription
+		}
+		// Publish in the background so a slow peer can't block the save path.
+		f.wg.Add(1)
+		go func(r *nostr.Relay, url string) {
+			defer f.wg.Done()
+			ctx, cancel := context.WithTimeout(f.dialCtx, 10*time.Second)
+			defer cancel()
+			if err := r.Publish(ctx, evt); err != nil {
+				log.Printf("federation: publish to %s failed: %v", url, err)
+			}
+		}(relay, peer.URL)
+	}
+}
+
+// Summary is a short human-readable description for logging/inspection.
+func (c Config) Summary() string {
+	if !c.Enabled() {
+		return "federation disabled"
+	}
+	var b strings.Builder
+	b.WriteString("federation: ")
+	for i, p := range c.Peers {
+		if i > 0 {
+			b.WriteString(", ")
+		}
+		scope := "*"
+		if !p.AllChans {
+			ids := make([]string, 0, len(p.Channels))
+			for id := range p.Channels {
+				ids = append(ids, id)
+			}
+			scope = strings.Join(ids, "/")
+		}
+		b.WriteString(p.URL)
+		b.WriteString("(")
+		b.WriteString(p.Mode.String())
+		b.WriteString(":")
+		b.WriteString(scope)
+		b.WriteString(")")
+	}
+	return b.String()
+}

internal/federation/federation_test.go

@@ -0,0 +1,215 @@
+package federation
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"fiatjaf.com/nostr"
+)
+
+func TestParseConfig(t *testing.T) {
+	t.Run("empty spec disables federation", func(t *testing.T) {
+		cfg, warnings := ParseConfig("")
+		if cfg.Enabled() {
+			t.Fatal("expected federation disabled for empty spec")
+		}
+		if len(warnings) != 0 {
+			t.Fatalf("unexpected warnings: %v", warnings)
+		}
+	})
+
+	t.Run("bare url defaults to mirror all channels", func(t *testing.T) {
+		cfg, _ := ParseConfig("wss://a.example")
+		if len(cfg.Peers) != 1 {
+			t.Fatalf("expected 1 peer, got %d", len(cfg.Peers))
+		}
+		p := cfg.Peers[0]
+		if p.Mode != ModeMirror {
+			t.Errorf("expected mirror mode, got %s", p.Mode)
+		}
+		if !p.AllChans {
+			t.Error("expected AllChans=true for bare url")
+		}
+		if !p.wants("anything") {
+			t.Error("bare url should federate every channel")
+		}
+	})
+
+	t.Run("explicit mode and channels", func(t *testing.T) {
+		cfg, _ := ParseConfig("wss://a.example|ingest:general,dev")
+		p := cfg.Peers[0]
+		if p.Mode != ModeIngest {
+			t.Errorf("expected ingest, got %s", p.Mode)
+		}
+		if p.AllChans {
+			t.Error("expected scoped channels, got AllChans")
+		}
+		if !p.wants("general") || !p.wants("dev") {
+			t.Error("should federate listed channels")
+		}
+		if p.wants("secret") {
+			t.Error("should NOT federate unlisted channel")
+		}
+	})
+
+	t.Run("multiple peers separated by semicolons", func(t *testing.T) {
+		cfg, _ := ParseConfig("wss://a.example|mirror:general; wss://b.example|ingest:*")
+		if len(cfg.Peers) != 2 {
+			t.Fatalf("expected 2 peers, got %d", len(cfg.Peers))
+		}
+		if cfg.Peers[0].Mode != ModeMirror || cfg.Peers[1].Mode != ModeIngest {
+			t.Error("modes parsed incorrectly")
+		}
+		if !cfg.Peers[1].AllChans {
+			t.Error("ingest:* should mean all channels")
+		}
+	})
+
+	t.Run("channel names tolerate a leading #", func(t *testing.T) {
+		cfg, _ := ParseConfig("wss://a.example|mirror:#general")
+		if !cfg.Peers[0].wants("general") {
+			t.Error("should strip leading # from channel names")
+		}
+	})
+
+	t.Run("invalid url is skipped with a warning", func(t *testing.T) {
+		cfg, warnings := ParseConfig("http://not-websocket.example")
+		if cfg.Enabled() {
+			t.Error("non-ws url should be skipped")
+		}
+		if len(warnings) == 0 {
+			t.Error("expected a warning for invalid url")
+		}
+	})
+
+	t.Run("explicit but empty channel list is skipped", func(t *testing.T) {
+		cfg, warnings := ParseConfig("wss://a.example|mirror:,")
+		if cfg.Enabled() {
+			t.Error("empty channel list should skip the peer")
+		}
+		if len(warnings) == 0 {
+			t.Error("expected a warning for empty channel list")
+		}
+	})
+
+	t.Run("unknown mode falls back to mirror with a warning", func(t *testing.T) {
+		cfg, warnings := ParseConfig("wss://a.example|bogus:general")
+		if cfg.Peers[0].Mode != ModeMirror {
+			t.Error("unknown mode should fall back to mirror")
+		}
+		if len(warnings) == 0 {
+			t.Error("expected a warning for unknown mode")
+		}
+	})
+}
+
+func TestFederatesChannel(t *testing.T) {
+	cfg, _ := ParseConfig("wss://a.example|mirror:general;wss://b.example|ingest:dev")
+	if !cfg.FederatesChannel("general") {
+		t.Error("general federates via peer a")
+	}
+	if !cfg.FederatesChannel("dev") {
+		t.Error("dev federates via peer b")
+	}
+	if cfg.FederatesChannel("private") {
+		t.Error("private federates with nobody")
+	}
+}
+
+// TestReceiveLoopPrevention proves the loop guard: an event injected twice is
+// only stored (and thus only forwardable) once, because the injector reports the
+// second sight as not-newly-stored. This mirrors AddEvent's ErrDupEvent path.
+func TestReceiveLoopPrevention(t *testing.T) {
+	seen := map[nostr.ID]bool{}
+	var mu sync.Mutex
+	injectCount := 0
+	storedCount := 0
+
+	inject := func(_ context.Context, evt nostr.Event) bool {
+		mu.Lock()
+		defer mu.Unlock()
+		injectCount++
+		if seen[evt.ID] {
+			return false // duplicate: not newly stored (the loop breaker)
+		}
+		seen[evt.ID] = true
+		storedCount++
+		return true
+	}
+
+	cfg, _ := ParseConfig("wss://peer.example|mirror:general")
+	f := NewFederator(cfg, inject)
+	peer := cfg.Peers[0]
+
+	sk := nostr.Generate()
+	evt := nostr.Event{
+		Kind:      nostr.KindSimpleGroupChatMessage,
+		CreatedAt: nostr.Now(),
+		Content:   "hello federation",
+		Tags:      nostr.Tags{{"h", "general"}},
+	}
+	if err := evt.Sign(sk); err != nil {
+		t.Fatalf("sign: %v", err)
+	}
+
+	// Same event received twice (e.g. A→B then echoed B→A→B).
+	f.receive(peer, evt)
+	f.receive(peer, evt)
+
+	if injectCount != 2 {
+		t.Errorf("expected 2 inject attempts, got %d", injectCount)
+	}
+	if storedCount != 1 {
+		t.Errorf("loop guard failed: event stored %d times, want 1", storedCount)
+	}
+}
+
+// TestReceiveDropsBadSignature ensures a peer can't inject forged events.
+func TestReceiveDropsBadSignature(t *testing.T) {
+	injected := false
+	inject := func(_ context.Context, _ nostr.Event) bool {
+		injected = true
+		return true
+	}
+	cfg, _ := ParseConfig("wss://peer.example|mirror:general")
+	f := NewFederator(cfg, inject)
+
+	// Unsigned (invalid) event tagged for a federated channel.
+	evt := nostr.Event{
+		Kind:      nostr.KindSimpleGroupChatMessage,
+		CreatedAt: nostr.Now(),
+		Content:   "forged",
+		Tags:      nostr.Tags{{"h", "general"}},
+	}
+	f.receive(cfg.Peers[0], evt)
+	if injected {
+		t.Error("event with an invalid signature must not be injected")
+	}
+}
+
+// TestReceiveIgnoresUnfederatedChannel ensures scope is enforced on ingress.
+func TestReceiveIgnoresUnfederatedChannel(t *testing.T) {
+	injected := false
+	inject := func(_ context.Context, _ nostr.Event) bool {
+		injected = true
+		return true
+	}
+	cfg, _ := ParseConfig("wss://peer.example|mirror:general")
+	f := NewFederator(cfg, inject)
+
+	sk := nostr.Generate()
+	evt := nostr.Event{
+		Kind:      nostr.KindSimpleGroupChatMessage,
+		CreatedAt: nostr.Now(),
+		Content:   "off-topic",
+		Tags:      nostr.Tags{{"h", "other-channel"}},
+	}
+	if err := evt.Sign(sk); err != nil {
+		t.Fatalf("sign: %v", err)
+	}
+	f.receive(cfg.Peers[0], evt)
+	if injected {
+		t.Error("chat for an unfederated channel must not be injected")
+	}
+}