Add nip 9a push support

README.md

@@ -59,6 +59,12 @@ Configures blossom support.
 
 - `enabled` - whether blossom is enabled.
 
+### `[push]`
+
+Configures NIP 9a push support.
+
+- `enabled` - whether push is enabled.
+
 ### `[roles]`
 
 Defines roles that can be assigned to different users and attendant privileges. Each role is defined by a `[roles.{role_name}]` header and has the following options:
@@ -99,6 +105,9 @@ methods = ["supportedmethods", "banpubkey", "allowpubkey"]
 [blossom]
 enabled = false
 
+[push]
+enabled = false
+
 [roles.member]
 can_invite = true
 

cmd/import/main.go

@@ -28,8 +28,8 @@ func main() {
 
 	var (
 		relay = flag.String("relay", "", "Relay name (required)")
-		reset  = flag.Bool("reset", false, "Delete all events from the store before importing")
+		reset = flag.Bool("reset", false, "Delete all events from the store before importing")
-		force  = flag.Bool("force", false, "Skip validation prompts and import valid events only")
+		force = flag.Bool("force", false, "Skip validation prompts and import valid events only")
 	)
 	flag.Parse()
 

zooid/config.go

@@ -36,6 +36,10 @@ type Config struct {
 		AutoJoin bool `toml:"auto_join"`
 	} `toml:"groups"`
 
+	Push struct {
+		Enabled bool `toml:"enabled"`
+	} `toml:"push"`
+
 	Management struct {
 		Enabled bool     `toml:"enabled"`
 		Methods []string `toml:"methods"`

zooid/instance.go

@@ -6,7 +6,6 @@ import (
 	"log"
 	"net/http"
 	"slices"
-	"strings"
 
 	"fiatjaf.com/nostr"
 	"fiatjaf.com/nostr/khatru"
@@ -20,6 +19,7 @@ type Instance struct {
 	Blossom    *BlossomStore
 	Management *ManagementStore
 	Groups     *GroupStore
+	Push       *PushManager
 }
 
 func MakeInstance(filename string) (*Instance, error) {
@@ -54,6 +54,13 @@ func MakeInstance(filename string) (*Instance, error) {
 		Management: management,
 	}
 
+	push := &PushManager{
+		Config:     config,
+		Events:     events,
+		Management: management,
+		Groups:     groups,
+	}
+
 	instance := &Instance{
 		Relay:      relay,
 		Config:     config,
@@ -61,23 +68,23 @@ func MakeInstance(filename string) (*Instance, error) {
 		Blossom:    blossom,
 		Management: management,
 		Groups:     groups,
+		Push:       push,
 	}
 
 	// NIP 11 info
 
-	// self := config.GetSelf()
+	self := config.GetSelf()
 	owner := config.GetOwner()
 
 	instance.Relay.Negentropy = true
 	instance.Relay.Info.Name = config.Info.Name
 	instance.Relay.Info.Icon = config.Info.Icon
-	// instance.Relay.Info.Self = &self
+	instance.Relay.Info.Self = &self
 	instance.Relay.Info.PubKey = &owner
 	instance.Relay.Info.Description = config.Info.Description
 	instance.Relay.Info.Software = "https://github.com/coracle-social/zooid"
 	instance.Relay.Info.Version = "v0.1.0"
 	instance.Relay.Info.SupportedNIPs = append(instance.Relay.Info.SupportedNIPs, "43")
-	instance.Relay.Info.SupportedNIPs = append(instance.Relay.Info.SupportedNIPs, "9a")
 
 	// Handlers
 
@@ -92,8 +99,9 @@ func MakeInstance(filename string) (*Instance, error) {
 	instance.Relay.OnEventSaved = instance.OnEventSaved
 	instance.Relay.OnEphemeralEvent = instance.OnEphemeralEvent
 
-	// Todo: when there's a new version of khatru
+	// Expiration
-	// instance.Relay.StartExpirationManager()
+
+	instance.Relay.StartExpirationManager(instance.Relay.QueryStored, instance.Relay.DeleteEvent)
 
 	// HTTP request handling
 
@@ -125,6 +133,10 @@ func MakeInstance(filename string) (*Instance, error) {
 		instance.Groups.Enable(instance)
 	}
 
+	if config.Push.Enabled {
+		instance.Push.Enable(instance)
+	}
+
 	// Update managed membership/admin lists
 
 	instance.Management.AllowPubkey(config.GetSelf())
@@ -142,14 +154,13 @@ func MakeInstance(filename string) (*Instance, error) {
 }
 
 func (instance *Instance) Cleanup() {
+	instance.Relay.DisableExpirationManager()
 	instance.Events.Close()
 }
 
 // Utility methods
 
-func (instance *Instance) StripSignature(ctx context.Context, event nostr.Event) nostr.Event {
+func (instance *Instance) StripSignature(pubkey nostr.PubKey, event nostr.Event) nostr.Event {
-	pubkey, _ := khatru.GetAuthed(ctx)
-
 	if instance.Config.Policy.StripSignatures && !instance.Config.CanManage(pubkey) {
 		var zeroSig [64]byte
 		event.Sig = zeroSig
@@ -181,37 +192,6 @@ func (instance *Instance) AllowRecipientEvent(event nostr.Event) bool {
 	return false
 }
 
-func (instance *Instance) IsInternalEvent(event nostr.Event) bool {
-	if event.Kind == nostr.KindApplicationSpecificData {
-		tag := event.Tags.Find("d")
-
-		if tag != nil && strings.HasPrefix(tag[1], "zooid/") {
-			return true
-		}
-	}
-
-	return false
-}
-
-func (instance *Instance) IsReadOnlyEvent(event nostr.Event) bool {
-	readOnlyEventKinds := []nostr.Kind{
-		RELAY_ADD_MEMBER,
-		RELAY_REMOVE_MEMBER,
-		RELAY_MEMBERS,
-	}
-
-	return slices.Contains(readOnlyEventKinds, event.Kind)
-}
-
-func (instance *Instance) IsWriteOnlyEvent(event nostr.Event) bool {
-	writeOnlyEventKinds := []nostr.Kind{
-		RELAY_JOIN,
-		RELAY_LEAVE,
-	}
-
-	return slices.Contains(writeOnlyEventKinds, event.Kind)
-}
-
 func (instance *Instance) GenerateInviteEvent(pubkey nostr.PubKey) nostr.Event {
 	filter := nostr.Filter{
 		Kinds: []nostr.Kind{RELAY_INVITE},
@@ -247,7 +227,7 @@ func (instance *Instance) OnConnect(ctx context.Context) {
 }
 
 func (instance *Instance) PreventBroadcast(ws *khatru.WebSocket, filter nostr.Filter, event nostr.Event) bool {
-	return instance.IsWriteOnlyEvent(event)
+	return IsWriteOnlyEvent(event)
 }
 
 func (instance *Instance) StoreEvent(ctx context.Context, event nostr.Event) error {
@@ -299,21 +279,17 @@ func (instance *Instance) QueryStored(ctx context.Context, filter nostr.Filter)
 					continue
 				}
 
-				if !yield(instance.StripSignature(ctx, event)) {
+				if !yield(instance.StripSignature(pubkey, event)) {
 					return
 				}
 			}
 
 			for event := range instance.Events.QueryEvents(filter, 1000) {
-				if event.Kind == RELAY_INVITE {
+				if !IsReadableEvent(event) {
 					continue
 				}
 
-				if instance.IsInternalEvent(event) {
+				if event.Kind == PUSH_SUBSCRIPTION && event.PubKey != pubkey {
-					continue
-				}
-
-				if instance.IsWriteOnlyEvent(event) {
 					continue
 				}
 
@@ -321,7 +297,7 @@ func (instance *Instance) QueryStored(ctx context.Context, filter nostr.Filter)
 					continue
 				}
 
-				if !yield(instance.StripSignature(ctx, event)) {
+				if !yield(instance.StripSignature(pubkey, event)) {
 					return
 				}
 			}
@@ -348,15 +324,19 @@ func (instance *Instance) OnEvent(ctx context.Context, event nostr.Event) (rejec
 		return instance.Management.ValidateJoinRequest(event)
 	}
 
+	if event.Kind == PUSH_SUBSCRIPTION {
+		return instance.Push.ValidatePushSubscription(event)
+	}
+
 	if !instance.Management.IsMember(pubkey) {
 		return true, "restricted: you are not a member of this relay"
 	}
 
-	if instance.IsInternalEvent(event) {
+	if IsInternalEvent(event) {
 		return true, "invalid: this event's kind is not accepted"
 	}
 
-	if instance.IsReadOnlyEvent(event) {
+	if IsReadOnlyEvent(event) {
 		return true, "invalid: this event's kind is not accepted"
 	}
 
@@ -407,6 +387,10 @@ func (instance *Instance) OnEventSaved(ctx context.Context, event nostr.Event) {
 	if event.Kind == nostr.KindSimpleGroupDeleteGroup {
 		instance.Groups.DeleteGroup(h)
 	}
+
+	if instance.Config.Push.Enabled && !IsWriteOnlyEvent(event) {
+		instance.Push.HandleEvent(event)
+	}
 }
 
 func (instance *Instance) OnEphemeralEvent(ctx context.Context, event nostr.Event) {
@@ -417,4 +401,8 @@ func (instance *Instance) OnEphemeralEvent(ctx context.Context, event nostr.Even
 	if event.Kind == RELAY_LEAVE {
 		instance.Management.RemoveMember(event.PubKey)
 	}
+
+	if instance.Config.Push.Enabled && !IsWriteOnlyEvent(event) {
+		instance.Push.HandleEvent(event)
+	}
 }

zooid/push.go

@@ -0,0 +1,245 @@
+package zooid
+
+import (
+	"bytes"
+	"encoding/json"
+	"log"
+	"net/http"
+	"net/url"
+	"slices"
+	"sync"
+	"time"
+
+	"fiatjaf.com/nostr"
+)
+
+// Struct definition
+
+type PushManager struct {
+	Config       *Config
+	Events       *EventStore
+	Management   *ManagementStore
+	Groups       *GroupStore
+	client       *http.Client
+	errorCounts  map[string]int // tracks consecutive errors per callback URL
+	errorCountMu sync.Mutex     // protects errorCounts map
+}
+
+type PushPayload struct {
+	ID    string       `json:"id"`
+	Relay string       `json:"relay"`
+	Event *nostr.Event `json:"event,omitempty"`
+}
+
+// Handlers
+
+func (p *PushManager) ValidatePushSubscription(event nostr.Event) (reject bool, msg string) {
+	if event.Tags.GetD() == "" {
+		return true, "invalid: missing or empty d tag"
+	}
+
+	if event.Tags.FindWithValue("relay", "wss://"+p.Config.Host+"/") == nil {
+		return true, "invalid: relay tag does not match this relay's URL"
+	}
+
+	filterTags := slices.Collect(event.Tags.FindAll("filter"))
+	if len(filterTags) == 0 {
+		return true, "invalid: at least one filter tag is required"
+	}
+
+	for _, filterTag := range filterTags {
+		if len(filterTag) < 2 {
+			return true, "invalid: filter tag is malformed"
+		}
+
+		var filter nostr.Filter
+		if err := json.Unmarshal([]byte(filterTag[1]), &filter); err != nil {
+			return true, "invalid: filter tag contains invalid JSON: " + err.Error()
+		}
+	}
+
+	for ignoreTag := range event.Tags.FindAll("ignore") {
+		if len(ignoreTag) < 2 {
+			return true, "invalid: ignore tag is malformed"
+		}
+
+		var filter nostr.Filter
+		if err := json.Unmarshal([]byte(ignoreTag[1]), &filter); err != nil {
+			return true, "invalid: ignore tag contains invalid JSON: " + err.Error()
+		}
+	}
+
+	callbackTags := slices.Collect(event.Tags.FindAll("callback"))
+
+	if len(callbackTags) < 1 {
+		return true, "invalid: missing callback tag"
+	}
+
+	if len(callbackTags) > 1 {
+		return true, "invalid: too many callback tags"
+	}
+
+	for _, callbackTag := range callbackTags {
+		if len(callbackTag) < 2 || callbackTag[1] == "" {
+			return true, "invalid: empty callback tag"
+		}
+
+		callbackURL := callbackTag[1]
+		if parsedURL, err := url.Parse(callbackURL); err != nil || (parsedURL.Scheme != "http" && parsedURL.Scheme != "https") {
+			return true, "invalid: callback must be a valid HTTP or HTTPS URL"
+		}
+	}
+
+	filter := nostr.Filter{
+		Kinds:   []nostr.Kind{PUSH_SUBSCRIPTION},
+		Authors: []nostr.PubKey{event.PubKey},
+	}
+
+	count, err := p.Events.CountEvents(filter)
+	if err != nil {
+		return true, "internal: failed to query database"
+	}
+
+	if count > 10 {
+		return true, "invalid: too many subscriptions registered"
+	}
+
+	return false, ""
+}
+
+func (p *PushManager) HandleEvent(event nostr.Event) {
+	if !IsReadableEvent(event) {
+		return
+	}
+
+	filter := nostr.Filter{
+		Kinds: []nostr.Kind{PUSH_SUBSCRIPTION},
+	}
+
+	for subscriptionEvent := range p.Events.QueryEvents(filter, 0) {
+		if event.PubKey == subscriptionEvent.PubKey {
+			continue
+		}
+
+		if p.Groups.IsGroupEvent(event) && !p.Groups.CanRead(subscriptionEvent.PubKey, event) {
+			continue
+		}
+
+		filterTags := subscriptionEvent.Tags.FindAll("filter")
+		matched := false
+		for filterTag := range filterTags {
+			if len(filterTag) < 2 {
+				continue
+			}
+
+			var filter nostr.Filter
+			if err := json.Unmarshal([]byte(filterTag[1]), &filter); err != nil {
+				continue
+			}
+
+			if filter.Matches(event) {
+				matched = true
+				break
+			}
+		}
+
+		if !matched {
+			continue
+		}
+
+		ignoreTags := subscriptionEvent.Tags.FindAll("ignore")
+		ignored := false
+		for ignoreTag := range ignoreTags {
+			if len(ignoreTag) < 2 {
+				continue
+			}
+
+			var ignore nostr.Filter
+			if err := json.Unmarshal([]byte(ignoreTag[1]), &ignore); err != nil {
+				continue
+			}
+
+			if ignore.Matches(event) {
+				ignored = true
+				break
+			}
+		}
+
+		if ignored {
+			continue
+		}
+
+		callbackTag := subscriptionEvent.Tags.Find("callback")
+
+		if callbackTag == nil || len(callbackTag) < 2 {
+			continue
+		}
+
+		callback := callbackTag[1]
+
+		payload := PushPayload{
+			ID:    event.ID.Hex(),
+			Relay: "wss://" + p.Config.Host + "/",
+		}
+
+		if subscriptionEvent.Tags.Find("include_event") != nil {
+			payload.Event = &event
+		}
+
+		payloadBytes, err := json.Marshal(payload)
+		if err != nil {
+			continue
+		}
+
+		go p.sendCallback(subscriptionEvent.ID, callback, payloadBytes)
+	}
+}
+
+func (p *PushManager) sendCallback(subscriptionID nostr.ID, callback string, payloadBytes []byte) {
+	resp, err := p.client.Post(callback, "application/json", bytes.NewReader(payloadBytes))
+	if resp != nil {
+		defer resp.Body.Close()
+	}
+
+	incrementError := func() (count int) {
+		p.errorCountMu.Lock()
+		p.errorCounts[callback]++
+		count = p.errorCounts[callback]
+		p.errorCountMu.Unlock()
+
+		return count
+	}
+
+	clearError := func() {
+		p.errorCountMu.Lock()
+		delete(p.errorCounts, callback)
+		p.errorCountMu.Unlock()
+	}
+
+	if err == nil && resp.StatusCode == 200 {
+		clearError()
+	} else if err == nil && resp.StatusCode == 404 {
+		log.Printf("Callback returned 404, deleting subscription %s", subscriptionID.Hex())
+		p.Events.DeleteEvent(subscriptionID)
+		clearError()
+	} else {
+		count := incrementError()
+
+		if count >= 10 {
+			log.Printf("Deleting subscription %s due to 10 consecutive failures", subscriptionID.Hex())
+			p.Events.DeleteEvent(subscriptionID)
+			clearError()
+		}
+	}
+}
+
+// Middleware
+
+func (p *PushManager) Enable(instance *Instance) {
+	p.client = &http.Client{
+		Timeout: 10 * time.Second,
+	}
+	p.errorCounts = make(map[string]int)
+
+	instance.Relay.Info.SupportedNIPs = append(instance.Relay.Info.SupportedNIPs, "9a")
+}

zooid/util.go

@@ -14,10 +14,59 @@ const (
 	RELAY_JOIN          = 28934
 	RELAY_INVITE        = 28935
 	RELAY_LEAVE         = 28936
+	PUSH_SUBSCRIPTION   = 30390
 	BANNED_PUBKEYS      = "zooid/banned_pubkeys"
 	BANNED_EVENTS       = "zooid/banned_events"
 )
 
+func IsInternalEvent(event nostr.Event) bool {
+	if event.Kind == nostr.KindApplicationSpecificData {
+		tag := event.Tags.Find("d")
+
+		if tag != nil && strings.HasPrefix(tag[1], "zooid/") {
+			return true
+		}
+	}
+
+	return false
+}
+
+func IsReadOnlyEvent(event nostr.Event) bool {
+	readOnlyEventKinds := []nostr.Kind{
+		RELAY_ADD_MEMBER,
+		RELAY_REMOVE_MEMBER,
+		RELAY_MEMBERS,
+	}
+
+	return slices.Contains(readOnlyEventKinds, event.Kind)
+}
+
+func IsWriteOnlyEvent(event nostr.Event) bool {
+	writeOnlyEventKinds := []nostr.Kind{
+		RELAY_JOIN,
+		RELAY_LEAVE,
+		PUSH_SUBSCRIPTION,
+	}
+
+	return slices.Contains(writeOnlyEventKinds, event.Kind)
+}
+
+func IsReadableEvent(event nostr.Event) bool {
+	if event.Kind == RELAY_INVITE {
+		return false
+	}
+
+	if IsInternalEvent(event) {
+		return false
+	}
+
+	if IsWriteOnlyEvent(event) {
+		return false
+	}
+
+	return true
+}
+
 func First[T any](s []T) T {
 	if len(s) == 0 {
 		var zero T