feat: pubsub

cmd/bee/cmd/cmd.go

@@ -91,6 +91,7 @@ const (
 	optionAutoTLSRegistrationEndpoint      = "autotls-registration-endpoint"
 	optionAutoTLSCAEndpoint                = "autotls-ca-endpoint"
 	optionUseSIMD                          = "use-simd-hashing"
+	optionNamePubsubBrokerMode             = "pubsub-broker-mode"
 
 	// blockchain-rpc
 	optionNameBlockchainRpcEndpoint    = "blockchain-rpc-endpoint"
@@ -339,6 +340,7 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
 	cmd.Flags().String(optionAutoTLSDomain, p2pforge.DefaultForgeDomain, "autotls domain")
 	cmd.Flags().String(optionAutoTLSRegistrationEndpoint, p2pforge.DefaultForgeEndpoint, "autotls registration endpoint")
 	cmd.Flags().String(optionAutoTLSCAEndpoint, p2pforge.DefaultCAEndpoint, "autotls certificate authority endpoint")
+	cmd.Flags().Bool(optionNamePubsubBrokerMode, true, "enable pubsub broker mode")
 	cmd.Flags().Bool(optionUseSIMD, false, "use SIMD BMT hasher (available only on linux amd64 platforms)")
 }
 

cmd/bee/cmd/start.go

@@ -353,6 +353,7 @@ func buildBeeNode(ctx context.Context, c *command, cmd *cobra.Command, logger lo
 		WarmupTime:                    c.config.GetDuration(optionWarmUpTime),
 		WelcomeMessage:                c.config.GetString(optionWelcomeMessage),
 		WhitelistedWithdrawalAddress:  c.config.GetStringSlice(optionNameWhitelistedWithdrawalAddress),
+		PubsubBrokerMode:              c.config.GetBool(optionNamePubsubBrokerMode),
 	})
 
 	return b, err

openapi/Swarm.yaml

@@ -2541,3 +2541,57 @@ paths:
           $ref: "SwarmCommon.yaml#/components/responses/400"
         default:
           description: Default response.
+
+  "/pubsub/{topic}":
+    get:
+      summary: Connect to a pubsub topic via WebSocket
+      description: |
+        Opens a WebSocket connection to a pubsub topic. The connection acts as either a publisher (read+write)
+        or subscriber (read-only) depending on the presence of GSOC query params.
+
+        **WebSocket protocol:**
+        - Inbound (client → node, publisher only): raw SOC payload `[sig:65B][span:8B][payload:N B]`
+        - Outbound (node → client): raw SOC payload `[sig:65B][span:8B][payload:N B]`
+      tags:
+        - Pubsub
+      parameters:
+        - in: path
+          name: topic
+          schema:
+            type: string
+          required: true
+          description: Topic identifier (hex-encoded address or arbitrary string to be hashed)
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPubsubPeer"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPubsubGsocEthAddress"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmPubsubGsocTopic"
+        - in: header
+          name: swarm-keep-alive
+          schema:
+            type: integer
+          required: false
+          description: "WebSocket ping period in seconds (default: 60)"
+      responses:
+        "101":
+          description: WebSocket upgrade successful
+        "400":
+          $ref: "SwarmCommon.yaml#/components/responses/400"
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"
+
+  "/pubsub/":
+    get:
+      summary: List all pubsub topics
+      description: Returns a list of all active pubsub topics this node is participating in (as broker or subscriber)
+      tags:
+        - Pubsub
+      responses:
+        "200":
+          description: List of pubsub topics
+          content:
+            application/json:
+              schema:
+                $ref: "SwarmCommon.yaml#/components/schemas/PubsubTopicListResponse"
+        "400":
+          $ref: "SwarmCommon.yaml#/components/responses/400"
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"

openapi/SwarmCommon.yaml

@@ -1075,6 +1075,33 @@ components:
       properties:
         transactionHash:
           $ref: "#/components/schemas/TransactionHash"
+
+    PubsubTopicInfo:
+      type: object
+      properties:
+        topicAddress:
+          type: string
+          description: "Hex-encoded topic address"
+        mode:
+          type: integer
+          description: "Pubsub mode identifier"
+        role:
+          type: string
+          description: "Role of this node: 'broker' or 'subscriber'"
+        connections:
+          type: array
+          items:
+            type: string
+          description: "List of connected peer overlays"
+
+    PubsubTopicListResponse:
+      type: object
+      properties:
+        topics:
+          type: array
+          items:
+            $ref: "#/components/schemas/PubsubTopicInfo"
+
   headers:
     SwarmTag:
       description: "Tag UID"
@@ -1118,7 +1145,6 @@ components:
       required: false
       description: "Indicates which feed version was resolved (v1 or v2)"
 
