1package relay23import (4 "context"5 "crypto/rand"6 "encoding/hex"7 "encoding/json"8 "errors"9 "fmt"10 "net"11 "net/http"12 "net/url"13 "strings"14 "sync"15 "time"1617 "github.com/FelineStateMachine/tinyrelay/internal/event"18 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"19 "github.com/FelineStateMachine/tinyrelay/protocol/auth"20 "github.com/coder/websocket"21)2223type Session = auth.Session2425// Backend publishes while the Router's write gate is held. Publish performs26// the backend's admission, persistence, duplicate handling or control action27// before returning. A nil error allows Router to fan out the event unless the28// returned reason identifies a duplicate. Persistence and delivery planning29// remain backend responsibilities for event kinds that require them.30type Backend interface {31 Publish(ctx context.Context, event event.Event, session Session) (reason string, err error)32 Query(ctx context.Context, filters []event.Filter, session Session) ([]event.Event, error)33 Count(ctx context.Context, filters []event.Filter, session Session) (any, error)34}3536// Reader-side privacy can be stricter than the query filter. It is checked for37// both historical and live delivery, including encrypted signer traffic.38type Reader interface {39 CanRead(event event.Event, session Session) bool40}4142// FilterReader applies reader privacy with the subscription filter available43// to the policy decision.44type FilterReader interface {45 CanReadFilter(event event.Event, session Session, filter *event.Filter) bool46}4748// QueryHints can return query results together with completion hints. Router49// places these values in the EOSE response; the relay backend uses values such50// as "finish", "more" and "auth".51type QueryHints interface {52 QueryHints(ctx context.Context, filters []event.Filter, session Session) ([]event.Event, []string, error)53}5455// PolicyChanges can invalidate subscriptions in storage after an access56// policy revision. Router removes client state before invoking the callback.57type PolicyChanges interface {58 CloseSubscriptions(session Session)59}6061// SyncBackend supplies the authorized event snapshot for a NIP-77 session.62// The Router holds its read gate while calling Sync, so a concurrent Publish63// cannot create an opening between snapshot selection and NEG-OPEN.64type SyncBackend interface {65 Sync(ctx context.Context, filter event.Filter, session Session) ([]syncprotocol.Item, error)66}6768type Config struct {69 // RelayURL is the default relay identity included in sessions.70 RelayURL string71 // RequestRelayURL optionally derives the relay identity from each request.72 RequestRelayURL func(*http.Request) string73 // OnAuthenticate checks the authenticated session candidate and may persist74 // authentication bookkeeping. The Session argument is a value snapshot;75 // Router records the newly authenticated public key after this callback.76 OnAuthenticate func(context.Context, Session) error77 // MaxMessageBytes limits websocket text messages. Zero leaves the limit78 // unlimited; a negative value selects the 512 KiB default.79 MaxMessageBytes int6480 // MaxPendingBytes bounds queued output per connection.81 MaxPendingBytes int82 // ReadLimit, WriteLimit and PingInterval control websocket liveness.83 ReadLimit time.Duration84 WriteLimit time.Duration85 PingInterval time.Duration86 // OriginPatterns is passed to the websocket origin check.87 OriginPatterns []string88 // OnConnection and OnSubscription receive +1 and -1 lifecycle deltas.89 OnConnection func(int)90 OnSubscription func(int)91 // ChangesAccess identifies host-specific events that invalidate existing92 // subscriptions. A nil value leaves subscriptions intact.93 ChangesAccess func(event.Event) bool94}9596func (c Config) withDefaults() Config {97 if c.MaxMessageBytes < 0 {98 c.MaxMessageBytes = 512 << 1099 }100 if c.MaxPendingBytes <= 0 {101 c.MaxPendingBytes = 4 << 20102 }103 if c.ReadLimit <= 0 {104 c.ReadLimit = 60 * time.Second105 }106 if c.WriteLimit <= 0 {107 c.WriteLimit = 10 * time.Second108 }109 if c.PingInterval <= 0 {110 c.PingInterval = 30 * time.Second111 }112 return c113}114115type Router struct {116 backend Backend117 cfg Config118 mu sync.RWMutex // publish is exclusive with REQ registration + query119 clientsMu sync.Mutex120 clients map[*client]struct{}121 listenersMu sync.Mutex122 listeners map[*listener]struct{}123 closed chan struct{}124 closeOnce sync.Once125 clientsWG sync.WaitGroup126}127128type listener struct{ fn func(event.Event) }129130func New(backend Backend, cfg Config) *Router {131 return &Router{backend: backend, cfg: cfg.withDefaults(), clients: make(map[*client]struct{}), listeners: make(map[*listener]struct{}), closed: make(chan struct{})}132}133134// Listen registers an in-process observer of live fan-out. It sees every135// event delivered to websocket subscribers, before the reader's own access136// check. The callback runs under the publish fence and must not block; the137// returned function removes it.138func (r *Router) Listen(fn func(event.Event)) func() {139 l := &listener{fn: fn}140 r.listenersMu.Lock()141 r.listeners[l] = struct{}{}142 r.listenersMu.Unlock()143 return func() {144 r.listenersMu.Lock()145 delete(r.listeners, l)146 r.listenersMu.Unlock()147 }148}149150func (r *Router) HandleHTTP(w http.ResponseWriter, req *http.Request) {151 if req.Method != http.MethodGet {152 http.Error(w, "method not allowed", http.StatusMethodNotAllowed)153 return154 }155 conn, err := websocket.Accept(w, req, &websocket.AcceptOptions{OriginPatterns: r.cfg.OriginPatterns})156 if err != nil {157 return158 }159 c := newClient(r, conn, req)160 r.clientsMu.Lock()161 select {162 case <-r.closed:163 r.clientsMu.Unlock()164 _ = conn.Close(websocket.StatusGoingAway, "relay shutting down")165 return166 default:167 }168 r.clients[c] = struct{}{}169 if r.cfg.OnConnection != nil {170 r.cfg.OnConnection(1)171 }172 r.clientsWG.Add(1)173 r.clientsMu.Unlock()174 go c.run()175}176177// Publish is the shared ingress path for websocket EVENT and HTTP-generated178// events. It applies the same historical/live fence and fan-out rules.179func (r *Router) Publish(ctx context.Context, e event.Event, s Session) (string, error) {180 r.mu.Lock()181 defer r.mu.Unlock()182 reason, err := r.backend.Publish(ctx, e, s)183 if err == nil && !duplicateReason(reason) {184 if r.changesAccess(e) {185 r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")186 }187 r.fanout(e)188 }189 return reason, err190}191192func (r *Router) publishClient(c *client, e event.Event, s Session) (string, error) {193 r.mu.Lock()194 defer r.mu.Unlock()195 reason, err := r.backend.Publish(c.ctx, e, s)196 if err != nil {197 return reason, err198 }199 // Queue the sender's OK before any live EVENT fanout on that socket.200 c.replyOK(e.ID, true, reason)201 if r.changesAccess(e) {202 c.r.closeSubscriptionsLocked("blocked: relay access policy changed; subscribe again")203 }204 if reason == "" {205 r.fanout(e)206 }207 return reason, nil208}209210// BroadcastGenerated delivers an event already durably accepted by an211// out-of-band producer.212func (r *Router) BroadcastGenerated(e event.Event) {213 r.mu.Lock()214 defer r.mu.Unlock()215 r.fanout(e)216}217218// CommitGenerated runs an imported or server-generated event commit under the219// same historical/live fence as client EVENT publication. The callback must220// durably commit the event before returning; a returned error suppresses221// fanout. This prevents a REQ snapshot from racing an out-of-band broadcast.222func (r *Router) CommitGenerated(ctx context.Context, commit func(context.Context) (event.Event, error)) (event.Event, error) {223 r.mu.Lock()224 defer r.mu.Unlock()225 e, err := commit(ctx)226 if err == nil && e.ID != "" {227 r.fanout(e)228 }229 return e, err230}231232func (r *Router) Close(ctx context.Context) error {233 r.closeOnce.Do(func() { close(r.closed) })234 r.clientsMu.Lock()235 cs := make([]*client, 0, len(r.clients))236 for c := range r.clients {237 cs = append(cs, c)238 }239 r.clientsMu.Unlock()240 for _, c := range cs {241 c.close(websocket.StatusGoingAway, "relay shutting down")242 }243 select {244 case <-ctx.Done():245 return ctx.Err()246 case <-r.clientsDone():247 return nil248 }249}250251// CloseSubscriptions invalidates all live subscriptions after a policy252// revision. Existing sockets remain usable for AUTH and a subsequent REQ.253func (r *Router) CloseSubscriptions(reason string) {254 r.mu.Lock()255 defer r.mu.Unlock()256 r.closeSubscriptionsLocked(reason)257}258259// ApplyACLChange serializes an access-control mutation with REQ snapshots,260// NEG sessions, and live publication. A successful mutation invalidates both261// subscription families before another transport operation can begin.262func (r *Router) ApplyACLChange(ctx context.Context, change func(context.Context) (any, error), reason string) (any, error) {263 r.mu.Lock()264 defer r.mu.Unlock()265 result, err := change(ctx)266 if err == nil {267 r.closeSubscriptionsLocked(reason)268 }269 return result, err270}271272func (r *Router) closeSubscriptionsLocked(reason string) {273 if reason == "" {274 reason = "blocked: relay policy changed"275 }276 r.clientsMu.Lock()277 cs := make([]*client, 0, len(r.clients))278 for c := range r.clients {279 cs = append(cs, c)280 }281 r.clientsMu.Unlock()282 for _, c := range cs {283 c.mu.Lock()284 ids := make([]string, 0, len(c.subs))285 for id := range c.subs {286 ids = append(ids, id)287 delete(c.subs, id)288 }289 syncIDs := make([]string, 0, len(c.syncs))290 for id := range c.syncs {291 syncIDs = append(syncIDs, id)292 delete(c.syncs, id)293 }294 c.mu.Unlock()295 for _, id := range ids {296 if r.cfg.OnSubscription != nil {297 r.cfg.OnSubscription(-1)298 }299 b, _ := json.Marshal([]any{"CLOSED", id, reason})300 _ = c.enqueueBytes(b)301 }302 for _, id := range syncIDs {303 b, _ := json.Marshal([]any{"NEG-CLOSE", id})304 _ = c.enqueueBytes(b)305 }306 }307 if p, ok := r.backend.(PolicyChanges); ok {308 p.CloseSubscriptions(Session{RelayURL: r.cfg.RelayURL})309 }310}311312func (r *Router) clientsDone() <-chan struct{} {313 done := make(chan struct{})314 go func() { r.clientsWG.Wait(); close(done) }()315 return done316}317318type subscription struct{ filters []event.Filter }319type outbound struct {320 data []byte321 size int322}323324type client struct {325 r *Router326 conn *websocket.Conn327 ctx context.Context328 cancel context.CancelFunc329 mu sync.Mutex330 session Session331 challenge string332 subs map[string]subscription333 syncs map[string]*syncprotocol.Session334 queue []outbound335 pending int336 wake chan struct{}337 done chan struct{}338 once sync.Once339 wg sync.WaitGroup340}341342func newClient(r *Router, conn *websocket.Conn, req *http.Request) *client {343 ctx, cancel := context.WithCancel(context.Background())344 b := make([]byte, 16)345 _, _ = rand.Read(b)346 relayURL := r.cfg.RelayURL347 if r.cfg.RequestRelayURL != nil {348 if requested := strings.TrimSpace(r.cfg.RequestRelayURL(req)); requested != "" {349 relayURL = requested350 }351 }352 return &client{r: r, conn: conn, ctx: ctx, cancel: cancel, challenge: hex.EncodeToString(b), subs: make(map[string]subscription), syncs: make(map[string]*syncprotocol.Session), wake: make(chan struct{}, 1), done: make(chan struct{}), session: Session{RelayURL: relayURL, RemoteIP: remoteIP(req.RemoteAddr)}}353}354355func (c *client) run() {356 defer c.r.clientsWG.Done()357 c.wg.Add(1)358 go c.writer()359 _ = c.enqueue(fmt.Sprintf(`["AUTH",%q]`, c.challenge))360 c.readLoop()361 c.close(websocket.StatusNormalClosure, "closed")362 c.wg.Wait()363}364365func (c *client) readLoop() {366 defer c.conn.CloseNow()367 readLimit := c.r.cfg.MaxMessageBytes368 if readLimit == 0 {369 readLimit = -1370 }371 c.conn.SetReadLimit(readLimit)372 for {373 // Read contexts must live for the connection. A per-message timeout374 // measures time between application messages, so it closes a healthy375 // idle session even when the websocket ping/pong control traffic is376 // flowing. The writer owns protocol keepalive; cancellation closes a377 // connection during shutdown.378 typ, data, err := c.conn.Read(c.ctx)379 if err != nil {380 return381 }382 if typ != websocket.MessageText {383 _ = c.enqueue(`["NOTICE","error: binary frames are not supported"]`)384 continue385 }386 if c.r.cfg.MaxMessageBytes > 0 && int64(len(data)) > c.r.cfg.MaxMessageBytes {387 _ = c.enqueue(`["NOTICE","error: message too large"]`)388 continue389 }390 var raw []json.RawMessage391 if err := json.Unmarshal(data, &raw); err != nil || len(raw) == 0 {392 _ = c.enqueue(`["NOTICE","error: could not parse message"]`)393 continue394 }395 var typName string396 if json.Unmarshal(raw[0], &typName) != nil {397 _ = c.enqueue(`["NOTICE","error: message type must be a string"]`)398 continue399 }400 c.dispatch(typName, raw[1:])401 }402}403404func (c *client) dispatch(kind string, args []json.RawMessage) {405 switch kind {406 case "EVENT":407 c.handleEvent(args)408 case "REQ", "COUNT":409 c.handleRequest(kind, args)410 case "CLOSE":411 c.handleClose(args)412 case "AUTH":413 c.handleAuth(args)414 case "NEG-OPEN", "NEG-MSG", "NEG-CLOSE":415 c.handleSync(kind, args)416 default:417 _ = c.enqueue(fmt.Sprintf(`["NOTICE","error: unknown message type %s"]`, kind))418 }419}420421func (c *client) handleSync(kind string, args []json.RawMessage) {422 if len(args) < 1 {423 _ = c.enqueue(fmt.Sprintf(`["NEG-ERR","",%q]`, "error: "+kind+" needs a subscription id"))424 return425 }426 var id string427 if json.Unmarshal(args[0], &id) != nil || id == "" || len(id) > 64 {428 _ = c.enqueue(`["NEG-ERR","","error: bad subscription id"]`)429 return430 }431 if kind == "NEG-CLOSE" {432 c.mu.Lock()433 delete(c.syncs, id)434 c.mu.Unlock()435 return436 }437 if kind == "NEG-OPEN" {438 if len(args) < 3 {439 c.syncError(id, "error: NEG-OPEN needs a filter and a message")440 return441 }442 f, err := event.ParseFilter(args[1])443 if err != nil {444 c.syncError(id, "invalid: bad filter: "+err.Error())445 return446 }447 var message string448 if json.Unmarshal(args[2], &message) != nil {449 c.syncError(id, "invalid: message must be hex")450 return451 }452 if _, err := hex.DecodeString(message); err != nil {453 c.syncError(id, "invalid: message must be hex")454 return455 }456 b, ok := c.r.backend.(SyncBackend)457 if !ok {458 c.syncError(id, "unsupported: sync is switched off on this relay")459 return460 }461 c.r.mu.RLock()462 items, err := b.Sync(c.ctx, f, c.snapshot())463 if err == nil {464 var sess *syncprotocol.Session465 sess, err = syncprotocol.NewSession(items, false)466 if err == nil {467 var result syncprotocol.Result468 result, err = sess.Reconcile(message)469 if err == nil {470 if result.Response == "" {471 _ = c.enqueue(fmt.Sprintf(`["NEG-CLOSE",%q]`, id))472 } else {473 c.mu.Lock()474 c.syncs[id] = sess475 c.mu.Unlock()476 c.sendSyncMessage(id, result.Response)477 }478 }479 }480 }481 c.r.mu.RUnlock()482 if err != nil {483 c.syncError(id, syncErrorReason(err))484 }485 return486 }487 if len(args) < 2 {488 c.syncError(id, "error: NEG-MSG needs a message")489 return490 }491 var message string492 if json.Unmarshal(args[1], &message) != nil {493 c.syncError(id, "invalid: message must be hex")494 return495 }496 if _, err := hex.DecodeString(message); err != nil {497 c.syncError(id, "invalid: message must be hex")498 return499 }500 c.mu.Lock()501 sess := c.syncs[id]502 c.mu.Unlock()503 if sess == nil {504 c.syncError(id, "closed: no such sync, send NEG-OPEN first")505 return506 }507 result, err := sess.Reconcile(message)508 if err != nil {509 c.mu.Lock()510 delete(c.syncs, id)511 c.mu.Unlock()512 c.syncError(id, "invalid: "+err.Error())513 return514 }515 if result.Response == "" {516 c.mu.Lock()517 delete(c.syncs, id)518 c.mu.Unlock()519 _ = c.enqueue(fmt.Sprintf(`["NEG-CLOSE",%q]`, id))520 } else {521 c.sendSyncMessage(id, result.Response)522 }523}524525func (c *client) sendSyncMessage(id, message string) {526 b, _ := json.Marshal([]any{"NEG-MSG", id, message})527 _ = c.enqueueBytes(b)528}529530func (c *client) syncError(id, reason string) {531 b, _ := json.Marshal([]any{"NEG-ERR", id, reason})532 _ = c.enqueueBytes(b)533}534535func (c *client) handleEvent(args []json.RawMessage) {536 if len(args) < 1 {537 _ = c.enqueue(`["NOTICE","error: EVENT needs an event"]`)538 return539 }540 providedID := rawEventID(args[0])541 e, err := event.Parse(args[0])542 if err != nil {543 c.replyOK(providedID, false, err.Error())544 return545 }546 if e.Kind == event.KIND_AUTH {547 c.replyOK(e.ID, false, "blocked: AUTH events use AUTH")548 return549 }550 reason, err := c.r.publishClient(c, e, c.snapshot())551 if err != nil {552 c.replyOK(e.ID, false, err.Error())553 return554 }555 _ = reason556}557558func (c *client) handleRequest(kind string, args []json.RawMessage) {559 if len(args) < 2 {560 _ = c.enqueue(fmt.Sprintf(`["NOTICE",%q]`, "error: "+kind+" needs an id and at least one filter"))561 return562 }563 var id string564 if json.Unmarshal(args[0], &id) != nil || id == "" || len(id) > 64 {565 _ = c.enqueue(`["NOTICE","error: bad subscription id"]`)566 return567 }568 filters := make([]event.Filter, 0, len(args)-1)569 for _, raw := range args[1:] {570 f, err := event.ParseFilter(raw)571 if err != nil {572 c.closedSub(id, "invalid: bad filter: "+err.Error())573 return574 }575 filters = append(filters, f)576 }577 if kind == "COUNT" {578 c.handleCount(id, filters)579 return580 }581 c.r.mu.Lock()582 c.mu.Lock()583 _, replaced := c.subs[id]584 c.subs[id] = subscription{filters: filters}585 if !replaced && c.r.cfg.OnSubscription != nil {586 c.r.cfg.OnSubscription(1)587 }588 c.mu.Unlock()589 var hints []string590 var events []event.Event591 var err error592 if hinted, ok := c.r.backend.(QueryHints); ok {593 events, hints, err = hinted.QueryHints(c.ctx, filters, c.snapshot())594 } else {595 events, err = c.r.backend.Query(c.ctx, filters, c.snapshot())596 }597 if err == nil {598 for _, e := range events {599 matched := false600 for _, f := range filters {601 if event.Matches(f, e) && c.canRead(e, &f) {602 matched = true603 break604 }605 }606 if matched {607 if !c.sendEvent(id, e) {608 err = errors.New("slow consumer")609 break610 }611 }612 }613 }614 if err == nil {615 if len(hints) > 0 {616 _ = c.enqueue(fmt.Sprintf(`["EOSE",%q,%s]`, id, mustJSON(hints)))617 } else {618 _ = c.enqueue(fmt.Sprintf(`["EOSE",%q]`, id))619 }620 }621 c.r.mu.Unlock()622 if err != nil {623 c.closedSub(id, queryErrorReason(err))624 }625}626627func (c *client) handleCount(id string, filters []event.Filter) {628 c.r.mu.RLock()629 value, err := c.r.backend.Count(c.ctx, filters, c.snapshot())630 c.r.mu.RUnlock()631 if err != nil {632 c.closedSub(id, queryErrorReason(err))633 return634 }635 _ = c.enqueue(fmt.Sprintf(`["COUNT",%q,%s]`, id, mustJSON(value)))636}637638func (c *client) handleClose(args []json.RawMessage) {639 if len(args) > 0 {640 var id string641 if json.Unmarshal(args[0], &id) == nil {642 c.mu.Lock()643 _, existed := c.subs[id]644 delete(c.subs, id)645 c.mu.Unlock()646 if existed && c.r.cfg.OnSubscription != nil {647 c.r.cfg.OnSubscription(-1)648 }649 }650 }651}652653func (c *client) handleAuth(args []json.RawMessage) {654 if len(args) < 1 {655 c.replyOK("", false, "invalid: AUTH needs an event")656 return657 }658 providedID := rawEventID(args[0])659 e, err := event.Parse(args[0])660 if err != nil {661 c.replyOK(providedID, false, err.Error())662 return663 }664 if e.Kind != event.KIND_AUTH {665 c.replyOK(e.ID, false, "invalid: auth event must be kind 22242")666 return667 }668 if abs(time.Now().Unix()-e.CreatedAt) > 600 {669 c.replyOK(e.ID, false, "invalid: auth event created_at is too far from the current time")670 return671 }672 if event.Tag(e, "challenge") != c.challenge {673 c.replyOK(e.ID, false, "invalid: auth challenge does not match")674 return675 }676 if c.session.RelayURL != "" && normalizeAuthRelay(event.Tag(e, "relay")) != normalizeAuthRelay(c.session.RelayURL) {677 c.replyOK(e.ID, false, "invalid: auth relay tag does not name this relay")678 return679 }680 candidate := c.snapshot()681 seen := false682 for _, key := range candidate.PubKeys {683 if key == e.PubKey {684 seen = true685 break686 }687 }688 if !seen {689 candidate.PubKeys = append(candidate.PubKeys, e.PubKey)690 }691 if c.r.cfg.OnAuthenticate != nil {692 if err := c.r.cfg.OnAuthenticate(c.ctx, candidate); err != nil {693 c.replyOK(e.ID, false, "error: authentication bookkeeping failed: "+err.Error())694 return695 }696 }697 if !seen {698 c.mu.Lock()699 c.session.PubKeys = append(c.session.PubKeys, e.PubKey)700 c.mu.Unlock()701 }702 c.replyOK(e.ID, true, "")703}704705func (r *Router) fanout(e event.Event) {706 r.clientsMu.Lock()707 cs := make([]*client, 0, len(r.clients))708 for x := range r.clients {709 cs = append(cs, x)710 }711 r.clientsMu.Unlock()712 for _, x := range cs {713 x.mu.Lock()714 subs := make(map[string]subscription, len(x.subs))715 for id, s := range x.subs {716 subs[id] = s717 }718 x.mu.Unlock()719 for id, s := range subs {720 for _, f := range s.filters {721 if event.Matches(f, e) && x.canRead(e, &f) {722 if !x.sendEvent(id, e) {723 x.close(websocket.StatusPolicyViolation, "slow consumer")724 }725 break726 }727 }728 }729 }730 r.listenersMu.Lock()731 ls := make([]*listener, 0, len(r.listeners))732 for l := range r.listeners {733 ls = append(ls, l)734 }735 r.listenersMu.Unlock()736 for _, l := range ls {737 l.fn(e)738 }739}740741func (c *client) canRead(e event.Event, f *event.Filter) bool {742 if x, ok := c.r.backend.(FilterReader); ok {743 return x.CanReadFilter(e, c.snapshot(), f)744 }745 if x, ok := c.r.backend.(Reader); ok {746 return x.CanRead(e, c.snapshot())747 }748 return true749}750func (c *client) snapshot() Session {751 c.mu.Lock()752 defer c.mu.Unlock()753 s := c.session754 s.PubKeys = append([]string(nil), s.PubKeys...)755 return s756}757func (c *client) sendEvent(id string, e event.Event) bool {758 b, _ := json.Marshal([]any{"EVENT", id, e})759 return c.enqueueBytes(b) == nil760}761func (c *client) replyOK(id string, ok bool, reason string) {762 b, _ := json.Marshal([]any{"OK", id, ok, reason})763 _ = c.enqueueBytes(b)764}765func (c *client) closedSub(id, reason string) {766 c.mu.Lock()767 delete(c.subs, id)768 c.mu.Unlock()769 b, _ := json.Marshal([]any{"CLOSED", id, reason})770 _ = c.enqueueBytes(b)771}772func (c *client) enqueue(s string) error { return c.enqueueBytes([]byte(s)) }773func (c *client) enqueueBytes(b []byte) error {774 c.mu.Lock()775 defer c.mu.Unlock()776 if c.ctx.Err() != nil {777 return c.ctx.Err()778 }779 if c.pending+len(b) > c.r.cfg.MaxPendingBytes {780 return errors.New("slow consumer")781 }782 c.queue = append(c.queue, outbound{data: append([]byte(nil), b...), size: len(b)})783 c.pending += len(b)784 select {785 case c.wake <- struct{}{}:786 default:787 }788 return nil789}790791func (c *client) writer() {792 defer c.wg.Done()793 ticker := time.NewTicker(c.r.cfg.PingInterval)794 defer ticker.Stop()795 for {796 c.mu.Lock()797 if len(c.queue) == 0 {798 c.mu.Unlock()799 select {800 case <-c.wake:801 case <-c.done:802 return803 case <-ticker.C:804 ctx, cancel := context.WithTimeout(c.ctx, c.r.cfg.WriteLimit)805 err := c.conn.Ping(ctx)806 cancel()807 if err != nil {808 return809 }810 }811 continue812 }813 item := c.queue[0]814 c.queue = c.queue[1:]815 c.pending -= item.size816 c.mu.Unlock()817 ctx, cancel := context.WithTimeout(c.ctx, c.r.cfg.WriteLimit)818 err := c.conn.Write(ctx, websocket.MessageText, item.data)819 cancel()820 if err != nil {821 return822 }823 }824}825func (c *client) close(code websocket.StatusCode, reason string) {826 c.once.Do(func() {827 close(c.done)828 c.cancel()829 _ = c.conn.Close(code, reason)830 c.r.clientsMu.Lock()831 delete(c.r.clients, c)832 c.r.clientsMu.Unlock()833 if c.r.cfg.OnConnection != nil {834 c.r.cfg.OnConnection(-1)835 }836 c.mu.Lock()837 subscriptions := len(c.subs)838 c.subs = make(map[string]subscription)839 c.mu.Unlock()840 if c.r.cfg.OnSubscription != nil && subscriptions > 0 {841 c.r.cfg.OnSubscription(-subscriptions)842 }843 })844}845func mustJSON(v any) string {846 b, _ := json.Marshal(v)847 if len(b) == 0 {848 return "null"849 }850 return string(b)851}852func abs(v int64) int64 {853 if v < 0 {854 return -v855 }856 return v857}858func normalizeRelay(v string) string {859 u, err := url.Parse(strings.TrimSpace(v))860 if err != nil || u.Hostname() == "" {861 return ""862 }863 return strings.ToLower(u.Hostname())864}865866func normalizeAuthRelay(v string) string {867 u, err := url.Parse(strings.TrimSpace(v))868 if err != nil || u.Scheme == "" || u.Host == "" || u.User != nil || u.Fragment != "" || u.Opaque != "" {869 return ""870 }871 host := strings.ToLower(u.Hostname())872 if port := u.Port(); port != "" {873 host += ":" + port874 }875 path := strings.TrimRight(u.EscapedPath(), "/")876 if path == "" {877 path = "/"878 }879 return strings.ToLower(u.Scheme) + "://" + host + path880}881882func remoteIP(address string) string {883 if host, _, err := net.SplitHostPort(strings.TrimSpace(address)); err == nil {884 return host885 }886 if ip := net.ParseIP(strings.TrimSpace(address)); ip != nil {887 return ip.String()888 }889 return strings.TrimSpace(address)890}891892func rawEventID(raw json.RawMessage) string {893 var probe struct {894 ID string `json:"id"`895 }896 if json.Unmarshal(raw, &probe) != nil {897 return ""898 }899 return probe.ID900}901902func queryErrorReason(err error) string {903 if err == nil {904 return ""905 }906 if knownProtocolReason(err.Error()) {907 return err.Error()908 }909 return "error: query failed: " + err.Error()910}911912func syncErrorReason(err error) string {913 if err == nil {914 return ""915 }916 if knownProtocolReason(err.Error()) {917 return err.Error()918 }919 return "error: sync failed: " + err.Error()920}921922func knownProtocolReason(reason string) bool {923 for _, prefix := range []string{"auth-required:", "restricted:", "blocked:", "invalid:", "unsupported:"} {924 if strings.HasPrefix(reason, prefix) {925 return true926 }927 }928 return false929}930931func duplicateReason(reason string) bool {932 return strings.HasPrefix(reason, "duplicate:")933}934935func (r *Router) changesAccess(e event.Event) bool {936 return r.cfg.ChangesAccess != nil && r.cfg.ChangesAccess(e)937}