eventstore: replace bluge with bleve.

bluge seems to be abandoned and bleve should work better, who knows.

eventstore/bleve/bleve_test.go

@@ -0,0 +1,73 @@
+package bleve
+
+import (
+	"os"
+	"testing"
+
+	"fiatjaf.com/nostr"
+	"fiatjaf.com/nostr/eventstore/lmdb"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBleveFlow(t *testing.T) {
+	os.RemoveAll("/tmp/blevetest-lmdb")
+	os.RemoveAll("/tmp/blevetest-bleve")
+
+	bb := &lmdb.LMDBBackend{Path: "/tmp/blevetest-lmdb"}
+	bb.Init()
+	defer bb.Close()
+
+	bl := BleveBackend{
+		Path:          "/tmp/blevetest-bleve",
+		RawEventStore: bb,
+	}
+	bl.Init()
+	defer bl.Close()
+
+	willDelete := make([]nostr.Event, 0, 3)
+
+	for i, content := range []string{
+		"good morning mr paper maker",
+		"good night",
+		"I'll see you again in the paper house",
+		"tonight we dine in my house",
+		"the paper in this house if very good, mr",
+	} {
+		evt := nostr.Event{Content: content, Tags: nostr.Tags{}}
+		evt.Sign(nostr.MustSecretKeyFromHex("0000000000000000000000000000000000000000000000000000000000000001"))
+
+		bb.SaveEvent(evt)
+		bl.SaveEvent(evt)
+
+		if i%2 == 0 {
+			willDelete = append(willDelete, evt)
+		}
+	}
+
+	{
+		n := 0
+		t.Logf("searching for 'good' (should find 3)")
+		for range bl.QueryEvents(nostr.Filter{Search: "good"}, 400) {
+			n++
+		}
+		t.Logf("found %d results", n)
+		assert.Equal(t, 3, n)
+	}
+
+	for _, evt := range willDelete {
+		bl.DeleteEvent(evt.ID)
+	}
+
+	{
+		n := 0
+		for evt := range bl.QueryEvents(nostr.Filter{Search: "good"}, 400) {
+			n++
+			assert.Equal(t, evt.Content, "good night")
+			assert.Equal(t,
+				nostr.MustPubKeyFromHex("79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"),
+				evt.PubKey,
+			)
+		}
+		assert.Equal(t, 1, n)
+	}
+}

eventstore/bleve/delete.go

@@ -0,0 +1,9 @@
+package bleve
+
+import (
+	"fiatjaf.com/nostr"
+)
+
+func (b *BleveBackend) DeleteEvent(id nostr.ID) error {
+	return b.index.Delete(id.Hex())
+}

eventstore/bleve/helpers.go

@@ -0,0 +1,9 @@
+package bleve
+
+const (
+	idField        = "i"
+	contentField   = "c"
+	kindField      = "k"
+	createdAtField = "a"
+	pubkeyField    = "p"
+)

eventstore/bleve/lib.go

@@ -0,0 +1,61 @@
+package bleve
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"fiatjaf.com/nostr"
+	"fiatjaf.com/nostr/eventstore"
+	bleve "github.com/blevesearch/bleve/v2"
+	bleveMapping "github.com/blevesearch/bleve/v2/mapping"
+)
+
+var _ eventstore.Store = (*BleveBackend)(nil)
+
+type BleveBackend struct {
+	sync.Mutex
+	// Path is where the index will be saved
+	Path string
+
+	// RawEventStore is where we'll fetch the raw events from
+	// bleve will only store ids, so the actual events must be somewhere else
+	RawEventStore eventstore.Store
+
+	index bleve.Index
+}
+
+func (b *BleveBackend) Close() {
+	if b.index != nil {
+		b.index.Close()
+	}
+}
+
+func (b *BleveBackend) Init() error {
+	if b.Path == "" {
+		return fmt.Errorf("missing Path")
+	}
+	if b.RawEventStore == nil {
+		return fmt.Errorf("missing RawEventStore")
+	}
+
+	// try to open existing index
+	index, err := bleve.Open(b.Path)
+	if err == bleve.ErrorIndexPathDoesNotExist {
+		// create new index with default mapping
+		mapping := bleveMapping.NewIndexMapping()
+		index, err = bleve.New(b.Path, mapping)
+		if err != nil {
+			return fmt.Errorf("error creating index: %w", err)
+		}
+	} else if err != nil {
+		return fmt.Errorf("error opening index: %w", err)
+	}
+
+	b.index = index
+	return nil
+}
+
+func (b *BleveBackend) CountEvents(nostr.Filter) (uint32, error) {
+	return 0, errors.New("not supported")
+}

eventstore/bleve/query.go

