1package replication23import (4 "context"5 "encoding/json"6 "fmt"7 "net/http"8 "sync"9 "testing"10 "time"1112 "github.com/FelineStateMachine/tinyrelay/internal/event"13 "github.com/FelineStateMachine/tinyrelay/internal/storage"14)1516type negCloseDialer struct {17 mu sync.Mutex18 calls int19 sock *negCloseSocket20}2122type negCloseSocket struct {23 mu sync.Mutex24 responses [][]byte25 requests int26}2728type batchQueryDialer struct {29 mu sync.Mutex30 sockets []*batchQuerySocket31 events map[string]event.Event32}3334type batchQuerySocket struct {35 mu sync.Mutex36 responses [][]byte37 filters []event.Filter38 events map[string]event.Event39}4041func (d *batchQueryDialer) Dial(context.Context, string) (Socket, *http.Response, error) {42 socket := &batchQuerySocket{events: d.events}43 d.mu.Lock()44 d.sockets = append(d.sockets, socket)45 d.mu.Unlock()46 return socket, nil, nil47}4849func (s *batchQuerySocket) Read(ctx context.Context) ([]byte, error) {50 for {51 s.mu.Lock()52 if len(s.responses) > 0 {53 data := s.responses[0]54 s.responses = s.responses[1:]55 s.mu.Unlock()56 return data, nil57 }58 s.mu.Unlock()59 select {60 case <-ctx.Done():61 return nil, ctx.Err()62 case <-time.After(time.Millisecond):63 }64 }65}6667func (s *batchQuerySocket) Write(_ context.Context, data []byte) error {68 var message []json.RawMessage69 if err := json.Unmarshal(data, &message); err != nil || len(message) < 3 {70 return nil71 }72 var kind, id string73 _ = json.Unmarshal(message[0], &kind)74 _ = json.Unmarshal(message[1], &id)75 if kind != "REQ" {76 return nil77 }78 filter, err := event.ParseFilter(message[2])79 if err != nil {80 return err81 }82 s.mu.Lock()83 s.filters = append(s.filters, filter)84 for _, eventID := range filter.IDs {85 if item, ok := s.events[eventID]; ok {86 response, _ := json.Marshal([]any{"EVENT", id, item})87 s.responses = append(s.responses, response)88 }89 }90 response, _ := json.Marshal([]any{"EOSE", id})91 s.responses = append(s.responses, response)92 s.mu.Unlock()93 return nil94}9596func (s *batchQuerySocket) Close(error) error { return nil }9798func (d *negCloseDialer) Dial(context.Context, string) (Socket, *http.Response, error) {99 d.mu.Lock()100 d.calls++101 if d.sock == nil {102 d.sock = &negCloseSocket{}103 }104 sock := d.sock105 d.mu.Unlock()106 return sock, nil, nil107}108109func (s *negCloseSocket) Read(ctx context.Context) ([]byte, error) {110 for {111 s.mu.Lock()112 if len(s.responses) > 0 {113 data := s.responses[0]114 s.responses = s.responses[1:]115 s.mu.Unlock()116 return data, nil117 }118 s.mu.Unlock()119 select {120 case <-ctx.Done():121 return nil, ctx.Err()122 case <-time.After(time.Millisecond):123 }124 }125}126127func (s *negCloseSocket) Write(_ context.Context, data []byte) error {128 var message []json.RawMessage129 if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 {130 return nil131 }132 var kind, id string133 _ = json.Unmarshal(message[0], &kind)134 _ = json.Unmarshal(message[1], &id)135 var response []any136 switch kind {137 case "NEG-OPEN":138 response = []any{"NEG-CLOSE", id}139 case "REQ":140 s.mu.Lock()141 s.requests++142 s.mu.Unlock()143 response = []any{"EOSE", id}144 default:145 return nil146 }147 raw, _ := json.Marshal(response)148 s.mu.Lock()149 s.responses = append(s.responses, raw)150 s.mu.Unlock()151 return nil152}153154func (s *negCloseSocket) Close(error) error { return nil }155156func TestQueryNegentropyEqualInventoryCloseCompletesImmediately(t *testing.T) {157 dialer := &negCloseDialer{}158 transport := &NostrTransport{Dialer: dialer, Timeout: 2 * time.Second}159 started := time.Now()160 events, err := transport.QueryNegentropy(context.Background(), "ws://relay.example", event.Filter{}, nil)161 if err != nil {162 t.Fatal(err)163 }164 if len(events) != 0 {165 t.Fatalf("equal inventory returned %d events", len(events))166 }167 if elapsed := time.Since(started); elapsed >= time.Second {168 t.Fatalf("equal inventory close took %s; likely waited for timeout", elapsed)169 }170 dialer.mu.Lock()171 calls := dialer.calls172 sock := dialer.sock173 dialer.mu.Unlock()174 if calls != 1 {175 t.Fatalf("transport dials = %d, want negentropy only", calls)176 }177 sock.mu.Lock()178 requests := sock.requests179 sock.mu.Unlock()180 if requests != 0 {181 t.Fatalf("equal inventory issued %d unrestricted REQ requests", requests)182 }183}184185type negRejectDialer struct{ calls int }186187type negRejectSocket struct{ responses [][]byte }188189func (d *negRejectDialer) Dial(context.Context, string) (Socket, *http.Response, error) {190 d.calls++191 return &negRejectSocket{}, nil, nil192}193194func (s *negRejectSocket) Read(ctx context.Context) ([]byte, error) {195 if len(s.responses) == 0 {196 return nil, ctx.Err()197 }198 item := s.responses[0]199 s.responses = s.responses[1:]200 return item, nil201}202203func (s *negRejectSocket) Write(_ context.Context, data []byte) error {204 var message []json.RawMessage205 if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 {206 return nil207 }208 var kind, id string209 _ = json.Unmarshal(message[0], &kind)210 _ = json.Unmarshal(message[1], &id)211 var response []any212 if kind == "NEG-OPEN" {213 response = []any{"NEG-ERR", id, "unsupported"}214 } else if kind == "REQ" {215 response = []any{"EOSE", id}216 }217 raw, _ := json.Marshal(response)218 s.responses = append(s.responses, raw)219 return nil220}221222func (s *negRejectSocket) Close(error) error { return nil }223224func TestQuerySynchronizedFallsBackWhenNegentropyIsUnsupported(t *testing.T) {225 dialer := &negRejectDialer{}226 transport := &NostrTransport{Dialer: dialer, Timeout: time.Second}227 items, err := transport.QuerySynchronized(context.Background(), "ws://relay.example", event.Filter{}, nil)228 if err != nil {229 t.Fatal(err)230 }231 if len(items) != 0 || dialer.calls != 2 {232 t.Fatalf("fallback result=%d dials=%d, want empty result and two dials", len(items), dialer.calls)233 }234}235236func TestNegentropyNeededIDsAreFetchedInBoundedFilteredBatches(t *testing.T) {237 const total = negentropyFetchBatch + 1238 secret := "1111111111111111111111111111111111111111111111111111111111111111"239 dialer := &batchQueryDialer{events: make(map[string]event.Event, total)}240 ids := make([]string, 0, total)241 for i := 0; i < total; i++ {242 item := event.Event{Kind: 1, CreatedAt: int64(i + 1), Content: fmt.Sprintf("item-%d", i), Tags: [][]string{{"p", "recipient"}}}243 if err := event.Sign(&item, secret); err != nil {244 t.Fatal(err)245 }246 dialer.events[item.ID] = item247 ids = append(ids, item.ID)248 }249 filter := event.Filter{Kinds: []int{1}, Tags: map[string][]string{"p": {"recipient"}}}250 transport := &NostrTransport{Dialer: dialer, Timeout: time.Second}251 got, err := transport.queryNegentropyIDs(context.Background(), "ws://relay.example", filter, ids)252 if err != nil {253 t.Fatal(err)254 }255 if len(got) != total {256 t.Fatalf("fetched %d events, want %d", len(got), total)257 }258 dialer.mu.Lock()259 sockets := append([]*batchQuerySocket(nil), dialer.sockets...)260 dialer.mu.Unlock()261 if len(sockets) != 2 {262 t.Fatalf("query opened %d sockets, want two batches", len(sockets))263 }264 for i, socket := range sockets {265 socket.mu.Lock()266 if len(socket.filters) != 1 {267 socket.mu.Unlock()268 t.Fatalf("batch %d recorded %d filters", i, len(socket.filters))269 }270 batch := socket.filters[0]271 socket.mu.Unlock()272 if len(batch.IDs) == 0 || len(batch.IDs) > negentropyFetchBatch || batch.Limit == nil || *batch.Limit != len(batch.IDs) || len(batch.Kinds) != 1 || batch.Kinds[0] != 1 || len(batch.Tags["p"]) != 1 || batch.Tags["p"][0] != "recipient" {273 t.Fatalf("batch %d did not preserve bounded membership: %#v", i, batch)274 }275 }276}277278type tiePushTransport struct{ sent []string }279280func (p *tiePushTransport) Send(context.Context, string, event.Event) (DeliveryResult, error) {281 p.sent = append(p.sent, "sent")282 return DeliveryResult{Accepted: true}, nil283}284285func TestPushCursorUsesSequenceWhenCreatedAtTies(t *testing.T) {286 ctx := context.Background()287 store, err := storage.Open(ctx, t.TempDir()+"/relay.db")288 if err != nil {289 t.Fatal(err)290 }291 defer store.Close()292 secret := "1111111111111111111111111111111111111111111111111111111111111111"293 for i := 0; i < 2; i++ {294 e := event.Event{Kind: 1, CreatedAt: 100, Content: string(rune('a' + i))}295 if err := event.Sign(&e, secret); err != nil {296 t.Fatal(err)297 }298 if _, err := store.Save(ctx, e, storage.SaveOptions{Now: 100}); err != nil {299 t.Fatal(err)300 }301 }302 transport := &tiePushTransport{}303 stats, cursor, err := PushAfter(ctx, store, transport, "ws://relay.example", 0, event.Filter{}, 100)304 if err != nil {305 t.Fatal(err)306 }307 if stats.Sent != 2 || len(transport.sent) != 2 || cursor < 2 {308 t.Fatalf("same-timestamp push = stats=%+v sent=%d cursor=%d", stats, len(transport.sent), cursor)309 }310 stats, next, err := PushAfter(ctx, store, transport, "ws://relay.example", cursor, event.Filter{}, 100)311 if err != nil {312 t.Fatal(err)313 }314 if stats.Sent != 0 || next != cursor || len(transport.sent) != 2 {315 t.Fatalf("cursor replay = stats=%+v next=%d sent=%d", stats, next, len(transport.sent))316 }317}