blossom: again a bunch of fixes. require Authorization on /upload again always.

blossom/authorization.go

@@ -0,0 +1,46 @@
+package blossom
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/nbd-wtf/go-nostr"
+)
+
+func readAuthorization(r *http.Request) (*nostr.Event, error) {
+	token := r.Header.Get("Authorization")
+	if !strings.HasPrefix(token, "Nostr ") {
+		return nil, nil
+	}
+
+	var reader io.Reader
+	reader = bytes.NewReader([]byte(token)[6:])
+	reader = base64.NewDecoder(base64.StdEncoding, reader)
+	var evt nostr.Event
+	err := json.NewDecoder(reader).Decode(&evt)
+
+	if err != nil || evt.Kind != 24242 || len(evt.ID) != 64 || !evt.CheckID() {
+		return nil, fmt.Errorf("invalid event")
+	}
+
+	if ok, _ := evt.CheckSignature(); !ok {
+		return nil, fmt.Errorf("invalid signature")
+	}
+
+	expirationTag := evt.Tags.GetFirst([]string{"expiration", ""})
+	if expirationTag == nil {
+		return nil, fmt.Errorf("missing \"expiration\" tag")
+	}
+	expiration, _ := strconv.ParseInt((*expirationTag)[1], 10, 64)
+	if nostr.Timestamp(expiration) < nostr.Now() {
+		return nil, fmt.Errorf("event expired")
+	}
+
+	return &evt, nil
+}

blossom/blob.go

@@ -0,0 +1,26 @@
+package blossom
+
+import (
+	"context"
+
+	"github.com/nbd-wtf/go-nostr"
+)
+
+type BlobDescriptor struct {
+	URL      string          `json:"url"`
+	SHA256   string          `json:"sha256"`
+	Size     int             `json:"size"`
+	Type     string          `json:"type"`
+	Uploaded nostr.Timestamp `json:"uploaded"`
+
+	Owner string `json:"-"`
+}
+
+type BlobIndex interface {
+	Keep(ctx context.Context, blob BlobDescriptor, pubkey string) error
+	List(ctx context.Context, pubkey string) (chan BlobDescriptor, error)
+	Get(ctx context.Context, sha256 string) (*BlobDescriptor, error)
+	Delete(ctx context.Context, sha256 string, pubkey string) error
+}
+
+var _ BlobIndex = (*EventStoreBlobIndexWrapper)(nil)

blossom/eventstorewrapper.go