-
   parameters:
     GasPriceParameter:
       in: header
@@ -1331,6 +1357,30 @@ components:
       required: false
       description: "ACT history Unix timestamp"
 
+    SwarmPubsubPeer:
+      in: query
+      name: peer
+      schema:
+        type: string
+      required: true
+      description: "Multiaddress of the broker peer to connect to for pubsub"
+
+    SwarmPubsubGsocEthAddress:
+      in: query
+      name: gsoc-eth-address
+      schema:
+        $ref: "#/components/schemas/HexString"
+      required: false
+      description: "GSOC owner Ethereum address (20 bytes, hex-encoded) for publisher role. Required together with gsoc-topic to upgrade to publisher."
+
+    SwarmPubsubGsocTopic:
+      in: query
+      name: gsoc-topic
+      schema:
+        $ref: "#/components/schemas/HexString"
+      required: false
+      description: "GSOC topic identifier (hex) for publisher role. Required together with gsoc-eth-address to upgrade to publisher."
+
   responses:
     "200":
       description: Success

pkg/api/api.go

@@ -40,6 +40,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/postage"
 	"github.com/ethersphere/bee/v2/pkg/postage/postagecontract"
 	"github.com/ethersphere/bee/v2/pkg/pss"
+	"github.com/ethersphere/bee/v2/pkg/pubsub"
 	"github.com/ethersphere/bee/v2/pkg/resolver"
 	"github.com/ethersphere/bee/v2/pkg/resolver/client/ens"
 	"github.com/ethersphere/bee/v2/pkg/resolver/multiresolver"
