make it so ephemeral events respond with ok:false if no one is listening.

adding.go

@@ -11,34 +11,46 @@ import (
 
 // AddEvent sends an event through then normal add pipeline, as if it was received from a websocket.
 func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	if evt == nil {
 		return false, errors.New("error: event is nil")
 	}
 
+	if nostr.IsEphemeralKind(evt.Kind) {
+		return false, rl.handleEphemeral(ctx, evt)
+	} else {
+		return rl.handleNormal(ctx, evt)
+	}
+}
+
+func (rl *Relay) handleNormal(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error) {
 	for _, reject := range rl.RejectEvent {
 		if reject, msg := reject(ctx, evt); reject {
 			if msg == "" {
-				return false, errors.New("blocked: no reason")
+				return true, errors.New("blocked: no reason")
 			} else {
-				return false, errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
+				return true, errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
 			}
 		}
 	}
 
-	if 20000 <= evt.Kind && evt.Kind < 30000 {
+	// will store
-		// do not store ephemeral events
+	// regular kinds are just saved directly
-		for _, oee := range rl.OnEphemeralEvent {
+	if nostr.IsRegularKind(evt.Kind) {
-			oee(ctx, evt)
+		for _, store := range rl.StoreEvent {
+			if err := store(ctx, evt); err != nil {
+				switch err {
+				case eventstore.ErrDupEvent:
+					return true, nil
+				default:
+					return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(err.Error(), "error"))
+				}
+			}
 		}
 	} else {
-		// will store
+		// otherwise it's a replaceable -- so we'll use the replacer functions if we have any
-		// regular kinds are just saved directly
+		if len(rl.ReplaceEvent) > 0 {
-		if nostr.IsRegularKind(evt.Kind) {
+			for _, repl := range rl.ReplaceEvent {
-			for _, store := range rl.StoreEvent {
+				if err := repl(ctx, evt); err != nil {
-				if err := store(ctx, evt); err != nil {
 					switch err {
 					case eventstore.ErrDupEvent:
 						return true, nil
@@ -48,68 +60,54 @@ func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast
 				}
 			}
 		} else {
-			// otherwise it's a replaceable -- so we'll use the replacer functions if we have any
+			// otherwise do it the manual way
-			if len(rl.ReplaceEvent) > 0 {
+			filter := nostr.Filter{Limit: 1, Kinds: []int{evt.Kind}, Authors: []string{evt.PubKey}}
-				for _, repl := range rl.ReplaceEvent {
+			if nostr.IsAddressableKind(evt.Kind) {
-					if err := repl(ctx, evt); err != nil {
+				// when addressable, add the "d" tag to the filter
-						switch err {
+				filter.Tags = nostr.TagMap{"d": []string{evt.Tags.GetD()}}
+			}
+
+			// now we fetch old events and delete them
+			shouldStore := true
+			for _, query := range rl.QueryEvents {
+				ch, err := query(ctx, filter)
+				if err != nil {
+					continue
+				}
+				for previous := range ch {
+					if isOlder(previous, evt) {
+						for _, del := range rl.DeleteEvent {
+							del(ctx, previous)
+						}
+					} else {
+						// we found a more recent event, so we won't delete it and also will not store this new one
+						shouldStore = false
+					}
+				}
+			}
+
+			// store
+			if shouldStore {
+				for _, store := range rl.StoreEvent {
+					if saveErr := store(ctx, evt); saveErr != nil {
+						switch saveErr {
 						case eventstore.ErrDupEvent:
 							return true, nil
 						default:
-							return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(err.Error(), "error"))
+							return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(saveErr.Error(), "error"))
-						}
-					}
-				}
-			} else {
-				// otherwise do it the manual way
-				filter := nostr.Filter{Limit: 1, Kinds: []int{evt.Kind}, Authors: []string{evt.PubKey}}
-				if nostr.IsAddressableKind(evt.Kind) {
-					// when addressable, add the "d" tag to the filter
-					filter.Tags = nostr.TagMap{"d": []string{evt.Tags.GetD()}}
-				}
-
-				// now we fetch old events and delete them
-				shouldStore := true
-				for _, query := range rl.QueryEvents {
-					ch, err := query(ctx, filter)
-					if err != nil {
-						continue
-					}
-					for previous := range ch {
-						if isOlder(previous, evt) {
-							for _, del := range rl.DeleteEvent {
-								del(ctx, previous)
-							}
-						} else {
-							// we found a more recent event, so we won't delete it and also will not store this new one
-							shouldStore = false
-						}
-					}
-				}
-
-				// store
-				if shouldStore {
-					for _, store := range rl.StoreEvent {
-						if saveErr := store(ctx, evt); saveErr != nil {
-							switch saveErr {
-							case eventstore.ErrDupEvent:
-								return true, nil
-							default:
-								return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(saveErr.Error(), "error"))
-							}
 						}
 					}
 				}
 			}
 		}
-
-		for _, ons := range rl.OnEventSaved {
-			ons(ctx, evt)
-		}
-
-		// track event expiration if applicable
-		rl.expirationManager.trackEvent(evt)
 	}
 
+	for _, ons := range rl.OnEventSaved {
+		ons(ctx, evt)
+	}
+
+	// track event expiration if applicable
+	rl.expirationManager.trackEvent(evt)
+
 	return false, nil
 }

blossom/handlers.go

@@ -25,7 +25,7 @@ func (bs BlossomServer) handleUploadCheck(w http.ResponseWriter, r *http.Request
 		blossomError(w, "missing \"Authorization\" header", 401)
 		return
 	}
-	if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
+	if auth.Tags.FindWithValue("t", "upload") == nil {
 		blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
 		return
 	}
@@ -59,7 +59,7 @@ func (bs BlossomServer) handleUpload(w http.ResponseWriter, r *http.Request) {
 		blossomError(w, "missing \"Authorization\" header", 401)
 		return
 	}
-	if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
+	if auth.Tags.FindWithValue("t", "upload") == nil {
 		blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
 		return
 	}
@@ -163,13 +163,13 @@ func (bs BlossomServer) handleGetBlob(w http.ResponseWriter, r *http.Request) {
 
 	// if there is one, we check if it has the extra requirements
 	if auth != nil {
-		if auth.Tags.GetFirst([]string{"t", "get"}) == nil {
+		if auth.Tags.FindWithValue("t", "get") == nil {
 			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
 			return
 		}
 
-		if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
+		if auth.Tags.FindWithValue("x", hhash) == nil &&
-			auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
+			auth.Tags.FindWithValue("server", bs.ServiceURL) == nil {
 			blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
 			return
 		}
@@ -239,7 +239,7 @@ func (bs BlossomServer) handleList(w http.ResponseWriter, r *http.Request) {
 
 	// if there is one, we check if it has the extra requirements
 	if auth != nil {
-		if auth.Tags.GetFirst([]string{"t", "list"}) == nil {
+		if auth.Tags.FindWithValue("t", "list") == nil {
 			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
 			return
 		}
@@ -283,7 +283,7 @@ func (bs BlossomServer) handleDelete(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if auth != nil {
-		if auth.Tags.GetFirst([]string{"t", "delete"}) == nil {
+		if auth.Tags.FindWithValue("t", "delete") == nil {
 			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
 			return
 		}
@@ -296,8 +296,8 @@ func (bs BlossomServer) handleDelete(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	hhash = hhash[1:]
-	if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
+	if auth.Tags.FindWithValue("x", hhash) == nil &&
-		auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
+		auth.Tags.FindWithValue("server", bs.ServiceURL) == nil {
 		blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
 		return
 	}

broadcasting.go

@@ -6,6 +6,6 @@ import (
 
 // BroadcastEvent emits an event to all listeners whose filters' match, skipping all filters and actions
 // it also doesn't attempt to store the event or trigger any reactions or callbacks
-func (rl *Relay) BroadcastEvent(evt *nostr.Event) {
+func (rl *Relay) BroadcastEvent(evt *nostr.Event) int {
-	rl.notifyListeners(evt)
+	return rl.notifyListeners(evt)
 }

docs/core/routing.md

@@ -47,7 +47,7 @@ router.Route().
 			return true
 		case event.Kind <= 12 && event.Kind >= 9:
 			return true
-		case event.Tags.GetFirst([]string{"h", ""}) != nil:
+		case event.Tags.Find("h") != nil:
 			return true
 		default:
 			return false

ephemeral.go

@@ -0,0 +1,26 @@
+package khatru
+
+import (
+	"context"
+	"errors"
+
+	"github.com/nbd-wtf/go-nostr"
+)
+
+func (rl *Relay) handleEphemeral(ctx context.Context, evt *nostr.Event) error {
+	for _, reject := range rl.RejectEvent {
+		if reject, msg := reject(ctx, evt); reject {
+			if msg == "" {
+				return errors.New("blocked: no reason")
+			} else {
+				return errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
+			}
+		}
+	}
+
+	for _, oee := range rl.OnEphemeralEvent {
+		oee(ctx, evt)
+	}
+
+	return nil
+}

examples/basic-lmdb/main.go

@@ -13,7 +13,7 @@ func main() {
 	relay := khatru.NewRelay()
 
 	db := lmdb.LMDBBackend{Path: "/tmp/khatru-lmdb-tmp"}
-	os.MkdirAll(db.Path, 0755)
+	os.MkdirAll(db.Path, 0o755)
 	if err := db.Init(); err != nil {
 		panic(err)
 	}

examples/exclusive/main.go

@@ -16,7 +16,7 @@ func main() {
 	relay := khatru.NewRelay()
 
 	db := lmdb.LMDBBackend{Path: "/tmp/exclusive"}
-	os.MkdirAll(db.Path, 0755)
+	os.MkdirAll(db.Path, 0o755)
 	if err := db.Init(); err != nil {
 		panic(err)
 	}

examples/routing/main.go

@@ -52,7 +52,7 @@ func main() {
 			return slices.Contains(filter.Kinds, 1) && slices.Contains(filter.Tags["t"], "spam")
 		}).
 		Event(func(event *nostr.Event) bool {
-			return event.Kind == 1 && event.Tags.GetFirst([]string{"t", "spam"}) != nil
+			return event.Kind == 1 && event.Tags.FindWithValue("t", "spam") != nil
 		}).
 		Relay(r2)
 

expiration.go

@@ -73,6 +73,7 @@ func (em *expirationManager) initialScan(ctx context.Context) {
 	defer em.mu.Unlock()
 
 	// query all events
+	ctx = context.WithValue(ctx, internalCallKey, struct{}{})
 	for _, query := range em.relay.QueryEvents {
 		ch, err := query(ctx, nostr.Filter{})
 		if err != nil {

handlers.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"net/http"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -141,6 +142,7 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 				continue
 			}
 
+			// this is safe because ReadMessage() will always create a new slice
 			message := unsafe.String(unsafe.SliceData(msgb), len(msgb))
 
 			// parse messages sequentially otherwise sonic breaks
@@ -215,9 +217,12 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 					if env.Event.Kind == 5 {
 						// this always returns "blocked: " whenever it returns an error
 						writeErr = srl.handleDeleteRequest(ctx, &env.Event)
+					} else if nostr.IsEphemeralKind(env.Event.Kind) {
+						// this will also always return a prefixed reason
+						writeErr = srl.handleEphemeral(ctx, &env.Event)
 					} else {
 						// this will also always return a prefixed reason
-						skipBroadcast, writeErr = srl.AddEvent(ctx, &env.Event)
+						skipBroadcast, writeErr = srl.handleNormal(ctx, &env.Event)
 					}
 
 					var reason string
@@ -227,9 +232,20 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 							ovw(ctx, &env.Event)
 						}
 						if !skipBroadcast {
-							srl.notifyListeners(&env.Event)
+							n := srl.notifyListeners(&env.Event)
+
+							// the number of notified listeners matters in ephemeral events
+							if nostr.IsEphemeralKind(env.Event.Kind) {
+								if n == 0 {
+									ok = false
+									reason = "mute: no one was listening for this"
+								} else {
+									reason = "broadcasted to " + strconv.Itoa(n) + " listeners"
+								}
+							}
 						}
 					} else {
+						ok = false
 						reason = writeErr.Error()
 						if strings.HasPrefix(reason, "auth-required:") {
 							RequestAuth(ctx)
@@ -244,14 +260,13 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 
 					var total int64
 					var hll *hyperloglog.HyperLogLog
-					uneligibleForHLL := false
 
 					srl := rl
 					if rl.getSubRelayFromFilter != nil {
 						srl = rl.getSubRelayFromFilter(env.Filter)
 					}
 
-					if offset := nip45.HyperLogLogEventPubkeyOffsetForFilter(env.Filter); offset != -1 && !uneligibleForHLL {
+					if offset := nip45.HyperLogLogEventPubkeyOffsetForFilter(env.Filter); offset != -1 {
 						total, hll = srl.handleCountRequestWithHLL(ctx, ws, env.Filter, offset)
 					} else {
 						total = srl.handleCountRequest(ctx, ws, env.Filter)

listener.go

@@ -132,15 +132,20 @@ func (rl *Relay) removeClientAndListeners(ws *WebSocket) {
 	delete(rl.clients, ws)
 }
 
-func (rl *Relay) notifyListeners(event *nostr.Event) {
+// returns how many listeners were notified
+func (rl *Relay) notifyListeners(event *nostr.Event) int {
+	count := 0
+listenersloop:
 	for _, listener := range rl.listeners {
 		if listener.filter.Matches(event) {
 			for _, pb := range rl.PreventBroadcast {
 				if pb(listener.ws, event) {
-					return
+					continue listenersloop
 				}
 			}
 			listener.ws.WriteJSON(nostr.EventEnvelope{SubscriptionID: &listener.id, Event: *event})
+			count++
 		}
 	}
+	return count
 }

nip86.go

@@ -86,10 +86,10 @@ func (rl *Relay) HandleNIP86(w http.ResponseWriter, r *http.Request) {
 			goto respond
 		}
 
-		if uTag := evt.Tags.GetFirst([]string{"u", ""}); uTag == nil || rl.getBaseURL(r) != (*uTag)[1] {
+		if uTag := evt.Tags.Find("u"); uTag == nil || rl.getBaseURL(r) != uTag[1] {
 			resp.Error = "invalid 'u' tag"
 			goto respond
-		} else if pht := evt.Tags.GetFirst([]string{"payload", hex.EncodeToString(payloadHash[:])}); pht == nil {
+		} else if pht := evt.Tags.FindWithValue("payload", hex.EncodeToString(payloadHash[:])); pht == nil {
 			resp.Error = "invalid auth event payload hash"
 			goto respond
 		} else if evt.CreatedAt < nostr.Now()-30 {

policies/nip04.go

@@ -2,7 +2,6 @@ package policies
 
 import (
 	"context"
-
 	"slices"
 
 	"github.com/fiatjaf/khatru"