@@ -0,0 +1,95 @@
+package bleve
+
+import (
+	"iter"
+	"strconv"
+
+	"fiatjaf.com/nostr"
+	bleve "github.com/blevesearch/bleve/v2"
+	"github.com/blevesearch/bleve/v2/search/query"
+)
+
+func (b *BleveBackend) QueryEvents(filter nostr.Filter, maxLimit int) iter.Seq[nostr.Event] {
+	return func(yield func(nostr.Event) bool) {
+		limit := maxLimit
+		if filter.LimitZero {
+			return
+		} else if filter.Limit > 0 && filter.Limit < limit {
+			limit = filter.Limit
+		}
+
+		if len(filter.Search) < 2 {
+			return
+		}
+
+		searchQ := bleve.NewMatchQuery(filter.Search)
+		searchQ.SetField(contentField)
+		var q query.Query = searchQ
+
+		conjQueries := []query.Query{searchQ}
+
+		if len(filter.Kinds) > 0 {
+			eitherKind := bleve.NewDisjunctionQuery()
+			for _, kind := range filter.Kinds {
+				kindQ := bleve.NewTermQuery(strconv.Itoa(int(kind)))
+				kindQ.SetField(kindField)
+				eitherKind.AddQuery(kindQ)
+			}
+			conjQueries = append(conjQueries, eitherKind)
+		}
+
+		if len(filter.Authors) > 0 {
+			eitherPubkey := bleve.NewDisjunctionQuery()
+			for _, pubkey := range filter.Authors {
+				if len(pubkey) != 64 {
+					continue
+				}
+				pubkeyQ := bleve.NewTermQuery(pubkey.Hex()[56:])
+				pubkeyQ.SetField(pubkeyField)
+				eitherPubkey.AddQuery(pubkeyQ)
+			}
+			conjQueries = append(conjQueries, eitherPubkey)
+		}
+
+		if filter.Since != 0 || filter.Until != 0 {
+			var min *float64
+			if filter.Since != 0 {
+				minVal := float64(filter.Since)
+				min = &minVal
+			}
+			var max *float64
+			if filter.Until != 0 {
+				maxVal := float64(filter.Until)
+				max = &maxVal
+			}
+			dateRangeQ := bleve.NewNumericRangeInclusiveQuery(min, max, nil, nil)
+			dateRangeQ.SetField(createdAtField)
+			conjQueries = append(conjQueries, dateRangeQ)
+		}
+
+		if len(conjQueries) > 1 {
+			q = bleve.NewConjunctionQuery(conjQueries...)
+		}
+
+		req := bleve.NewSearchRequest(q)
+		req.Size = limit
+		req.From = 0
+
+		result, err := b.index.Search(req)
+		if err != nil {
+			return
+		}
+
+		for _, hit := range result.Hits {
+			id, err := nostr.IDFromHex(hit.ID)
+			if err != nil {
+				continue
+			}
+			for evt := range b.RawEventStore.QueryEvents(nostr.Filter{IDs: []nostr.ID{id}}, 1) {
+				if !yield(evt) {
+					return
+				}
+			}
+		}
+	}
+}

eventstore/bleve/replace.go

@@ -0,0 +1,38 @@
+package bleve
+
+import (
+	"fmt"
+
+	"fiatjaf.com/nostr"
+	"fiatjaf.com/nostr/eventstore"
+	"fiatjaf.com/nostr/eventstore/internal"
+)
+
+func (b *BleveBackend) ReplaceEvent(evt nostr.Event) error {
+	b.Lock()
+	defer b.Unlock()
+
+	filter := nostr.Filter{Kinds: []nostr.Kind{evt.Kind}, Authors: []nostr.PubKey{evt.PubKey}}
+	if evt.Kind.IsAddressable() {
+		filter.Tags = nostr.TagMap{"d": []string{evt.Tags.GetD()}}
+	}
+
+	shouldStore := true
+	for previous := range b.QueryEvents(filter, 1) {
+		if internal.IsOlder(previous, evt) {
+			if err := b.DeleteEvent(previous.ID); err != nil {
+				return fmt.Errorf("failed to delete event for replacing: %w", err)
+			}
+		} else {
+			shouldStore = false
+		}
+	}
+
+	if shouldStore {
+		if err := b.SaveEvent(evt); err != nil && err != eventstore.ErrDupEvent {
+			return fmt.Errorf("failed to save: %w", err)
+		}
+	}
+
+	return nil
+}

eventstore/bleve/save.go

@@ -0,0 +1,23 @@
+package bleve
+
+import (
+	"fmt"
+	"strconv"
+
+	"fiatjaf.com/nostr"
+)
+
+func (b *BleveBackend) SaveEvent(evt nostr.Event) error {
+	doc := map[string]interface{}{
+		contentField:   evt.Content,
+		kindField:      strconv.Itoa(int(evt.Kind)),
+		pubkeyField:    evt.PubKey.Hex()[56:],
+		createdAtField: float64(evt.CreatedAt),
+	}
+
+	if err := b.index.Index(evt.ID.Hex(), doc); err != nil {
+		return fmt.Errorf("failed to index '%s' document: %w", evt.ID, err)
+	}
+
+	return nil
+}