@@ -0,0 +1,104 @@
+package blossom
+
+import (
+	"context"
+	"strconv"
+
+	"github.com/fiatjaf/eventstore"
+	"github.com/nbd-wtf/go-nostr"
+)
+
+// EventStoreBlobIndexWrapper uses fake events to keep track of what blobs we have stored and who owns them
+type EventStoreBlobIndexWrapper struct {
+	eventstore.Store
+
+	ServiceURL string
+}
+
+func (es EventStoreBlobIndexWrapper) Keep(ctx context.Context, blob BlobDescriptor, pubkey string) error {
+	ch, err := es.Store.QueryEvents(ctx, nostr.Filter{Authors: []string{pubkey}, Kinds: []int{24242}, Tags: nostr.TagMap{"x": []string{blob.SHA256}}})
+	if err != nil {
+		return err
+	}
+
+	if <-ch == nil {
+		// doesn't exist, save
+		evt := &nostr.Event{
+			PubKey: pubkey,
+			Kind:   24242,
+			Tags: nostr.Tags{
+				{"x", blob.SHA256},
+				{"type", blob.Type},
+				{"size", strconv.Itoa(blob.Size)},
+			},
+			CreatedAt: blob.Uploaded,
+		}
+		evt.ID = evt.GetID()
+		es.Store.SaveEvent(ctx, evt)
+	}
+
+	return nil
+}
+
+func (es EventStoreBlobIndexWrapper) List(ctx context.Context, pubkey string) (chan BlobDescriptor, error) {
+	ech, err := es.Store.QueryEvents(ctx, nostr.Filter{Authors: []string{pubkey}, Kinds: []int{24242}})
+	if err != nil {
+		return nil, err
+	}
+
+	ch := make(chan BlobDescriptor)
+
+	go func() {
+		for evt := range ech {
+			ch <- es.parseEvent(evt)
+		}
+		close(ch)
+	}()
+
+	return ch, nil
+}
+
+func (es EventStoreBlobIndexWrapper) Get(ctx context.Context, sha256 string) (*BlobDescriptor, error) {
+	ech, err := es.Store.QueryEvents(ctx, nostr.Filter{Tags: nostr.TagMap{"x": []string{sha256}}, Kinds: []int{24242}, Limit: 1})
+	if err != nil {
+		return nil, err
+	}
+
+	evt := <-ech
+	if evt != nil {
+		bd := es.parseEvent(evt)
+		return &bd, nil
+	}
+
+	return nil, nil
+}
+
+func (es EventStoreBlobIndexWrapper) Delete(ctx context.Context, sha256 string, pubkey string) error {
+	ech, err := es.Store.QueryEvents(ctx, nostr.Filter{Authors: []string{pubkey}, Tags: nostr.TagMap{"x": []string{sha256}}, Kinds: []int{24242}, Limit: 1})
+	if err != nil {
+		return err
+	}
+
+	evt := <-ech
+	if evt != nil {
+		return es.Store.DeleteEvent(ctx, evt)
+	}
+
+	return nil
+}
+
+func (es EventStoreBlobIndexWrapper) parseEvent(evt *nostr.Event) BlobDescriptor {
+	hhash := evt.Tags[0][1]
+	mimetype := evt.Tags[1][1]
+	ext := getExtension(mimetype)
+	size, _ := strconv.Atoi(evt.Tags[2][1])
+
+	return BlobDescriptor{
+		Owner:    evt.PubKey,
+		Uploaded: evt.CreatedAt,
+		URL:      es.ServiceURL + "/" + hhash + ext,
+		SHA256:   hhash,
+		Type:     mimetype,
+		Size:     size,
+	}
+}

blossom/handlers.go

