| 1 | package replication |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "net/http" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 13 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 14 | ) |
| 15 | |
| 16 | type negCloseDialer struct { |
| 17 | mu sync.Mutex |
| 18 | calls int |
| 19 | sock *negCloseSocket |
| 20 | } |
| 21 | |
| 22 | type negCloseSocket struct { |
| 23 | mu sync.Mutex |
| 24 | responses [][]byte |
| 25 | requests int |
| 26 | } |
| 27 | |
| 28 | type batchQueryDialer struct { |
| 29 | mu sync.Mutex |
| 30 | sockets []*batchQuerySocket |
| 31 | events map[string]event.Event |
| 32 | } |
| 33 | |
| 34 | type batchQuerySocket struct { |
| 35 | mu sync.Mutex |
| 36 | responses [][]byte |
| 37 | filters []event.Filter |
| 38 | events map[string]event.Event |
| 39 | } |
| 40 | |
| 41 | func (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, nil |
| 47 | } |
| 48 | |
| 49 | func (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, nil |
| 57 | } |
| 58 | s.mu.Unlock() |
| 59 | select { |
| 60 | case <-ctx.Done(): |
| 61 | return nil, ctx.Err() |
| 62 | case <-time.After(time.Millisecond): |
| 63 | } |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | func (s *batchQuerySocket) Write(_ context.Context, data []byte) error { |
| 68 | var message []json.RawMessage |
| 69 | if err := json.Unmarshal(data, &message); err != nil || len(message) < 3 { |
| 70 | return nil |
| 71 | } |
| 72 | var kind, id string |
| 73 | _ = json.Unmarshal(message[0], &kind) |
| 74 | _ = json.Unmarshal(message[1], &id) |
| 75 | if kind != "REQ" { |
| 76 | return nil |
| 77 | } |
| 78 | filter, err := event.ParseFilter(message[2]) |
| 79 | if err != nil { |
| 80 | return err |
| 81 | } |
| 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 nil |
| 94 | } |
| 95 | |
| 96 | func (s *batchQuerySocket) Close(error) error { return nil } |
| 97 | |
| 98 | func (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.sock |
| 105 | d.mu.Unlock() |
| 106 | return sock, nil, nil |
| 107 | } |
| 108 | |
| 109 | func (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, nil |
| 117 | } |
| 118 | s.mu.Unlock() |
| 119 | select { |
| 120 | case <-ctx.Done(): |
| 121 | return nil, ctx.Err() |
| 122 | case <-time.After(time.Millisecond): |
| 123 | } |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func (s *negCloseSocket) Write(_ context.Context, data []byte) error { |
| 128 | var message []json.RawMessage |
| 129 | if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 { |
| 130 | return nil |
| 131 | } |
| 132 | var kind, id string |
| 133 | _ = json.Unmarshal(message[0], &kind) |
| 134 | _ = json.Unmarshal(message[1], &id) |
| 135 | var response []any |
| 136 | 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 nil |
| 146 | } |
| 147 | raw, _ := json.Marshal(response) |
| 148 | s.mu.Lock() |
| 149 | s.responses = append(s.responses, raw) |
| 150 | s.mu.Unlock() |
| 151 | return nil |
| 152 | } |
| 153 | |
| 154 | func (s *negCloseSocket) Close(error) error { return nil } |
| 155 | |
| 156 | func 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.calls |
| 172 | sock := dialer.sock |
| 173 | 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.requests |
| 179 | sock.mu.Unlock() |
| 180 | if requests != 0 { |
| 181 | t.Fatalf("equal inventory issued %d unrestricted REQ requests" , requests) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | type negRejectDialer struct{ calls int } |
| 186 | |
| 187 | type negRejectSocket struct{ responses [][]byte } |
| 188 | |
| 189 | func (d *negRejectDialer) Dial(context.Context, string) (Socket, *http.Response, error) { |
| 190 | d.calls++ |
| 191 | return &negRejectSocket{}, nil, nil |
| 192 | } |
| 193 | |
| 194 | func (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, nil |
| 201 | } |
| 202 | |
| 203 | func (s *negRejectSocket) Write(_ context.Context, data []byte) error { |
| 204 | var message []json.RawMessage |
| 205 | if err := json.Unmarshal(data, &message); err != nil || len(message) < 2 { |
| 206 | return nil |
| 207 | } |
| 208 | var kind, id string |
| 209 | _ = json.Unmarshal(message[0], &kind) |
| 210 | _ = json.Unmarshal(message[1], &id) |
| 211 | var response []any |
| 212 | 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 nil |
| 220 | } |
| 221 | |
| 222 | func (s *negRejectSocket) Close(error) error { return nil } |
| 223 | |
| 224 | func 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 | } |
| 235 | |
| 236 | func TestNegentropyNeededIDsAreFetchedInBoundedFilteredBatches(t *testing.T) { |
| 237 | const total = negentropyFetchBatch + 1 |
| 238 | 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] = item |
| 247 | 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 | } |
| 277 | |
| 278 | type tiePushTransport struct{ sent []string } |
| 279 | |
| 280 | func (p *tiePushTransport) Send(context.Context, string, event.Event) (DeliveryResult, error) { |
| 281 | p.sent = append(p.sent, "sent" ) |
| 282 | return DeliveryResult{Accepted: true}, nil |
| 283 | } |
| 284 | |
| 285 | func 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 | } |