1package daemon23import (4 "context"5 "database/sql"6 "encoding/json"7 "errors"8 "fmt"9 "net/http"10 "sort"11 "testing"12 "time"1314 "github.com/FelineStateMachine/tinyrelay/internal/event"15 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"16 "github.com/FelineStateMachine/tinyrelay/internal/replication"17 "github.com/FelineStateMachine/tinyrelay/internal/storage"18)1920// cursorRelay is deliberately a small NIP-01 relay. It applies the filter's21// limit and until values, which makes the test exercise the same bounded head22// and paginated older queries used by production sync.23type cursorRelay struct {24 events []event.Event25 stallHistory bool26}2728type cursorSocket struct {29 relay *cursorRelay30 responses [][]byte31}3233type cursorDialer struct{ relay *cursorRelay }3435func (d cursorDialer) Dial(context.Context, string) (replication.Socket, *http.Response, error) {36 return &cursorSocket{relay: d.relay}, nil, nil37}3839func (s *cursorSocket) Read(ctx context.Context) ([]byte, error) {40 if len(s.responses) == 0 {41 <-ctx.Done()42 return nil, ctx.Err()43 }44 item := s.responses[0]45 s.responses = s.responses[1:]46 return item, nil47}4849func (s *cursorSocket) Write(_ context.Context, data []byte) error {50 var raw []json.RawMessage51 if err := json.Unmarshal(data, &raw); err != nil || len(raw) < 3 {52 return nil53 }54 var kind, sub string55 if json.Unmarshal(raw[0], &kind) != nil || json.Unmarshal(raw[1], &sub) != nil || kind != "REQ" {56 return nil57 }58 f, err := event.ParseFilter(raw[2])59 if err != nil {60 return err61 }62 if f.Until != nil && s.relay.stallHistory {63 return nil64 }65 items := append([]event.Event(nil), s.relay.events...)66 sort.Slice(items, func(i, j int) bool {67 if items[i].CreatedAt != items[j].CreatedAt {68 return items[i].CreatedAt > items[j].CreatedAt69 }70 return items[i].ID < items[j].ID71 })72 limit := 073 if f.Limit != nil {74 limit = *f.Limit75 }76 for _, item := range items {77 if !event.Matches(f, item) {78 continue79 }80 if limit > 0 && len(s.responses) >= limit {81 break82 }83 message, _ := json.Marshal([]any{"EVENT", sub, item})84 s.responses = append(s.responses, message)85 }86 eose, _ := json.Marshal([]any{"EOSE", sub})87 s.responses = append(s.responses, eose)88 return nil89}9091func (s *cursorSocket) Close(error) error { return nil }9293func TestGitPullFilterAdvancesBeyondBoundedLocalInventory(t *testing.T) {94 _, tenant := testTenant(t)95 owner := tenant.Policy().Owner96 secret := "0000000000000000000000000000000000000000000000000000000000000001"97 now := time.Now().Unix()98 filter := event.Filter{Authors: []string{owner}, Kinds: []int{1}}99 for i := 0; i < 10001; i++ {100 // Seed the stored inventory directly; remote responses below still101 // undergo real signature validation and the normal ingestion path.102 item := event.Event{ID: fmt.Sprintf("%064x", i+1), PubKey: owner, Kind: 1, CreatedAt: now - int64(i+10000), Tags: [][]string{}, Content: "local"}103 if _, err := tenant.store.Save(context.Background(), item, storage.SaveOptions{Now: now, SearchMode: tenant.Policy().Features.Search}); err != nil {104 t.Fatal(err)105 }106 }107 remote := &cursorRelay{}108 for i := 0; i < gitHeadPageSize+1; i++ {109 item := event.Event{Kind: 1, CreatedAt: now - int64(i+1), Tags: [][]string{}, Content: "head"}110 if err := event.Sign(&item, secret); err != nil {111 t.Fatal(err)112 }113 remote.events = append(remote.events, item)114 }115 older := event.Event{Kind: 1, CreatedAt: now - 1000, Tags: [][]string{}, Content: "older"}116 if err := event.Sign(&older, secret); err != nil {117 t.Fatal(err)118 }119 remote.events = append(remote.events, older)120 transport := &replication.NostrTransport{Dialer: cursorDialer{relay: remote}, Timeout: time.Second}121122 if err := tenant.gitPullFilter(context.Background(), transport, "ws://cursor.test", filter); !errors.Is(err, replication.ErrPullIncomplete) {123 t.Fatalf("first pull error = %v, want incomplete", err)124 }125 var cursor storage.EventCursor126 if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); err != nil || cursor.CreatedAt == 0 {127 t.Fatalf("head cursor was not persisted: %v %#v", err, cursor)128 }129 if got := countContent(t, tenant, "older"); got != 0 {130 t.Fatalf("older event arrived during head seed: %d", got)131 }132 if err := tenant.gitPullFilter(context.Background(), transport, "ws://cursor.test", filter); !errors.Is(err, replication.ErrPullIncomplete) {133 t.Fatalf("second pull error = %v, want incomplete local inventory", err)134 }135 if got := countContent(t, tenant, "older"); got != 1 {136 t.Fatalf("older event count = %d, want 1", got)137 }138139 // A completed sweep removes its cursor row so per-filter state cannot140 // accumulate; the next pass seeds a new head window.141 cursor = storage.EventCursor{}142 if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &cursor); !errors.Is(err, sql.ErrNoRows) {143 t.Fatalf("completed sweep left cursor %#v, err %v", cursor, err)144 }145 before := cursor146 ctx, cancel := context.WithCancel(context.Background())147 cancel()148 if err := tenant.gitPullFilter(ctx, transport, "ws://cursor.test", filter); err == nil {149 t.Fatal("canceled pull returned nil")150 }151 var after storage.EventCursor152 if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil && !errors.Is(err, sql.ErrNoRows) {153 t.Fatal(err)154 }155 if after != before {156 t.Fatalf("canceled pull advanced cursor from %#v to %#v", before, after)157 }158 // A stalled historical window must not consume the head or move its159 // checkpoint, even when the outer pass has a short deadline.160 before = storage.EventCursor{CreatedAt: older.CreatedAt, ID: older.ID}161 if err := tenant.store.PutSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), before); err != nil {162 t.Fatal(err)163 }164 newest := event.Event{Kind: 1, CreatedAt: now, Tags: [][]string{}, Content: "live-head"}165 if err := event.Sign(&newest, secret); err != nil {166 t.Fatal(err)167 }168 remote.events = []event.Event{newest}169 remote.stallHistory = true170 transport.Timeout = time.Minute171 ctx, cancel = context.WithTimeout(context.Background(), 3*time.Second)172 defer cancel()173 if err := tenant.gitPullFilter(ctx, transport, "ws://cursor.test", filter); !errors.Is(err, context.DeadlineExceeded) {174 t.Fatalf("stalled history error = %v", err)175 }176 if got := countContent(t, tenant, "live-head"); got != 1 {177 t.Fatalf("stalled history blocked newest event: %d", got)178 }179 if err := tenant.store.GetSetting(context.Background(), gitHistoryCursorKey("ws://cursor.test", filter), &after); err != nil {180 t.Fatal(err)181 }182 if after != before {183 t.Fatalf("stalled history moved cursor from %#v to %#v", before, after)184 }185}186187func countContent(t *testing.T, tenant *Tenant, content string) int {188 t.Helper()189 rows, err := tenant.store.Query(context.Background(), event.Filter{Kinds: []int{1}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})190 if err != nil {191 t.Fatal(err)192 }193 count := 0194 for _, item := range rows.Events {195 if item.Content == content {196 count++197 }198 }199 return count200}201202func TestGitSyncOutcomeClassifiesIncompleteAndBudgetDeadlines(t *testing.T) {203 ctx := context.Background()204 if got := gitSyncOutcome(ctx, errors.Join(context.DeadlineExceeded, replication.ErrPullIncomplete)); !errors.Is(got, gitrelay.ErrIncomplete) || errors.Is(got, replication.ErrPullIncomplete) {205 t.Fatalf("incomplete pass should map to gitrelay.ErrIncomplete, got %v", got)206 }207 if got := gitSyncOutcome(ctx, errors.Join(context.DeadlineExceeded, errors.New("dial failed"))); got == nil || errors.Is(got, gitrelay.ErrIncomplete) || errors.Is(got, context.DeadlineExceeded) {208 t.Fatalf("real errors must survive with deadlines dropped, got %v", got)209 }210 if got := gitSyncOutcome(ctx, context.DeadlineExceeded); got != nil {211 t.Fatalf("a bare budget deadline is not a failure, got %v", got)212 }213 canceled, cancel := context.WithCancel(ctx)214 cancel()215 if got := gitSyncOutcome(canceled, context.DeadlineExceeded); got == nil {216 t.Fatal("a canceled caller context must propagate")217 }218 if !gitIngestRejected(errors.New("blocked: this pubkey is banned")) || gitIngestRejected(errors.New("check pubkey ban: disk I/O error")) {219 t.Fatal("rejection classification must follow NIP-01 prefixes")220 }221}