@@ -0,0 +1,323 @@
+package blossom
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"mime"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/liamg/magic"
+	"github.com/nbd-wtf/go-nostr"
+)
+
+func (bs BlossomServer) handleUploadCheck(w http.ResponseWriter, r *http.Request) {
+	auth, err := readAuthorization(r)
+	if err != nil {
+		blossomError(w, err.Error(), 400)
+		return
+	}
+	if auth == nil {
+		blossomError(w, "missing \"Authorization\" header", 400)
+		return
+	}
+	if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
+		blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
+		return
+	}
+
+	mimetype := r.Header.Get("X-Content-Type")
+	exts, _ := mime.ExtensionsByType(mimetype)
+	var ext string
+	if len(exts) > 0 {
+		ext = exts[0]
+	}
+
+	// get the file size from the incoming header
+	size, _ := strconv.Atoi(r.Header.Get("X-Content-Length"))
+
+	for _, rb := range bs.RejectUpload {
+		reject, reason, code := rb(r.Context(), auth, size, ext)
+		if reject {
+			blossomError(w, reason, code)
+			return
+		}
+	}
+}
+
+func (bs BlossomServer) handleUpload(w http.ResponseWriter, r *http.Request) {
+	auth, err := readAuthorization(r)
+	if err != nil {
+		blossomError(w, "invalid \"Authorization\": "+err.Error(), 400)
+		return
+	}
+	if auth == nil {
+		blossomError(w, "missing \"Authorization\" header", 400)
+		return
+	}
+	if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
+		blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
+		return
+	}
+
+	// get the file size from the incoming header
+	size, _ := strconv.Atoi(r.Header.Get("Content-Length"))
+	if size == 0 {
+		blossomError(w, "missing \"Content-Length\" header", 400)
+		return
+	}
+
+	// read first bytes of upload so we can find out the filetype
+	b := make([]byte, min(50, size), size)
+	if _, err = r.Body.Read(b); err != nil {
+		blossomError(w, "failed to read initial bytes of upload body: "+err.Error(), 400)
+		return
+	}
+	var ext string
+	if ft, _ := magic.Lookup(b); ft != nil {
+		ext = "." + ft.Extension
+	} else {
+		// if we can't find, use the filetype given by the upload header
+		mimetype := r.Header.Get("Content-Type")
+		ext = getExtension(mimetype)
+	}
+
+	// run the reject hooks
+	for _, ru := range bs.RejectUpload {
+		reject, reason, code := ru(r.Context(), auth, size, ext)
+		if reject {
+			blossomError(w, reason, code)
+			return
+		}
+	}
+
+	// if it passes then we have to read the entire thing into memory so we can compute the sha256
+	for {
+		var n int
+		n, err = r.Body.Read(b[len(b):cap(b)])
+		b = b[:len(b)+n]
+		if err != nil {
+			if err == io.EOF {
+				err = nil
+			}
+			break
+		}
+		if len(b) == cap(b) {
+			// add more capacity (let append pick how much)
+			// if Content-Length was correct we shouldn't reach this
+			b = append(b, 0)[:len(b)]
+		}
+	}
+	if err != nil {
+		blossomError(w, "failed to read upload body: "+err.Error(), 400)
+		return
+	}
+
+	hash := sha256.Sum256(b)
+	hhash := hex.EncodeToString(hash[:])
+
+	// keep track of the blob descriptor
+	bd := BlobDescriptor{
+		URL:      bs.ServiceURL + "/" + hhash + ext,
+		SHA256:   hhash,
+		Size:     len(b),
+		Type:     mime.TypeByExtension(ext),
+		Uploaded: nostr.Now(),
+	}
+	if err := bs.Store.Keep(r.Context(), bd, auth.PubKey); err != nil {
+		blossomError(w, "failed to save event: "+err.Error(), 400)
+		return
+	}
+
+	// save actual blob
+	for _, sb := range bs.StoreBlob {
+		if err := sb(r.Context(), hhash, b); err != nil {
+			blossomError(w, "failed to save: "+err.Error(), 500)
+			return
+		}
+	}
+
+	// return response
+	json.NewEncoder(w).Encode(bd)
+}
+
+func (bs BlossomServer) handleGetBlob(w http.ResponseWriter, r *http.Request) {
+	spl := strings.SplitN(r.URL.Path, ".", 2)
+	hhash := spl[0]
+	if len(hhash) != 65 {
+		blossomError(w, "invalid /<sha256>[.ext] path", 400)
+		return
+	}
+	hhash = hhash[1:]
+
+	// check for an authorization tag, if any
+	auth, err := readAuthorization(r)
+	if err != nil {
+		blossomError(w, err.Error(), 400)
+		return
+	}
+
+	// if there is one, we check if it has the extra requirements
+	if auth != nil {
+		if auth.Tags.GetFirst([]string{"t", "get"}) == nil {
+			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
+			return
+		}
+
+		if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
+			auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
+			blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
+			return
+		}
+	}
+
+	for _, rg := range bs.RejectGet {
+		reject, reason, code := rg(r.Context(), auth, hhash)
+		if reject {
+			blossomError(w, reason, code)
+			return
+		}
+	}
+
+	var ext string
+	if len(spl) == 2 {
+		ext = "." + spl[1]
+	}
+
+	for _, lb := range bs.LoadBlob {
+		reader, _ := lb(r.Context(), hhash)
+		if reader != nil {
+			w.Header().Add("Content-Type", mime.TypeByExtension(ext))
+			io.Copy(w, reader)
+			return
+		}
+	}
+
+	blossomError(w, "file not found", 404)
+	return
+}
+
+func (bs BlossomServer) handleHasBlob(w http.ResponseWriter, r *http.Request) {
+	spl := strings.SplitN(r.URL.Path, ".", 2)
+	hhash := spl[0]
+	if len(hhash) != 65 {
+		blossomError(w, "invalid /<sha256>[.ext] path", 400)
+		return
+	}
+	hhash = hhash[1:]
+
+	bd, err := bs.Store.Get(r.Context(), hhash)
+	if err != nil {
+		blossomError(w, "failed to query: "+err.Error(), 500)
+		return
+	}
+
+	if bd == nil {
+		blossomError(w, "file not found", 404)
+		return
+	}
+
+	return
+}
+
+func (bs BlossomServer) handleList(w http.ResponseWriter, r *http.Request) {
+	// check for an authorization tag, if any
+	auth, err := readAuthorization(r)
+	if err != nil {
+		blossomError(w, err.Error(), 400)
+		return
+	}
+
+	// if there is one, we check if it has the extra requirements
+	if auth != nil {
+		if auth.Tags.GetFirst([]string{"t", "list"}) == nil {
+			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
+			return
+		}
+	}
+
+	pubkey := r.URL.Path[6:]
+
+	for _, rl := range bs.RejectList {
+		reject, reason, code := rl(r.Context(), auth, pubkey)
+		if reject {
+			blossomError(w, reason, code)
+			return
+		}
+	}
+
+	ch, err := bs.Store.List(r.Context(), pubkey)
+	if err != nil {
+		blossomError(w, "failed to query: "+err.Error(), 500)
+		return
+	}
+
+	w.Write([]byte{'['})
+	enc := json.NewEncoder(w)
+	for bd := range ch {
+		enc.Encode(bd)
+	}
+	w.Write([]byte{']'})
+}
+
+func (bs BlossomServer) handleDelete(w http.ResponseWriter, r *http.Request) {
+	auth, err := readAuthorization(r)
+	if err != nil {
+		blossomError(w, err.Error(), 400)
+		return
+	}
+
+	if auth != nil {
+		if auth.Tags.GetFirst([]string{"t", "delete"}) == nil {
+			blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
+			return
+		}
+	}
+
+	spl := strings.SplitN(r.URL.Path, ".", 2)
+	hhash := spl[0]
+	if len(hhash) != 65 {
+		blossomError(w, "invalid /<sha256>[.ext] path", 400)
+		return
+	}
+	hhash = hhash[1:]
+	if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
+		auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
+		blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
+		return
+	}
+
+	// should we accept this delete?
+	for _, rd := range bs.RejectDelete {
+		reject, reason, code := rd(r.Context(), auth, hhash)
+		if reject {
+			blossomError(w, reason, code)
+			return
+		}
+	}
+
+	// delete the entry that links this blob to this author
+	if err := bs.Store.Delete(r.Context(), hhash, auth.PubKey); err != nil {
+		blossomError(w, "delete of blob entry failed: "+err.Error(), 500)
+		return
+	}
+
+	// we will actually only delete the file if no one else owns it
+	if bd, err := bs.Store.Get(r.Context(), hhash); err == nil && bd == nil {
+		for _, del := range bs.DeleteBlob {
+			if err := del(r.Context(), hhash); err != nil {
+				blossomError(w, "failed to delete blob: "+err.Error(), 500)
+				return
+			}
+		}
+	}
+}
+
+func (bs BlossomServer) handleMirror(w http.ResponseWriter, r *http.Request) {
+}
+
+func (bs BlossomServer) handleNegentropy(w http.ResponseWriter, r *http.Request) {
+}

