|
| 1 | +// Copyright 2026 The Swarm Authors. All rights reserved. |
| 2 | +// Use of this source code is governed by a BSD-style |
| 3 | +// license that can be found in the LICENSE file. |
| 4 | + |
| 5 | +package api |
| 6 | + |
| 7 | +import ( |
| 8 | + "context" |
| 9 | + "encoding/hex" |
| 10 | + "net/http" |
| 11 | + "time" |
| 12 | + |
| 13 | + "github.com/ethersphere/bee/v2/pkg/jsonhttp" |
| 14 | + "github.com/ethersphere/bee/v2/pkg/pubsub" |
| 15 | + "github.com/ethersphere/bee/v2/pkg/swarm" |
| 16 | + "github.com/gorilla/mux" |
| 17 | + "github.com/gorilla/websocket" |
| 18 | + ma "github.com/multiformats/go-multiaddr" |
| 19 | +) |
| 20 | + |
| 21 | +func (s *Service) pubsubWsHandler(w http.ResponseWriter, r *http.Request) { |
| 22 | + logger := s.logger.WithName("pubsub").Build() |
| 23 | + |
| 24 | + paths := struct { |
| 25 | + Topic string `map:"topic" validate:"required"` |
| 26 | + }{} |
| 27 | + if response := s.mapStructure(mux.Vars(r), &paths); response != nil { |
| 28 | + response("invalid path params", logger, w) |
| 29 | + return |
| 30 | + } |
| 31 | + |
| 32 | + var topicAddr [32]byte |
| 33 | + if decoded, err := hex.DecodeString(paths.Topic); err == nil && len(decoded) == swarm.HashSize { |
| 34 | + copy(topicAddr[:], decoded) |
| 35 | + } else { |
| 36 | + h := swarm.NewHasher() |
| 37 | + _, _ = h.Write([]byte(paths.Topic)) |
| 38 | + copy(topicAddr[:], h.Sum(nil)) |
| 39 | + } |
| 40 | + |
| 41 | + // Required header: underlay multiaddr |
| 42 | + peerHeader := r.Header.Get(SwarmPubsubPeerHeader) |
| 43 | + if peerHeader == "" { |
| 44 | + jsonhttp.BadRequest(w, "missing Swarm-Pubsub-Peer header") |
| 45 | + return |
| 46 | + } |
| 47 | + underlay, err := ma.NewMultiaddr(peerHeader) |
| 48 | + if err != nil { |
| 49 | + logger.Debug("invalid peer multiaddr", "value", peerHeader, "error", err) |
| 50 | + jsonhttp.BadRequest(w, "invalid Swarm-Pubsub-Peer header") |
| 51 | + return |
| 52 | + } |
| 53 | + |
| 54 | + // Optional headers: GSOC fields for Participant upgrade |
| 55 | + var connectOpts pubsub.ConnectOptions |
| 56 | + |
| 57 | + gsocPubKeyHex := r.Header.Get(SwarmPubsubGsocPublicKeyHeader) |
| 58 | + gsocTopicHex := r.Header.Get(SwarmPubsubGsocTopicHeader) |
| 59 | + if gsocPubKeyHex != "" && gsocTopicHex != "" { |
| 60 | + gsocOwner, err := hex.DecodeString(gsocPubKeyHex) |
| 61 | + if err != nil { |
| 62 | + jsonhttp.BadRequest(w, "invalid Swarm-Pubsub-Gsoc-Public-Key header") |
| 63 | + return |
| 64 | + } |
| 65 | + gsocID, err := hex.DecodeString(gsocTopicHex) |
| 66 | + if err != nil { |
| 67 | + jsonhttp.BadRequest(w, "invalid Swarm-Pubsub-Gsoc-Topic header") |
| 68 | + return |
| 69 | + } |
| 70 | + connectOpts.GsocOwner = gsocOwner |
| 71 | + connectOpts.GsocID = gsocID |
| 72 | + connectOpts.ReadWrite = true |
| 73 | + } |
| 74 | + |
| 75 | + headers := struct { |
| 76 | + KeepAlive time.Duration `map:"Swarm-Keep-Alive"` |
| 77 | + }{} |
| 78 | + if response := s.mapStructure(r.Header, &headers); response != nil { |
| 79 | + response("invalid header params", logger, w) |
| 80 | + return |
| 81 | + } |
| 82 | + |
| 83 | + if s.beeMode == DevMode { |
| 84 | + logger.Warning("pubsub endpoint is disabled in dev mode") |
| 85 | + jsonhttp.BadRequest(w, errUnsupportedDevNodeOperation) |
| 86 | + return |
| 87 | + } |
| 88 | + |
| 89 | + // Connect to broker peer |
| 90 | + ctx, cancel := context.WithCancel(context.Background()) |
| 91 | + subscriberConn, err := s.pubsubSvc.Connect(ctx, underlay, topicAddr, pubsub.ModeGSOCEphemeral, connectOpts) |
| 92 | + if err != nil { |
| 93 | + cancel() |
| 94 | + logger.Debug("pubsub connect failed", "error", err) |
| 95 | + jsonhttp.InternalServerError(w, "pubsub connect failed") |
| 96 | + return |
| 97 | + } |
| 98 | + |
| 99 | + // Upgrade to WebSocket |
| 100 | + upgrader := websocket.Upgrader{ |
| 101 | + ReadBufferSize: swarm.ChunkWithSpanSize, |
| 102 | + WriteBufferSize: swarm.ChunkWithSpanSize, |
| 103 | + CheckOrigin: s.checkOrigin, |
| 104 | + } |
| 105 | + |
| 106 | + conn, err := upgrader.Upgrade(w, r, nil) |
| 107 | + if err != nil { |
| 108 | + cancel() |
| 109 | + _ = subscriberConn.Stream.Close() |
| 110 | + logger.Debug("websocket upgrade failed", "error", err) |
| 111 | + logger.Error(nil, "websocket upgrade failed") |
| 112 | + jsonhttp.InternalServerError(w, "upgrade failed") |
| 113 | + return |
| 114 | + } |
| 115 | + |
| 116 | + pingPeriod := headers.KeepAlive * time.Second |
| 117 | + if pingPeriod == 0 { |
| 118 | + pingPeriod = time.Minute |
| 119 | + } |
| 120 | + |
| 121 | + isParticipant := connectOpts.ReadWrite |
| 122 | + |
| 123 | + s.wsWg.Add(1) |
| 124 | + go func() { |
| 125 | + pubsub.ListeningWs(ctx, conn, pubsub.WsOptions{PingPeriod: pingPeriod, Cancel: cancel}, logger, subscriberConn, isParticipant) |
| 126 | + _ = conn.Close() |
| 127 | + subscriberConn.Cancel() |
| 128 | + s.wsWg.Done() |
| 129 | + }() |
| 130 | +} |
| 131 | + |
| 132 | +func (s *Service) pubsubListHandler(w http.ResponseWriter, r *http.Request) { |
| 133 | + if s.pubsubSvc == nil { |
| 134 | + jsonhttp.NotFound(w, "pubsub service not available") |
| 135 | + return |
| 136 | + } |
| 137 | + |
| 138 | + topics := s.pubsubSvc.Topics() |
| 139 | + jsonhttp.OK(w, struct { |
| 140 | + Topics []pubsub.TopicInfo `json:"topics"` |
| 141 | + }{ |
| 142 | + Topics: topics, |
| 143 | + }) |
| 144 | +} |
0 commit comments