1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "net"9 "net/http"10 "net/url"11 "strings"12 "sync"13 "sync/atomic"14 "time"1516 "github.com/FelineStateMachine/tinyrelay/internal/event"17 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"18 "github.com/coder/websocket"19)2021// Socket is the message-oriented connection used by NostrTransport. The22// caller that receives a Socket owns it and closes it when the exchange ends.23type Socket interface {24 Read(ctx context.Context) ([]byte, error)25 Write(ctx context.Context, message []byte) error26 Close(cause error) error27}2829// Dialer opens a Socket for one relay target and returns its HTTP handshake.30type Dialer interface {31 Dial(ctx context.Context, target string) (Socket, *http.Response, error)32}3334// IPResolver resolves relay hostnames for address-policy checks and dialing.35type IPResolver interface {36 LookupIPAddr(ctx context.Context, host string) ([]net.IPAddr, error)37}3839// WebsocketDialer validates websocket targets, blocks private addresses unless40// AllowPrivate is set and optionally answers NIP-42 authentication challenges.41// HTTPClient and Resolver are caller-supplied dependencies and remain owned by42// the caller.43type WebsocketDialer struct {44 AllowPrivate bool45 MaxMessageBytes int6446 Resolver IPResolver47 HTTPClient *http.Client48 // Authenticate answers a NIP-42 challenge when configured for an49 // operator-trusted private peer. Public relays remain unauthenticated.50 Authenticate func(context.Context, string, string) (event.Event, error)51}5253func (d WebsocketDialer) Dial(ctx context.Context, target string) (Socket, *http.Response, error) {54 parsed, err := url.Parse(target)55 if err != nil || (parsed.Scheme != "ws" && parsed.Scheme != "wss") || parsed.Host == "" || parsed.User != nil || parsed.Fragment != "" {56 return nil, nil, errors.New("replication: invalid websocket URL")57 }58 if d.MaxMessageBytes < 0 {59 d.MaxMessageBytes = 8 << 2060 }61 resolver := d.Resolver62 if resolver == nil {63 resolver = net.DefaultResolver64 }65 if err := validateHost(ctx, parsed.Hostname(), resolver, d.AllowPrivate); err != nil {66 return nil, nil, err67 }68 client := d.client(resolver, d.AllowPrivate)69 conn, response, err := websocket.Dial(ctx, target, &websocket.DialOptions{HTTPClient: client})70 if err != nil {71 return nil, response, err72 }73 limit := d.MaxMessageBytes74 if limit == 0 {75 limit = -176 }77 conn.SetReadLimit(limit)78 if d.Authenticate != nil {79 if err := authenticateSocket(ctx, conn, target, d.Authenticate); err != nil {80 _ = conn.Close(websocket.StatusPolicyViolation, "authentication failed")81 return nil, response, err82 }83 }84 return coderSocket{conn: conn}, response, nil85}8687func authenticateSocket(ctx context.Context, conn *websocket.Conn, target string, sign func(context.Context, string, string) (event.Event, error)) error {88 readCtx, cancel := context.WithTimeout(ctx, 10*time.Second)89 defer cancel()90 _, data, err := conn.Read(readCtx)91 if err != nil {92 return fmt.Errorf("replication: NIP-42 challenge: %w", err)93 }94 var raw []json.RawMessage95 if err := json.Unmarshal(data, &raw); err != nil || len(raw) < 2 {96 return errors.New("replication: invalid NIP-42 challenge")97 }98 var kind, challenge string99 if err := json.Unmarshal(raw[0], &kind); err != nil || kind != "AUTH" || json.Unmarshal(raw[1], &challenge) != nil || challenge == "" {100 return errors.New("replication: peer did not issue NIP-42 challenge")101 }102 proof, err := sign(ctx, target, challenge)103 if err != nil {104 return err105 }106 message, err := json.Marshal([]any{"AUTH", proof})107 if err != nil {108 return err109 }110 return conn.Write(ctx, websocket.MessageText, message)111}112113func (d WebsocketDialer) client(resolver IPResolver, allowPrivate bool) *http.Client {114 base := d.HTTPClient115 if base == nil {116 base = &http.Client{}117 }118 if base.Transport != nil {119 if _, standard := base.Transport.(*http.Transport); !standard {120 // Nonstandard transports are an explicit test/integration injection;121 // the caller owns their proxy and address policy.122 return base123 }124 }125 transport := http.DefaultTransport.(*http.Transport).Clone()126 if base.Transport != nil {127 if source, ok := base.Transport.(*http.Transport); ok {128 transport = source.Clone()129 }130 }131 transport.Proxy = nil132 transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) {133 host, port, err := net.SplitHostPort(address)134 if err != nil {135 return nil, err136 }137 if err := validateHost(ctx, host, resolver, allowPrivate); err != nil {138 return nil, err139 }140 ips, err := resolver.LookupIPAddr(ctx, host)141 if err != nil || len(ips) == 0 {142 if err == nil {143 err = errors.New("no address")144 }145 return nil, fmt.Errorf("replication: resolve %s: %w", host, err)146 }147 var last error148 for _, candidate := range ips {149 if err := validateIP(candidate.IP, allowPrivate); err != nil {150 return nil, err151 }152 conn, dialErr := (&net.Dialer{}).DialContext(ctx, network, net.JoinHostPort(candidate.IP.String(), port))153 if dialErr == nil {154 return conn, nil155 }156 last = dialErr157 }158 return nil, last159 }160 client := *base161 client.Transport = transport162 oldRedirect := base.CheckRedirect163 client.CheckRedirect = func(req *http.Request, via []*http.Request) error {164 if err := validateURL(req.URL); err != nil {165 return err166 }167 if err := validateHost(req.Context(), req.URL.Hostname(), resolver, allowPrivate); err != nil {168 return err169 }170 if len(via) >= 10 {171 return errors.New("replication: stopped after 10 redirects")172 }173 if oldRedirect != nil {174 return oldRedirect(req, via)175 }176 return nil177 }178 return &client179}180181func validateURL(target *url.URL) error {182 if target == nil || (target.Scheme != "ws" && target.Scheme != "wss" && target.Scheme != "http" && target.Scheme != "https") || target.Host == "" || target.User != nil || target.Fragment != "" {183 return errors.New("replication: invalid websocket redirect")184 }185 return nil186}187188func validateHost(ctx context.Context, host string, resolver IPResolver, allowPrivate bool) error {189 if host == "" {190 return errors.New("replication: websocket host is empty")191 }192 if ip := net.ParseIP(host); ip != nil {193 return validateIP(ip, allowPrivate)194 }195 ips, err := resolver.LookupIPAddr(ctx, host)196 if err != nil {197 return fmt.Errorf("replication: resolve %s: %w", host, err)198 }199 if len(ips) == 0 {200 return fmt.Errorf("replication: resolve %s: no address", host)201 }202 for _, item := range ips {203 if err := validateIP(item.IP, allowPrivate); err != nil {204 return err205 }206 }207 return nil208}209210func validateIP(ip net.IP, allowPrivate bool) error {211 if allowPrivate {212 return nil213 }214 if unusableIP(ip) {215 return fmt.Errorf("replication: private or local relay address %s is blocked", ip)216 }217 return nil218}219220func unusableIP(ip net.IP) bool {221 if ip == nil || ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() || ip.Equal(net.ParseIP("100.100.100.100")) {222 return true223 }224 v4 := ip.To4()225 if v4 == nil {226 return false227 }228 // CGNAT/Tailscale and documentation/reserved IPv4 ranges are not public229 // relay destinations unless the operator explicitly permits private peers.230 if v4[0] == 100 && v4[1]&0xc0 == 64 {231 return true232 }233 if v4[0] == 192 && v4[1] == 0 && v4[2] == 0 {234 return true235 }236 if v4[0] == 198 && v4[1] >= 18 && v4[1] <= 19 {237 return true238 }239 return v4[0] >= 240240}241242type coderSocket struct{ conn *websocket.Conn }243244func (s coderSocket) Read(ctx context.Context) ([]byte, error) {245 _, data, err := s.conn.Read(ctx)246 return data, err247}248249func (s coderSocket) Write(ctx context.Context, data []byte) error {250 return s.conn.Write(ctx, websocket.MessageText, data)251}252253func (s coderSocket) Close(err error) error {254 code := websocket.StatusNormalClosure255 if err != nil {256 code = websocket.StatusInternalError257 }258 return s.conn.Close(code, "replication complete")259}260261// NostrTransport implements PullTransport and PushTransport over Nostr262// websocket protocols. Dialer defaults to WebsocketDialer when unset, and263// Timeout defaults to 20 seconds.264type NostrTransport struct {265 Dialer Dialer266 Timeout time.Duration267 sequence atomic.Uint64268 legacyMu sync.Mutex269 legacy map[string]struct{}270 // LegacyCache may be shared by short-lived transports created for the same271 // tenant. Entries expire so a relay can gain NIP-77 support without a272 // process restart.273 LegacyCache *LegacyCache274}275276const legacyProbeTTL = 4 * time.Hour277278type LegacyCache struct {279 mu sync.Mutex280 entries map[string]time.Time281}282283func (c *LegacyCache) known(target string) bool {284 c.mu.Lock()285 defer c.mu.Unlock()286 when, ok := c.entries[target]287 if !ok || time.Since(when) >= legacyProbeTTL {288 if ok {289 delete(c.entries, target)290 }291 return false292 }293 return true294}295296func (c *LegacyCache) remember(target string) {297 c.mu.Lock()298 defer c.mu.Unlock()299 if c.entries == nil {300 c.entries = make(map[string]time.Time)301 }302 if len(c.entries) >= 64 {303 var oldest string304 var oldestAt time.Time305 for key, at := range c.entries {306 if oldest == "" || at.Before(oldestAt) {307 oldest, oldestAt = key, at308 }309 }310 delete(c.entries, oldest)311 }312 c.entries[target] = time.Now()313}314315// Subscribe keeps a bounded NIP-01 subscription open until ctx is canceled.316// Reconnection belongs to the lifecycle owner so it can apply peer backoff317// and avoid multiplying sockets when a relay is unhealthy.318func (t *NostrTransport) Subscribe(ctx context.Context, target string, filter event.Filter, onEvent func(event.Event) error) error {319 return t.SubscribeFilters(ctx, target, []event.Filter{filter}, onEvent)320}321322// SubscribeFilters opens one subscription for a set of NIP-01 filters. Relays323// treat filters in one REQ as an OR, which lets callers follow related event324// families without opening a socket for each family.325func (t *NostrTransport) SubscribeFilters(ctx context.Context, target string, filters []event.Filter, onEvent func(event.Event) error) error {326 if len(filters) == 0 {327 return errors.New("replication: live subscription has no filters")328 }329 socket, _, err := t.dial(ctx, target)330 if err != nil {331 return err332 }333 defer socket.Close(nil)334 sub := fmt.Sprintf("tiny-live-%d", t.sequence.Add(1))335 prepared := make([]event.Filter, len(filters))336 for i, filter := range filters {337 filter.Tags = ensureTags(filter.Tags)338 prepared[i] = filter339 }340 message, err := json.Marshal(append([]any{"REQ", sub}, filtersToAny(prepared)...))341 if err != nil {342 return fmt.Errorf("encode subscription: %w", err)343 }344 if err := socket.Write(ctx, message); err != nil {345 return err346 }347 for {348 data, err := socket.Read(ctx)349 if err != nil {350 return err351 }352 var raw []json.RawMessage353 if json.Unmarshal(data, &raw) != nil || len(raw) < 2 {354 continue355 }356 var kind, id string357 _ = json.Unmarshal(raw[0], &kind)358 _ = json.Unmarshal(raw[1], &id)359 if id != sub {360 continue361 }362 if kind == "CLOSED" {363 return errors.New("replication: live subscription closed")364 }365 if kind != "EVENT" || len(raw) < 3 {366 continue367 }368 var item event.Event369 if err := json.Unmarshal(raw[2], &item); err != nil {370 continue371 }372 if err := event.Validate(item); err != nil || !filterMatchesAny(prepared, item) {373 continue374 }375 if onEvent != nil {376 if err := onEvent(item); err != nil {377 return err378 }379 }380 }381}382383func filtersToAny(filters []event.Filter) []any {384 out := make([]any, len(filters))385 for i := range filters {386 out[i] = filters[i]387 }388 return out389}390391func filterMatchesAny(filters []event.Filter, item event.Event) bool {392 for _, filter := range filters {393 if event.Matches(filter, item) {394 return true395 }396 }397 return false398}399400const (401 // pullPageSize keeps a compatibility REQ small enough for relays to serve402 // without allocating an unbounded result set.403 pullPageSize = 500404 // pullPageLimit prevents a corrupt or very large repository announcement405 // from turning one maintenance pass into an unbounded import.406 pullPageLimit = 20407 pullEventLimit = pullPageSize * pullPageLimit408)409410// ErrPullIncomplete indicates that a relay returned a full bounded history411// window but the next page cannot be represented safely by an `until` cursor.412// Callers should retain the result and retry with a larger or more capable413// synchronization method.414var ErrPullIncomplete = errors.New("replication: pull history is incomplete")415416// ErrNegentropyUnsupported means the peer does not implement NIP-77. It is417// deliberately distinct from authentication, protocol and transport errors:418// only this condition is safe to downgrade to an ordinary REQ pull.419var ErrNegentropyUnsupported = errors.New("replication: negentropy unsupported")420421// QuerySynchronized prefers NIP-77 negentropy and falls back to bounded,422// descending REQ pages when the peer does not implement it. GRASP peers are423// commonly mixed-version, so lack of NIP-77 support must not prevent the424// ordinary NIP-01 pull from completing.425func (t *NostrTransport) QuerySynchronized(ctx context.Context, target string, filter event.Filter, local []syncprotocol.Item) ([]event.Event, error) {426 if t.knownLegacy(target) {427 return t.QueryPaginated(ctx, target, filter)428 }429 items, err := t.QueryNegentropy(ctx, target, filter, local)430 if err == nil {431 return items, nil432 }433 if !errors.Is(err, ErrNegentropyUnsupported) {434 return items, err435 }436 t.rememberLegacy(target)437 return t.QueryPaginated(ctx, target, filter)438}439440func (t *NostrTransport) knownLegacy(target string) bool {441 if t.LegacyCache != nil && t.LegacyCache.known(target) {442 return true443 }444 t.legacyMu.Lock()445 defer t.legacyMu.Unlock()446 _, ok := t.legacy[target]447 return ok448}449450func (t *NostrTransport) rememberLegacy(target string) {451 if t.LegacyCache != nil {452 t.LegacyCache.remember(target)453 return454 }455 t.legacyMu.Lock()456 defer t.legacyMu.Unlock()457 if t.legacy == nil {458 t.legacy = make(map[string]struct{})459 }460 // Bound the instance-local compatibility cache so untrusted relay URLs461 // cannot accumulate indefinitely. Once full, an unremembered peer simply462 // receives another bounded probe on a later sync.463 if len(t.legacy) < 64 {464 t.legacy[target] = struct{}{}465 }466}467468// QueryPaginated reads a bounded history window. It uses the oldest event in469// each page as the next upper bound, deduplicating events because relays may470// include the boundary event in their response.471func (t *NostrTransport) QueryPaginated(ctx context.Context, target string, filter event.Filter) ([]event.Event, error) {472 result := make([]event.Event, 0, pullPageSize)473 seen := make(map[string]struct{})474 pageFilter := filter475 for page := 0; page < pullPageLimit; page++ {476 if err := ctx.Err(); err != nil {477 return result, err478 }479 limit := pullPageSize480 pageFilter.Limit = &limit481 items, err := t.Query(ctx, target, pageFilter)482 added := 0483 for _, item := range items {484 if _, ok := seen[item.ID]; ok {485 continue486 }487 seen[item.ID] = struct{}{}488 result = append(result, item)489 added++490 }491 if len(result) >= pullEventLimit {492 return result, ErrPullIncomplete493 }494 if err != nil {495 return result, err496 }497 if len(items) == 0 {498 return result, nil499 }500 oldest := items[0].CreatedAt501 for _, item := range items {502 if item.CreatedAt < oldest {503 oldest = item.CreatedAt504 }505 }506 // The inclusive boundary is an overlap probe. If it yields no new507 // events, the relay has no older matching events available through this508 // cursor; a short response is therefore not treated as definitive by509 // itself.510 if page > 0 && added == 0 {511 allAtBoundary := true512 for _, item := range items {513 if item.CreatedAt != oldest {514 allAtBoundary = false515 break516 }517 }518 if allAtBoundary && len(items) >= pullPageSize {519 return result, ErrPullIncomplete520 }521 return result, nil522 }523 if oldest <= 0 {524 return result, nil525 }526 // Keep the boundary inclusive. A relay may cap every response below527 // our requested limit, and a strict cursor would silently drop events528 // sharing the oldest timestamp. Deduplication makes the overlap safe.529 until := oldest530 pageFilter.Until = &until531 }532 return result, ErrPullIncomplete533}534535// QueryNegentropy reconciles a local index with a remote relay, then fetches536// only IDs the local side lacks. It is optional at the PullTransport boundary537// so ordinary REQ remains a compatible fallback.538func (t *NostrTransport) QueryNegentropy(ctx context.Context, target string, filter event.Filter, local []syncprotocol.Item) ([]event.Event, error) {539 session, err := syncprotocol.NewSession(local, true)540 if err != nil {541 return nil, err542 }543 start, err := session.Start()544 if err != nil {545 return nil, err546 }547 socket, _, err := t.dial(ctx, target)548 if err != nil {549 return nil, err550 }551 defer socket.Close(nil)552 id := fmt.Sprintf("tiny-neg-%d", t.sequence.Add(1))553 filter.Tags = ensureTags(filter.Tags)554 open, err := json.Marshal([]any{"NEG-OPEN", id, filter, start})555 if err != nil {556 return nil, err557 }558 if err := socket.Write(ctx, open); err != nil {559 return nil, err560 }561 var needed []string562 progress := false563 deadline, cancel := context.WithTimeout(ctx, t.timeout())564 defer cancel()565 for {566 data, err := socket.Read(deadline)567 if err != nil {568 // Some legacy relays accept the websocket but silently ignore569 // NEG-OPEN. A bounded probe timeout is the only safe downgrade; a570 // caller cancellation or an authenticated/protocol response must571 // remain an error.572 if errors.Is(err, context.DeadlineExceeded) && !progress && ctx.Err() == nil {573 return nil, ErrNegentropyUnsupported574 }575 return nil, err576 }577 var raw []json.RawMessage578 if json.Unmarshal(data, &raw) != nil || len(raw) < 2 {579 continue580 }581 var kind, messageID, body string582 _ = json.Unmarshal(raw[0], &kind)583 _ = json.Unmarshal(raw[1], &messageID)584 if len(raw) > 2 {585 _ = json.Unmarshal(raw[2], &body)586 }587 if kind == "AUTH" {588 return nil, errors.New("replication: remote relay requires authentication")589 }590 if messageID != id {591 continue592 }593 // A responder may close immediately when its authorized snapshot is594 // already identical. Treat that as a completed empty reconciliation;595 // waiting for a NEG-MSG that will never arrive turns a successful NIP-77596 // no-op sync into a timeout.597 if kind == "NEG-CLOSE" {598 progress = true599 break600 }601 if kind == "CLOSED" {602 return nil, errors.New("replication: remote negentropy closed")603 }604 if kind == "NEG-ERR" {605 if negentropyUnsupported(body) {606 return nil, ErrNegentropyUnsupported607 }608 return nil, fmt.Errorf("replication: remote negentropy rejected: %s", body)609 }610 if kind != "NEG-MSG" {611 continue612 }613 progress = true614 result, err := session.Reconcile(body)615 if err != nil {616 return nil, err617 }618 needed = append(needed, result.Need...)619 if len(needed) > pullEventLimit {620 return nil, ErrPullIncomplete621 }622 if result.Done {623 closeMessage, _ := json.Marshal([]any{"NEG-CLOSE", id})624 _ = socket.Write(ctx, closeMessage)625 break626 }627 reply, err := json.Marshal([]any{"NEG-MSG", id, result.Response})628 if err != nil {629 return nil, err630 }631 if err := socket.Write(ctx, reply); err != nil {632 return nil, err633 }634 }635 if len(needed) == 0 {636 return nil, nil637 }638 return t.queryNegentropyIDs(ctx, target, filter, needed)639}640641func negentropyUnsupported(reason string) bool {642 reason = strings.ToLower(strings.TrimSpace(reason))643 if strings.HasPrefix(reason, "unsupported") || strings.HasPrefix(reason, "error: unsupported") {644 return true645 }646 switch reason {647 case "unsupported", "not supported", "negentropy unsupported", "nip-77 unsupported":648 return true649 default:650 return false651 }652}653654const negentropyFetchBatch = 500655656func (t *NostrTransport) queryNegentropyIDs(ctx context.Context, target string, filter event.Filter, ids []string) ([]event.Event, error) {657 result := make([]event.Event, 0, len(ids))658 seen := make(map[string]struct{}, len(ids))659 for start := 0; start < len(ids); start += negentropyFetchBatch {660 end := start + negentropyFetchBatch661 if end > len(ids) {662 end = len(ids)663 }664 batch := append([]string(nil), ids[start:end]...)665 request := filter666 request.IDs = batch667 limit := len(batch)668 request.Limit = &limit669 events, err := t.Query(ctx, target, request)670 if err != nil {671 for _, candidate := range events {672 if _, ok := seen[candidate.ID]; ok {673 continue674 }675 seen[candidate.ID] = struct{}{}676 result = append(result, candidate)677 }678 return result, err679 }680 allowed := make(map[string]struct{}, len(batch))681 for _, id := range batch {682 allowed[id] = struct{}{}683 }684 for _, candidate := range events {685 if _, ok := allowed[candidate.ID]; !ok {686 continue687 }688 if !event.Matches(filter, candidate) {689 continue690 }691 if _, ok := seen[candidate.ID]; ok {692 continue693 }694 seen[candidate.ID] = struct{}{}695 result = append(result, candidate)696 }697 }698 if len(seen) != len(uniqueIDs(ids)) {699 return result, ErrPullIncomplete700 }701 return result, nil702}703704func uniqueIDs(ids []string) map[string]struct{} {705 unique := make(map[string]struct{}, len(ids))706 for _, id := range ids {707 unique[id] = struct{}{}708 }709 return unique710}711712func (t *NostrTransport) timeout() time.Duration {713 if t.Timeout > 0 {714 return t.Timeout715 }716 return 20 * time.Second717}718719func (t *NostrTransport) Query(ctx context.Context, target string, filter event.Filter) ([]event.Event, error) {720 socket, _, err := t.dial(ctx, target)721 if err != nil {722 return nil, err723 }724 defer socket.Close(nil)725 sub := fmt.Sprintf("tiny-pull-%d", t.sequence.Add(1))726 filter.Tags = ensureTags(filter.Tags)727 message, err := json.Marshal([]any{"REQ", sub, filter})728 if err != nil {729 return nil, fmt.Errorf("encode query: %w", err)730 }731 if err := socket.Write(ctx, message); err != nil {732 return nil, err733 }734 return readEvents(ctx, socket, sub, t.timeout(), filter)735}736737func (t *NostrTransport) Send(ctx context.Context, target string, e event.Event) (DeliveryResult, error) {738 socket, _, err := t.dial(ctx, target)739 if err != nil {740 return DeliveryResult{}, err741 }742 defer socket.Close(nil)743 message, err := json.Marshal([]any{"EVENT", e})744 if err != nil {745 return DeliveryResult{}, err746 }747 if err := socket.Write(ctx, message); err != nil {748 return DeliveryResult{}, err749 }750 deadline, cancel := context.WithTimeout(ctx, t.timeout())751 defer cancel()752 for {753 data, err := socket.Read(deadline)754 if err != nil {755 return DeliveryResult{}, err756 }757 var message []json.RawMessage758 if json.Unmarshal(data, &message) != nil || len(message) < 4 {759 continue760 }761 var kind, id, text string762 var accepted bool763 _ = json.Unmarshal(message[0], &kind)764 _ = json.Unmarshal(message[1], &id)765 _ = json.Unmarshal(message[2], &accepted)766 _ = json.Unmarshal(message[3], &text)767 if kind == "OK" && id == e.ID {768 return DeliveryResult{Accepted: accepted, Message: text}, nil769 }770 }771}772773func (t *NostrTransport) dial(ctx context.Context, target string) (Socket, *http.Response, error) {774 dialer := t.Dialer775 if dialer == nil {776 dialer = WebsocketDialer{}777 }778 return dialer.Dial(ctx, target)779}780781func readEvents(ctx context.Context, socket Socket, subscription string, timeout time.Duration, filter event.Filter) ([]event.Event, error) {782 deadline, cancel := context.WithTimeout(ctx, timeout)783 defer cancel()784 items := make([]event.Event, 0)785 for {786 data, err := socket.Read(deadline)787 if err != nil {788 return items, err789 }790 var raw []json.RawMessage791 if err := json.Unmarshal(data, &raw); err != nil || len(raw) < 2 {792 continue793 }794 var kind, id string795 _ = json.Unmarshal(raw[0], &kind)796 _ = json.Unmarshal(raw[1], &id)797 switch kind {798 case "EVENT":799 if id != subscription || len(raw) < 3 {800 continue801 }802 var item event.Event803 if json.Unmarshal(raw[2], &item) == nil && event.Validate(item) == nil && event.Matches(filter, item) {804 items = append(items, item)805 if len(items) > pullEventLimit {806 return items[:pullEventLimit], ErrPullIncomplete807 }808 }809 case "EOSE":810 if id == subscription {811 return items, nil812 }813 case "CLOSED":814 if id == subscription {815 return items, errors.New("replication: remote query closed")816 }817 case "AUTH":818 return items, errors.New("replication: remote relay requires authentication")819 }820 }821}822823func ensureTags(tags map[string][]string) map[string][]string {824 if tags != nil {825 return tags826 }827 return map[string][]string{}828}