blossom/server.go

@@ -0,0 +1,76 @@
+package blossom
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+
+	"github.com/fiatjaf/khatru"
+	"github.com/nbd-wtf/go-nostr"
+)
+
+type BlossomServer struct {
+	ServiceURL string
+	Store      BlobIndex
+
+	StoreBlob  []func(ctx context.Context, sha256 string, body []byte) error
+	LoadBlob   []func(ctx context.Context, sha256 string) (io.Reader, error)
+	DeleteBlob []func(ctx context.Context, sha256 string) error
+
+	RejectUpload []func(ctx context.Context, auth *nostr.Event, size int, ext string) (bool, string, int)
+	RejectGet    []func(ctx context.Context, auth *nostr.Event, sha256 string) (bool, string, int)
+	RejectList   []func(ctx context.Context, auth *nostr.Event, pubkey string) (bool, string, int)
+	RejectDelete []func(ctx context.Context, auth *nostr.Event, sha256 string) (bool, string, int)
+}
+
+func New(rl *khatru.Relay, serviceURL string) *BlossomServer {
+	bs := &BlossomServer{
+		ServiceURL: serviceURL,
+	}
+
+	base := rl.Router()
+	mux := http.NewServeMux()
+
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/upload" {
+			if r.Method == "PUT" {
+				setCors(w)
+				bs.handleUpload(w, r)
+				return
+			} else if r.Method == "HEAD" {
+				setCors(w)
+				bs.handleUploadCheck(w, r)
+				return
+			}
+		}
+
+		if strings.HasPrefix(r.URL.Path, "/list/") && r.Method == "GET" {
+			setCors(w)
+			bs.handleList(w, r)
+			return
+		}
+
+		if len(strings.SplitN(r.URL.Path, ".", 2)[0]) == 65 {
+			if r.Method == "HEAD" {
+				setCors(w)
+				bs.handleHasBlob(w, r)
+				return
+			} else if r.Method == "GET" {
+				setCors(w)
+				bs.handleGetBlob(w, r)
+				return
+			} else if r.Method == "DELETE" {
+				setCors(w)
+				bs.handleDelete(w, r)
+				return
+			}
+		}
+
+		base.ServeHTTP(w, r)
+	})
+
+	rl.SetRouter(mux)
+
+	return bs
+}