@@ -93,11 +94,10 @@ const (
 	SwarmActTimestampHeader           = "Swarm-Act-Timestamp"
 	SwarmActPublisherHeader           = "Swarm-Act-Publisher"
 	SwarmActHistoryAddressHeader      = "Swarm-Act-History-Address"
-
-	ImmutableHeader = "Immutable"
-	GasPriceHeader  = "Gas-Price"
-	GasLimitHeader  = "Gas-Limit"
-	ETagHeader      = "ETag"
+	ImmutableHeader                   = "Immutable"
+	GasPriceHeader                    = "Gas-Price"
+	GasLimitHeader                    = "Gas-Limit"
+	ETagHeader                        = "ETag"
 
 	AuthorizationHeader        = "Authorization"
 	AcceptEncodingHeader       = "Accept-Encoding"
@@ -185,6 +185,7 @@ type Service struct {
 
 	topologyDriver topology.Driver
 	p2p            p2p.DebugService
+	pubsubSvc      *pubsub.Service
 	accounting     accounting.Interface
 	chequebook     chequebook.Service
 	pseudosettle   settlement.Interface
@@ -268,6 +269,7 @@ type ExtraOptions struct {
 	SyncStatus      func() (bool, error)
 	NodeStatus      *status.Service
 	PinIntegrity    PinIntegrity
+	PubsubService   *pubsub.Service
 }
 
 func New(
@@ -355,6 +357,7 @@ func (s *Service) Configure(signer crypto.Signer, tracer *tracing.Tracer, o Opti
 	s.lightNodes = e.LightNodes
 	s.pseudosettle = e.Pseudosettle
 	s.blockTime = e.BlockTime
+	s.pubsubSvc = e.PubsubService
 
 	s.statusSem = semaphore.NewWeighted(1)
 	s.postageSem = semaphore.NewWeighted(1)

pkg/api/api_test.go

@@ -42,6 +42,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/postage/postagecontract"
 	contractMock "github.com/ethersphere/bee/v2/pkg/postage/postagecontract/mock"
 	"github.com/ethersphere/bee/v2/pkg/pss"
+	"github.com/ethersphere/bee/v2/pkg/pubsub"
 	"github.com/ethersphere/bee/v2/pkg/pusher"
 	"github.com/ethersphere/bee/v2/pkg/resolver"
 	resolverMock "github.com/ethersphere/bee/v2/pkg/resolver/mock"
@@ -136,6 +137,7 @@ type testServerOptions struct {
 	ChequebookDisabled  bool
 	SwapDisabled        bool
 	Erc20ServiceNil     bool
+	PubsubService       *pubsub.Service
 }
 
 func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.Conn, string, *chanStorer) {
@@ -210,6 +212,7 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 		Staking:         o.StakingContract,
 		NodeStatus:      o.NodeStatus,
 		PinIntegrity:    o.PinIntegrity,
+		PubsubService:   o.PubsubService,
 	}
 
 	// By default bee mode is set to full mode.

pkg/api/pubsub.go

@@ -0,0 +1,137 @@
+// Copyright 2026 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package api
+
+import (
+	"context"
+	"encoding/hex"
+	"net/http"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/jsonhttp"
+	"github.com/ethersphere/bee/v2/pkg/pubsub"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+	"github.com/gorilla/mux"
+	"github.com/gorilla/websocket"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+func (s *Service) pubsubWsHandler(w http.ResponseWriter, r *http.Request) {
+	logger := s.logger.WithName("pubsub").Build()
+
+	paths := struct {
+		Topic string `map:"topic" validate:"required"`
+	}{}
+	if response := s.mapStructure(mux.Vars(r), &paths); response != nil {
+		response("invalid path params", logger, w)
+		return
+	}
+
+	var topicAddr [32]byte
+	if decoded, err := hex.DecodeString(paths.Topic); err == nil && len(decoded) == swarm.HashSize {
+		copy(topicAddr[:], decoded)
+	} else {
+		h := swarm.NewHasher()
+		_, _ = h.Write([]byte(paths.Topic))
+		copy(topicAddr[:], h.Sum(nil))
+	}
+
+	peerMultiaddr := r.URL.Query().Get("peer")
+	if peerMultiaddr == "" {
+		jsonhttp.BadRequest(w, "missing peer query param")
+		return
+	}
+	underlay, err := ma.NewMultiaddr(peerMultiaddr)
+	if err != nil {
+		logger.Info("invalid peer multiaddr", "value", peerMultiaddr, "error", err)
+		jsonhttp.BadRequest(w, "invalid peer query param")
+		return
+	}
+
+	var connectOpts pubsub.ConnectOptions
+
+	gsocEthAddrHex := r.URL.Query().Get("gsoc-eth-address")
+	gsocTopicHex := r.URL.Query().Get("gsoc-topic")
+	if gsocEthAddrHex != "" && gsocTopicHex != "" {
+		gsocOwner, err := hex.DecodeString(gsocEthAddrHex)
+		if err != nil {
+			jsonhttp.BadRequest(w, "invalid gsoc-eth-address query param")
+			return
+		}
+		gsocID, err := hex.DecodeString(gsocTopicHex)
+		if err != nil {
+			jsonhttp.BadRequest(w, "invalid gsoc-topic query param")
+			return
+		}
+		connectOpts.GsocOwner = gsocOwner
+		connectOpts.GsocID = gsocID
+		connectOpts.ReadWrite = true
+	}
+
+	headers := struct {
+		KeepAlive int `map:"Swarm-Keep-Alive"`
+	}{}
+	if response := s.mapStructure(r.Header, &headers); response != nil {
+		response("invalid header params", logger, w)
+		return
+	}
+
+	// Connect to broker peer
+	ctx, cancel := context.WithCancel(context.Background())
+	mode, err := s.pubsubSvc.Connect(ctx, underlay, topicAddr, pubsub.ModeGSOCEphemeral, connectOpts)
+	if err != nil {
+		cancel()
+		logger.Info("pubsub connect failed", "error", err)
+		jsonhttp.InternalServerError(w, "pubsub connect failed")
+		return
+	}
+	// Upgrade to WebSocket
+	upgrader := websocket.Upgrader{
+		ReadBufferSize:  swarm.ChunkWithSpanSize,
+		WriteBufferSize: swarm.ChunkWithSpanSize,
+		CheckOrigin:     s.checkOrigin,
+	}
+
+	logger.Info("upgrading to websocket")
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		// Upgrade() hijacks the connection before returning an error,
+		// so do NOT write an HTTP response here.
+		cancel()
+		logger.Info("websocket upgrade failed", "error", err)
+		logger.Error(nil, "websocket upgrade failed")
+		return
+	}
+	logger.Info("websocket upgrade successful")
+
+	pingPeriod := time.Duration(headers.KeepAlive) * time.Second
+	if pingPeriod == 0 {
+		pingPeriod = time.Minute
+	}
+
+	isPublisher := connectOpts.ReadWrite
+
+	s.wsWg.Add(1)
+	go func() {
+		pubsub.ListeningWs(ctx, conn, pubsub.WsOptions{PingPeriod: pingPeriod, Cancel: cancel}, logger, mode, isPublisher)
+		cancel()
+		_ = conn.Close()
+		s.wsWg.Done()
+	}()
+}
+
+func (s *Service) pubsubListHandler(w http.ResponseWriter, r *http.Request) {
+	if s.pubsubSvc == nil {
+		jsonhttp.NotFound(w, "pubsub service not available")
+		return
+	}
+
+	topics := s.pubsubSvc.Topics()
+	jsonhttp.OK(w, struct {
+		Topics []pubsub.TopicInfo `json:"topics"`
+	}{
+		Topics: topics,
+	})
+}