nostermd-relay/internal/federation/federation.go

// 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()
}