blossom/utils.go

@@ -0,0 +1,43 @@
+package blossom
+
+import (
+	"mime"
+	"net/http"
+)
+
+func setCors(w http.ResponseWriter) {
+	w.Header().Set("Access-Control-Allow-Origin", "*")
+	w.Header().Set("Access-Control-Allow-Headers", "Authorization")
+	w.Header().Set("Access-Control-Allow-Methods", "GET PUT DELETE")
+}
+
+func blossomError(w http.ResponseWriter, msg string, code int) {
+	w.Header().Add("X-Reason", msg)
+	w.WriteHeader(code)
+}
+
+func getExtension(mimetype string) string {
+	if mimetype == "" {
+		return ""
+	}
+
+	switch mimetype {
+	case "image/jpeg":
+		return ".jpg"
+	case "image/gif":
+		return ".gif"
+	case "image/png":
+		return ".png"
+	case "image/webp":
+		return ".webp"
+	case "video/mp4":
+		return ".mp4"
+	}
+
+	exts, _ := mime.ExtensionsByType(mimetype)
+	if len(exts) > 0 {
+		return exts[0]
+	}
+
+	return ""
+}

examples/basic-badger/main.go

@@ -20,6 +20,7 @@ func main() {
 	relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
 	relay.CountEvents = append(relay.CountEvents, db.CountEvents)
 	relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)
+	relay.Negentropy = true
 
 	fmt.Println("running on :3334")
 	http.ListenAndServe(":3334", relay)

examples/blossom/main.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/fiatjaf/eventstore/badger"
+	"github.com/fiatjaf/khatru"
+	"github.com/fiatjaf/khatru/blossom"
+)
+
+func main() {
+	relay := khatru.NewRelay()
+
+	db := &badger.BadgerBackend{Path: "/tmp/khatru-badger-blossom-tmp"}
+	if err := db.Init(); err != nil {
+		panic(err)
+	}
+
+	relay.StoreEvent = append(relay.StoreEvent, db.SaveEvent)
+	relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
+	relay.CountEvents = append(relay.CountEvents, db.CountEvents)
+	relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)
+
+	bl := blossom.New(relay, "http://localhost:3334")
+	bl.Store = blossom.EventStoreBlobIndexWrapper{Store: db, ServiceURL: bl.ServiceURL}
+	bl.StoreBlob = append(bl.StoreBlob, func(ctx context.Context, sha256 string, body []byte) error {
+		fmt.Println("storing", sha256, len(body))
+		return nil
+	})
+	bl.LoadBlob = append(bl.LoadBlob, func(ctx context.Context, sha256 string) ([]byte, error) {
+		fmt.Println("loading", sha256)
+		return []byte("aaaaa"), nil
+	})
+
+	fmt.Println("running on :3334")
+	http.ListenAndServe(":3334", relay)
+}

get-started.go

@@ -15,6 +15,10 @@ func (rl *Relay) Router() *http.ServeMux {
 	return rl.serveMux
 }
 
