1package daemon23import (4 "context"5 "encoding/json"6 "net/http"7 "net/http/httptest"8 "strings"9 "sync"10 "testing"11 "time"1213 "github.com/FelineStateMachine/tinyrelay/internal/event"14 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"15 "github.com/FelineStateMachine/tinyrelay/internal/relay"16 "github.com/FelineStateMachine/tinyrelay/internal/storage"17 "github.com/coder/websocket"18)1920type gitSyncEventRelay struct {21 mu sync.Mutex22 events []event.Event23 live *event.Event24 liveSent chan struct{}25 requests int26}2728func (r *gitSyncEventRelay) ServeHTTP(w http.ResponseWriter, req *http.Request) {29 if req.Header.Get("Accept") == "application/nostr+json" {30 w.Header().Set("Content-Type", "application/nostr+json")31 _, _ = w.Write([]byte(`{"supported_grasps":["GRASP-01"],"private_service":false}`))32 return33 }34 conn, err := websocket.Accept(w, req, nil)35 if err != nil {36 return37 }38 defer conn.Close(websocket.StatusNormalClosure, "done")39 _, data, err := conn.Read(req.Context())40 if err != nil {41 return42 }43 var message []json.RawMessage44 if json.Unmarshal(data, &message) != nil || len(message) < 3 {45 return46 }47 var kind, subscription string48 _ = json.Unmarshal(message[0], &kind)49 _ = json.Unmarshal(message[1], &subscription)50 if kind != "REQ" {51 if kind == "NEG-OPEN" {52 response, _ := json.Marshal([]any{"NEG-ERR", subscription, "unsupported"})53 _ = conn.Write(req.Context(), websocket.MessageText, response)54 }55 return56 }57 filter, err := event.ParseFilter(message[2])58 if err != nil {59 return60 }61 r.mu.Lock()62 r.requests++63 items := append([]event.Event(nil), r.events...)64 r.mu.Unlock()65 for _, item := range items {66 if event.Matches(filter, item) {67 response, _ := json.Marshal([]any{"EVENT", subscription, item})68 _ = conn.Write(req.Context(), websocket.MessageText, response)69 }70 }71 response, _ := json.Marshal([]any{"EOSE", subscription})72 _ = conn.Write(req.Context(), websocket.MessageText, response)73 if strings.HasPrefix(subscription, "tiny-live-") && r.live != nil {74 live, _ := json.Marshal([]any{"EVENT", subscription, *r.live})75 _ = conn.Write(req.Context(), websocket.MessageText, live)76 if r.liveSent != nil {77 select {78 case r.liveSent <- struct{}{}:79 default:80 }81 }82 }83}8485func (r *gitSyncEventRelay) requestCount() int {86 r.mu.Lock()87 defer r.mu.Unlock()88 return r.requests89}9091func TestGitSyncPeerImportsHistoricalIssueAndRootReplies(t *testing.T) {92 _, tenant := testTenant(t)93 tenant.app.cfg.AllowPrivateRelays = true94 p := tenant.Policy()95 p.Features.Grasp = true96 p.Features.Grasp02 = true97 p.Features.Grasp05 = true98 if err := tenant.applyPolicy(context.Background(), p); err != nil {99 t.Fatal(err)100 }101 owner := tenant.Policy().Owner102 secret := strings.Repeat("9", 63) + "8"103 author, err := event.PublicKey(secret)104 if err != nil {105 t.Fatal(err)106 }107 relay := &gitSyncEventRelay{}108 server := httptest.NewServer(relay)109 t.Cleanup(server.Close)110 target := "ws" + strings.TrimPrefix(server.URL, "http")111 coordinate := "30617:" + owner + ":remote"112 root := signedCollaborationEvent(t, secret, event.KIND_GIT_ISSUE, 100,113 [][]string{{"a", coordinate}, {"subject", "Remote issue"}}, "historical")114 reply := signedCollaborationEvent(t, secret, 1111, 101,115 [][]string{{"E", root.ID}, {"K", "1621"}, {"P", author}}, "remote reply")116 relay.events = []event.Event{root, reply}117 repo := gitrelay.Repository{Owner: owner, Identifier: "remote", Relays: []string{target}}118 if err := tenant.gitSyncPeer(context.Background(), repo, target); err != nil {119 t.Fatal(err)120 }121 rows, err := tenant.store.Query(context.Background(), event.Filter{IDs: []string{root.ID, reply.ID}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})122 if err != nil {123 t.Fatal(err)124 }125 if len(rows.Events) != 2 {126 t.Fatalf("synced collaboration events = %d, want 2", len(rows.Events))127 }128 if relay.requestCount() < 4 {129 t.Fatalf("multi-filter sync issued %d requests, want root and reply filters", relay.requestCount())130 }131}132133func TestGitSyncPeerFeatureOffDoesNotDial(t *testing.T) {134 _, tenant := testTenant(t)135 relay := &gitSyncEventRelay{}136 server := httptest.NewServer(relay)137 t.Cleanup(server.Close)138 target := "ws" + strings.TrimPrefix(server.URL, "http")139 repo := gitrelay.Repository{Owner: tenant.Policy().Owner, Identifier: "remote", Relays: []string{target}}140 if err := tenant.gitSyncPeer(context.Background(), repo, target); err == nil {141 t.Fatal("feature-off sync unexpectedly succeeded")142 }143 if got := relay.requestCount(); got != 0 {144 t.Fatalf("feature-off sync dialed relay with %d requests", got)145 }146}147148func TestGitLivePeerImportsEventAfterEOSE(t *testing.T) {149 _, tenant := testTenant(t)150 tenant.app.cfg.AllowPrivateRelays = true151 p := tenant.Policy()152 p.Features.Grasp = true153 p.Features.Grasp02 = true154 p.Features.Grasp05 = true155 if err := tenant.applyPolicy(context.Background(), p); err != nil {156 t.Fatal(err)157 }158 owner := tenant.Policy().Owner159 secret := strings.Repeat("a", 63) + "b"160 author, err := event.PublicKey(secret)161 if err != nil {162 t.Fatal(err)163 }164 remote := &gitSyncEventRelay{liveSent: make(chan struct{}, 1)}165 server := httptest.NewServer(remote)166 t.Cleanup(server.Close)167 target := "ws" + strings.TrimPrefix(server.URL, "http")168 coordinate := "30617:" + owner + ":live"169 announcement := event.Event{Kind: event.KIND_REPO, PubKey: owner, CreatedAt: 199,170 Tags: [][]string{{"d", "live"}, {"clone", "http://relay.test/" + owner + "/live.git"}, {"relays", "ws://relay.test"}}}171 if err := event.Sign(&announcement, strings.Repeat("0", 63)+"1"); err != nil {172 t.Fatal(err)173 }174 if _, err := tenant.router.Publish(context.Background(), announcement, relay.Session{PubKeys: []string{owner}}); err != nil {175 t.Fatal(err)176 }177 root := signedCollaborationEvent(t, secret, event.KIND_GIT_ISSUE, 200,178 [][]string{{"a", coordinate}, {"subject", "Live issue"}}, "root")179 liveReply := signedCollaborationEvent(t, secret, 1111, 201,180 [][]string{{"E", root.ID}, {"K", "1621"}, {"P", author}}, "after EOSE")181 remote.events = []event.Event{root}182 remote.live = &liveReply183 repo := gitrelay.Repository{Owner: owner, Identifier: "live", Relays: []string{target}}184 ctx, cancel := context.WithCancel(context.Background())185 done := make(chan struct{})186 go func() {187 tenant.gitLivePeer(ctx, repo, target)188 close(done)189 }()190 select {191 case <-remote.liveSent:192 case <-time.After(5 * time.Second):193 cancel()194 <-done195 t.Fatal("live subscription did not receive post-EOSE event")196 }197 deadline := time.Now().Add(2 * time.Second)198 for time.Now().Before(deadline) {199 rows, queryErr := tenant.store.Query(context.Background(), event.Filter{IDs: []string{liveReply.ID}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})200 if queryErr == nil && len(rows.Events) == 1 {201 break202 }203 time.Sleep(10 * time.Millisecond)204 }205 cancel()206 select {207 case <-done:208 case <-time.After(5 * time.Second):209 t.Fatal("live worker did not stop")210 }211 rows, err := tenant.store.Query(context.Background(), event.Filter{IDs: []string{liveReply.ID}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}})212 if err != nil {213 t.Fatal(err)214 }215 if len(rows.Events) != 1 {216 t.Fatalf("post-EOSE event was not imported: %d", len(rows.Events))217 }218}219220func TestGitSyncPeerPrivatePeerReadinessRefusesBeforeWebsocket(t *testing.T) {221 _, tenant := testTenant(t)222 tenant.app.cfg.AllowPrivateRelays = true223 var websocketRequests int224 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {225 if req.Header.Get("Accept") == "application/nostr+json" {226 w.Header().Set("Content-Type", "application/nostr+json")227 _, _ = w.Write([]byte(`{"supported_grasps":["GRASP-01"],"private_service":false}`))228 return229 }230 websocketRequests++231 }))232 t.Cleanup(server.Close)233 target := "ws" + strings.TrimPrefix(server.URL, "http")234 p := tenant.Policy()235 p.Features.Grasp = true236 p.Features.Grasp02 = true237 p.Features.Grasp08 = true238 p.Reads = "members"239 p.PrivatePeers = []string{target}240 if err := tenant.applyPolicy(context.Background(), p); err != nil {241 t.Fatal(err)242 }243 repo := gitrelay.Repository{Owner: p.Owner, Identifier: "private", Private: true, Relays: []string{target}}244 if err := tenant.gitSyncPeer(context.Background(), repo, target); err == nil {245 t.Fatal("unready private peer unexpectedly accepted")246 }247 if websocketRequests != 0 {248 t.Fatalf("private readiness failure opened %d websocket requests", websocketRequests)249 }250}