nostrlib/khatru/relay.go

package khatru

import (
	"context"
	"iter"
	"log"
	"net/http"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"fiatjaf.com/nostr"
	"fiatjaf.com/nostr/eventstore"
	"fiatjaf.com/nostr/nip11"
	"fiatjaf.com/nostr/nip45/hyperloglog"
	"github.com/fasthttp/websocket"
)

func NewRelay() *Relay {
	ctx := context.Background()

	rl := &Relay{
		Log: log.New(os.Stderr, "[khatru-relay] ", log.LstdFlags),

		Info: &nip11.RelayInformationDocument{
			Software:      "https://fiatjaf.com/nostr/khatru",
			Version:       "n/a",
			SupportedNIPs: []any{1, 11, 40, 42, 70, 86},
		},

		upgrader: websocket.Upgrader{
			ReadBufferSize:  1024,
			WriteBufferSize: 1024,
			CheckOrigin:     func(r *http.Request) bool { return true },
		},

		clients:   make(map[*WebSocket][]listenerSpec, 100),
		listeners: make([]listener, 0, 100),

		serveMux: &http.ServeMux{},

		WriteWait:      10 * time.Second,
		PongWait:       60 * time.Second,
		PingPeriod:     30 * time.Second,
		MaxMessageSize: 512000,
	}

	rl.expirationManager = newExpirationManager(rl)
	go rl.expirationManager.start(ctx)

	return rl
}

type Relay struct {
	// setting this variable overwrites the hackish workaround we do to try to figure out our own base URL
	ServiceURL string

	// hooks that will be called at various times
	OnEvent                   func(ctx context.Context, event nostr.Event) (reject bool, msg string)
	StoreEvent                func(ctx context.Context, event nostr.Event) error
	ReplaceEvent              func(ctx context.Context, event nostr.Event) error
	DeleteEvent               func(ctx context.Context, id nostr.ID) error
	OnEventSaved              func(ctx context.Context, event nostr.Event)
	OnEphemeralEvent          func(ctx context.Context, event nostr.Event)
	OnRequest                 func(ctx context.Context, filter nostr.Filter) (reject bool, msg string)
	OnCountFilter             func(ctx context.Context, filter nostr.Filter) (reject bool, msg string)
	QueryStored               func(ctx context.Context, filter nostr.Filter) iter.Seq[nostr.Event]
	Count                     func(ctx context.Context, filter nostr.Filter) (uint32, error)
	CountHLL                  func(ctx context.Context, filter nostr.Filter, offset int) (uint32, *hyperloglog.HyperLogLog, error)
	RejectConnection          func(r *http.Request) bool
	OnConnect                 func(ctx context.Context)
	OnDisconnect              func(ctx context.Context)
	OverwriteRelayInformation func(ctx context.Context, r *http.Request, info nip11.RelayInformationDocument) nip11.RelayInformationDocument
	PreventBroadcast          func(ws *WebSocket, event nostr.Event) bool

	// these are used when this relays acts as a router
	routes                []Route
	getSubRelayFromEvent  func(*nostr.Event) *Relay // used for handling EVENTs
	getSubRelayFromFilter func(nostr.Filter) *Relay // used for handling REQs

	// setting up handlers here will enable these methods
	ManagementAPI RelayManagementAPI

	// editing info will affect the NIP-11 responses
	Info *nip11.RelayInformationDocument

	// Default logger, as set by NewServer, is a stdlib logger prefixed with "[khatru-relay] ",
	// outputting to stderr.
	Log *log.Logger

	// for establishing websockets
	upgrader websocket.Upgrader

	// keep a connection reference to all connected clients for Server.Shutdown
	// also used for keeping track of who is listening to what
	clients      map[*WebSocket][]listenerSpec
	listeners    []listener
	clientsMutex sync.Mutex

	// set this to true to support negentropy
	Negentropy bool

	// in case you call Server.Start
	Addr       string
	serveMux   *http.ServeMux
	httpServer *http.Server

	// websocket options
	WriteWait      time.Duration // Time allowed to write a message to the peer.
	PongWait       time.Duration // Time allowed to read the next pong message from the peer.
	PingPeriod     time.Duration // Send pings to peer with this period. Must be less than pongWait.
	MaxMessageSize int64         // Maximum message size allowed from peer.

	// NIP-40 expiration manager
	expirationManager *expirationManager
}

func (rl *Relay) UseEventstore(store eventstore.Store) {
	rl.QueryStored = func(ctx context.Context, filter nostr.Filter) iter.Seq[nostr.Event] {
		return store.QueryEvents(filter)
	}
	rl.Count = func(ctx context.Context, filter nostr.Filter) (uint32, error) {
		return store.CountEvents(filter)
	}
	rl.StoreEvent = func(ctx context.Context, event nostr.Event) error {
		return store.SaveEvent(event)
	}
	rl.ReplaceEvent = func(ctx context.Context, event nostr.Event) error {
		return store.ReplaceEvent(event)
	}
	rl.DeleteEvent = func(ctx context.Context, id nostr.ID) error {
		return store.DeleteEvent(id)
	}
}

func (rl *Relay) getBaseURL(r *http.Request) string {
	if rl.ServiceURL != "" {
		return rl.ServiceURL
	}

	host := r.Header.Get("X-Forwarded-Host")
	if host == "" {
		host = r.Host
	}
	proto := r.Header.Get("X-Forwarded-Proto")
	if proto == "" {
		if host == "localhost" {
			proto = "http"
		} else if strings.Contains(host, ":") {
			// has a port number
			proto = "http"
		} else if _, err := strconv.Atoi(strings.ReplaceAll(host, ".", "")); err == nil {
			// it's a naked IP
			proto = "http"
		} else {
			proto = "https"
		}
	}
	return proto + "://" + host
}