+func (rl *Relay) SetRouter(mux *http.ServeMux) {
+	rl.serveMux = mux
+}
+
 // Start creates an http server and starts listening on given host and port.
 func (rl *Relay) Start(host string, port int, started ...chan bool) error {
 	addr := net.JoinHostPort(host, strconv.Itoa(port))

go.mod

@@ -3,8 +3,10 @@ module github.com/fiatjaf/khatru
 go 1.23.1
 
 require (
+	github.com/bep/debounce v1.2.1
 	github.com/fasthttp/websocket v1.5.7
 	github.com/fiatjaf/eventstore v0.12.0
+	github.com/liamg/magic v0.0.1
 	github.com/nbd-wtf/go-nostr v0.40.0
 	github.com/puzpuzpuz/xsync/v3 v3.4.0
 	github.com/rs/cors v1.7.0
@@ -18,6 +20,7 @@ require (
 	github.com/aquasecurity/esquery v0.2.0 // indirect
 	github.com/btcsuite/btcd/btcec/v2 v2.3.4 // indirect
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0 // indirect
+	github.com/cespare/xxhash v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/decred/dcrd/crypto/blake256 v1.1.0 // indirect
@@ -36,6 +39,7 @@ require (
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/flatbuffers v24.3.25+incompatible // indirect
+	github.com/greatroar/blobloom v0.8.0 // indirect
 	github.com/jmoiron/sqlx v1.3.5 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/klauspost/compress v1.17.10 // indirect

go.sum

@@ -2,17 +2,23 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMT
 fiatjaf.com/lib v0.2.0 h1:TgIJESbbND6GjOgGHxF5jsO6EMjuAxIzZHPo5DXYexs=
 fiatjaf.com/lib v0.2.0/go.mod h1:Ycqq3+mJ9jAWu7XjbQI1cVr+OFgnHn79dQR5oTII47g=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/OneOfOne/xxhash v1.2.2 h1:KMrpdQIwFcEqXDklaen+P1axHaj9BSKzvpUUfnHldSE=
+github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/PowerDNS/lmdb-go v1.9.2 h1:Cmgerh9y3ZKBZGz1irxSShhfmFyRUh+Zdk4cZk7ZJvU=
 github.com/PowerDNS/lmdb-go v1.9.2/go.mod h1:TE0l+EZK8Z1B4dx070ZxkWTlp8RG1mjN0/+FkFRQMtU=
 github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs=
 github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
 github.com/aquasecurity/esquery v0.2.0 h1:9WWXve95TE8hbm3736WB7nS6Owl8UGDeu+0jiyE9ttA=
 github.com/aquasecurity/esquery v0.2.0/go.mod h1:VU+CIFR6C+H142HHZf9RUkp4Eedpo9UrEKeCQHWf9ao=
+github.com/bep/debounce v1.2.1 h1:v67fRdBA9UQu2NhLFXrSg0Brw7CexQekrBwDMM8bzeY=
+github.com/bep/debounce v1.2.1/go.mod h1:H8yggRPQKLUhUoqrJC1bO2xNya7vanpDl7xR3ISbCJ0=
 github.com/btcsuite/btcd/btcec/v2 v2.3.4 h1:3EJjcN70HCu/mwqlUsGK8GcNVyLVxFDlWurTXGPFfiQ=
 github.com/btcsuite/btcd/btcec/v2 v2.3.4/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
 github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0 h1:59Kx4K6lzOW5w6nFlA0v5+lk/6sjybR934QNHSJZPTQ=
 github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0/go.mod h1:7SFka0XMvUgj3hfZtydOrQY2mwhPclbT2snogU7SQQc=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
+github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
@@ -87,6 +93,8 @@ github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/greatroar/blobloom v0.8.0 h1:I9RlEkfqK9/6f1v9mFmDYegDQ/x0mISCpiNpAm23Pt4=
+github.com/greatroar/blobloom v0.8.0/go.mod h1:mjMJ1hh1wjGVfr93QIHJ6FfDNVrA0IELv8OvMHJxHKs=
 github.com/jgroeneveld/schema v1.0.0 h1:J0E10CrOkiSEsw6dfb1IfrDJD14pf6QLVJ3tRPl/syI=
 github.com/jgroeneveld/schema v1.0.0/go.mod h1:M14lv7sNMtGvo3ops1MwslaSYgDYxrSmbzWIQ0Mr5rs=
 github.com/jgroeneveld/trial v2.0.0+incompatible h1:d59ctdgor+VqdZCAiUfVN8K13s0ALDioG5DWwZNtRuQ=
@@ -103,6 +111,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/liamg/magic v0.0.1 h1:Ru22ElY+sCh6RvRTWjQzKKCxsEco8hE0co8n1qe7TBM=
+github.com/liamg/magic v0.0.1/go.mod h1:yQkOmZZI52EA+SQ2xyHpVw8fNvTBruF873Y+Vt6S+fk=
 github.com/lib/pq v1.2.0/go.mod h1:5WUZQaWbwv1U+lTReE5YruASi9Al49XbQIvNi/34Woo=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
@@ -126,6 +136,8 @@ github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee h1:8Iv5m6xEo1NR1AvpV+7XmhI4r39LGNzwUL4YpMuL5vk=
 github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee/go.mod h1:qwtSXrKuJh/zsFQ12yEE89xfCrGKK63Rr7ctU/uCo4g=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -225,5 +237,7 @@ gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EV
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
+gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

handlers.go

@@ -10,9 +10,13 @@ import (
 	"sync"
 	"time"
 
+	"github.com/bep/debounce"
 	"github.com/fasthttp/websocket"
 	"github.com/nbd-wtf/go-nostr"
 	"github.com/nbd-wtf/go-nostr/nip42"
+	"github.com/nbd-wtf/go-nostr/nip77"
+	"github.com/nbd-wtf/go-nostr/nip77/negentropy"
+	"github.com/puzpuzpuz/xsync/v3"
 	"github.com/rs/cors"
 )
 
@@ -54,9 +58,10 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 	rand.Read(challenge)
 
 	ws := &WebSocket{
-		conn:      conn,
-		Request:   r,
-		Challenge: hex.EncodeToString(challenge),
+		conn:               conn,
+		Request:            r,
+		Challenge:          hex.EncodeToString(challenge),
+		negentropySessions: xsync.NewMapOf[string, *NegentropySession](),
 	}
 	ws.Context, ws.cancel = context.WithCancel(context.Background())
 
@@ -123,8 +128,14 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 			go func(message []byte) {
 				envelope := nostr.ParseMessage(message)
 				if envelope == nil {
-					// stop silently
-					return
+					if !rl.Negentropy {
+						// stop silently
+						return
+					}
+					envelope = nip77.ParseNegMessage(message)
+					if envelope == nil {
+						return
+					}
 				}
 
 				switch env := envelope.(type) {
@@ -272,6 +283,75 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 					} else {
 						ws.WriteJSON(nostr.OKEnvelope{EventID: env.Event.ID, OK: false, Reason: "error: failed to authenticate"})
 					}
+				case *nip77.OpenEnvelope:
+					srl := rl
+					if rl.getSubRelayFromFilter != nil {
+						srl = rl.getSubRelayFromFilter(env.Filter)
+						if !srl.Negentropy {
+							// ignore
+							return
+						}
+					}
+					vec, err := srl.startNegentropySession(ctx, env.Filter)
+					if err != nil {
+						// fail everything if any filter is rejected
+						reason := err.Error()
+						if strings.HasPrefix(reason, "auth-required:") {
+							RequestAuth(ctx)
+						}
+						ws.WriteJSON(nip77.ErrorEnvelope{SubscriptionID: env.SubscriptionID, Reason: reason})
+						return
+					}
+
+					// reconcile to get the next message and return it
+					neg := negentropy.New(vec, 1024*1024)
+					out, err := neg.Reconcile(env.Message)
+					if err != nil {
+						ws.WriteJSON(nip77.ErrorEnvelope{SubscriptionID: env.SubscriptionID, Reason: err.Error()})
+						return
+					}
+					ws.WriteJSON(nip77.MessageEnvelope{SubscriptionID: env.SubscriptionID, Message: out})
+
+					// if the message is not empty that means we'll probably have more reconciliation sessions, so store this
+					if out != "" {
+						deb := debounce.New(time.Second * 7)
+						negSession := &NegentropySession{
+							neg: neg,
+							postponeClose: func() {
+								deb(func() {
+									ws.negentropySessions.Delete(env.SubscriptionID)
+								})
+							},
+						}
+						negSession.postponeClose()
+
+						ws.negentropySessions.Store(env.SubscriptionID, negSession)
+					}
+				case *nip77.MessageEnvelope:
+					negSession, ok := ws.negentropySessions.Load(env.SubscriptionID)
+					if !ok {
+						// bad luck, your request was destroyed
+						ws.WriteJSON(nip77.ErrorEnvelope{SubscriptionID: env.SubscriptionID, Reason: "CLOSED"})
+						return
+					}
+					// reconcile to get the next message and return it
+					out, err := negSession.neg.Reconcile(env.Message)
+					if err != nil {
+						ws.WriteJSON(nip77.ErrorEnvelope{SubscriptionID: env.SubscriptionID, Reason: err.Error()})
+						ws.negentropySessions.Delete(env.SubscriptionID)
+						return
+					}
+					ws.WriteJSON(nip77.MessageEnvelope{SubscriptionID: env.SubscriptionID, Message: out})
+
+					// if there is more reconciliation to do, postpone this
+					if out != "" {
+						negSession.postponeClose()
+					} else {
+						// otherwise we can just close it
+						ws.negentropySessions.Delete(env.SubscriptionID)
+					}
+				case *nip77.CloseEnvelope:
+					ws.negentropySessions.Delete(env.SubscriptionID)
 				}
 			}(message)
 		}

negentropy.go

@@ -0,0 +1,50 @@
+package khatru
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/nbd-wtf/go-nostr"
+	"github.com/nbd-wtf/go-nostr/nip77/negentropy"
+	"github.com/nbd-wtf/go-nostr/nip77/negentropy/storage/vector"
+)
+
+type NegentropySession struct {
+	neg           *negentropy.Negentropy
+	postponeClose func()
+}
+
+func (rl *Relay) startNegentropySession(ctx context.Context, filter nostr.Filter) (*vector.Vector, error) {
+	// do the same overwrite/reject flow we do in normal REQs
+	for _, ovw := range rl.OverwriteFilter {
+		ovw(ctx, &filter)
+	}
+	if filter.LimitZero {
+		return nil, fmt.Errorf("invalid limit 0")
+	}
+	for _, reject := range rl.RejectFilter {
+		if reject, msg := reject(ctx, filter); reject {
+			return nil, errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
+		}
+	}
+
+	// fetch events and add them to a negentropy Vector store
+	vec := vector.New()
+	for _, query := range rl.QueryEvents {
+		ch, err := query(ctx, filter)
+		if err != nil {
+			continue
+		} else if ch == nil {
+			continue
+		}
+
+		for event := range ch {
+			// since the goal here is to sync databases we won't do fancy stuff like overwrite events
+			vec.Insert(event.CreatedAt, event.ID)
+		}
+	}
+	vec.Seal()
+
+	return vec, nil
+}

nip11.go

@@ -11,10 +11,13 @@ func (rl *Relay) HandleNIP11(w http.ResponseWriter, r *http.Request) {
 	info := *rl.Info
 
 	if len(rl.DeleteEvent) > 0 {
-		info.SupportedNIPs = append(info.SupportedNIPs, 9)
+		info.AddSupportedNIP(9)
 	}
 	if len(rl.CountEvents) > 0 {
-		info.SupportedNIPs = append(info.SupportedNIPs, 45)
+		info.AddSupportedNIP(45)
+	}
+	if rl.Negentropy {
+		info.AddSupportedNIP(77)
 	}
 
 	for _, ovw := range rl.OverwriteRelayInformation {

relay.go

@@ -46,7 +46,7 @@ func NewRelay() *Relay {
 type Relay struct {
 	ServiceURL string
 
-	// these structs keeps track of all the things that can be customized when handling events or requests
+	// hooks that will be called at various times
 	RejectEvent               []func(ctx context.Context, event *nostr.Event) (reject bool, msg string)
 	OverwriteDeletionOutcome  []func(ctx context.Context, target *nostr.Event, deletion *nostr.Event) (acceptDeletion bool, msg string)
 	StoreEvent                []func(ctx context.Context, event *nostr.Event) error
@@ -90,6 +90,9 @@ type Relay struct {
 	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

websocket.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/fasthttp/websocket"
+	"github.com/puzpuzpuz/xsync/v3"
 )
 
 type WebSocket struct {
@@ -24,6 +25,9 @@ type WebSocket struct {
 	AuthedPublicKey string
 	Authed          chan struct{}
 
+	// nip77
+	negentropySessions *xsync.MapOf[string, *NegentropySession]
+
 	authLock sync.Mutex
 }