wait, Querier can just return an iter like eventstore, it's ok.

This commit is contained in:
fiatjaf
2025-04-16 03:06:27 -03:00
parent 5b8954461f
commit f7884cea4f
4 changed files with 17 additions and 49 deletions

View File

@@ -1,26 +0,0 @@
package wrappers
import (
"context"
"fiatjaf.com/nostr"
"fiatjaf.com/nostr/eventstore"
)
var _ nostr.Querier = StoreQuerier{}
type StoreQuerier struct {
eventstore.Store
}
func (w StoreQuerier) QueryEvents(ctx context.Context, filter nostr.Filter) (chan nostr.Event, error) {
ch := make(chan nostr.Event)
go func() {
for evt := range w.Store.QueryEvents(filter) {
ch <- evt
}
}()
return ch, nil
}

View File

@@ -2,6 +2,7 @@ package nostr
import ( import (
"context" "context"
"iter"
) )
type Publisher interface { type Publisher interface {
@@ -9,7 +10,7 @@ type Publisher interface {
} }
type Querier interface { type Querier interface {
QueryEvents(context.Context, Filter) (chan Event, error) QueryEvents(Filter) iter.Seq[Event]
} }
type QuerierPublisher interface { type QuerierPublisher interface {

View File

@@ -36,18 +36,15 @@ func NegentropySync(
vec := vector.New() vec := vector.New()
neg := negentropy.New(vec, 1024*1024) neg := negentropy.New(vec, 1024*1024)
ch, err := store.QueryEvents(ctx, filter)
if err != nil {
return err
}
for evt := range ch { for evt := range store.QueryEvents(filter) {
vec.Insert(evt.CreatedAt, evt.ID) vec.Insert(evt.CreatedAt, evt.ID)
} }
vec.Seal() vec.Seal()
result := make(chan error) result := make(chan error)
var err error
var r *nostr.Relay var r *nostr.Relay
r, err = nostr.RelayConnect(ctx, url, nostr.RelayOptions{ r, err = nostr.RelayConnect(ctx, url, nostr.RelayOptions{
CustomHandler: func(data string) { CustomHandler: func(data string) {
@@ -118,12 +115,7 @@ func NegentropySync(
if len(ids) == 0 { if len(ids) == 0 {
return return
} }
evtch, err := dir.source.QueryEvents(ctx, nostr.Filter{IDs: ids}) for evt := range dir.source.QueryEvents(nostr.Filter{IDs: ids}) {
if err != nil {
result <- fmt.Errorf("error querying source on %s: %w", dir.label, err)
return
}
for evt := range evtch {
dir.target.Publish(ctx, evt) dir.target.Publish(ctx, evt)
} }
} }

View File

@@ -6,6 +6,7 @@ import (
"crypto/tls" "crypto/tls"
"errors" "errors"
"fmt" "fmt"
"iter"
"log" "log"
"net/http" "net/http"
"strconv" "strconv"
@@ -434,19 +435,21 @@ func (r *Relay) PrepareSubscription(ctx context.Context, filter Filter, opts Sub
} }
// implement Querier interface // implement Querier interface
func (r *Relay) QueryEvents(ctx context.Context, filter Filter) (chan Event, error) { func (r *Relay) QueryEvents(filter Filter) iter.Seq[Event] {
sub, err := r.Subscribe(ctx, filter, SubscriptionOptions{Label: "queryevents"}) ctx, cancel := context.WithCancel(r.connectionContext)
if err != nil {
return nil, err
}
ch := make(chan Event) return func(yield func(Event) bool) {
defer cancel()
sub, err := r.Subscribe(ctx, filter, SubscriptionOptions{Label: "queryevents"})
if err != nil {
return
}
go func() {
for { for {
select { select {
case evt := <-sub.Events: case evt := <-sub.Events:
ch <- evt yield(evt)
case <-sub.EndOfStoredEvents: case <-sub.EndOfStoredEvents:
return return
case <-sub.ClosedReason: case <-sub.ClosedReason:
@@ -455,9 +458,7 @@ func (r *Relay) QueryEvents(ctx context.Context, filter Filter) (chan Event, err
return return
} }
} }
}() }
return ch, nil
} }
// Count sends a "COUNT" command to the relay and returns the count of events matching the filters. // Count sends a "COUNT" command to the relay and returns the count of events matching the filters.