feat: gsoc subscribe

openapi/Swarm.yaml

@@ -818,6 +818,28 @@ paths:
         default:
           description: Default response
 
+  "/gsoc/subscribe/{address}":
+    get:
+      summary: Subscribe to GSOC payloads
+      tags:
+        - GSOC
+        - Subscribe
+        - Websocket
+      parameters:
+        - in: path
+          name: reference
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/SwarmReference"
+          required: true
+          description: "Single Owner Chunk address (which may have multiple payloads)"
+      responses:
+        "200":
+          description: Returns a WebSocket with a subscription for incoming message data on the requested SOC address.
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"
+        default:
+          description: Default response
+
   "/soc/{owner}/{id}":
     post:
       summary: Upload single owner chunk

pkg/api/api.go

@@ -33,6 +33,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/file/pipeline"
 	"github.com/ethersphere/bee/v2/pkg/file/pipeline/builder"
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy"
+	"github.com/ethersphere/bee/v2/pkg/gsoc"
 	"github.com/ethersphere/bee/v2/pkg/jsonhttp"
 	"github.com/ethersphere/bee/v2/pkg/log"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
@@ -151,6 +152,7 @@ type Service struct {
 	storer          Storer
 	resolver        resolver.Interface
 	pss             pss.Interface
+	gsoc            gsoc.Listener
 	steward         steward.Interface
 	logger          log.Logger
 	loggerV1        log.Logger
@@ -254,6 +256,7 @@ type ExtraOptions struct {
 	Storer          Storer
 	Resolver        resolver.Interface
 	Pss             pss.Interface
+	Gsoc            gsoc.Listener
 	FeedFactory     feeds.Factory
 	Post            postage.Service
 	AccessControl   accesscontrol.Controller
@@ -337,6 +340,7 @@ func (s *Service) Configure(signer crypto.Signer, tracer *tracing.Tracer, o Opti
 	s.storer = e.Storer
 	s.resolver = e.Resolver
 	s.pss = e.Pss
+	s.gsoc = e.Gsoc
 	s.feedFactory = e.FeedFactory
 	s.post = e.Post
 	s.accesscontrol = e.AccessControl
@@ -682,7 +686,12 @@ type putterSessionWrapper struct {
 }
 
 func (p *putterSessionWrapper) Put(ctx context.Context, chunk swarm.Chunk) error {
-	stamp, err := p.stamper.Stamp(chunk.Address())
+	idAddress, err := storage.IdentityAddress(chunk)
+	if err != nil {
+		return err
+	}
+
+	stamp, err := p.stamper.Stamp(chunk.Address(), idAddress)
 	if err != nil {
 		return err
 	}

pkg/api/api_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/file/pipeline"
 	"github.com/ethersphere/bee/v2/pkg/file/pipeline/builder"
 	"github.com/ethersphere/bee/v2/pkg/file/redundancy"
+	"github.com/ethersphere/bee/v2/pkg/gsoc"
 	"github.com/ethersphere/bee/v2/pkg/jsonhttp/jsonhttptest"
 	"github.com/ethersphere/bee/v2/pkg/log"
 	p2pmock "github.com/ethersphere/bee/v2/pkg/p2p/mock"
@@ -93,6 +94,7 @@ type testServerOptions struct {
 	StateStorer        storage.StateStorer
 	Resolver           resolver.Interface
 	Pss                pss.Interface
+	Gsoc               gsoc.Listener
 	WsPath             string
 	WsPingPeriod       time.Duration
 	Logger             log.Logger
@@ -194,6 +196,7 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 		Storer:          o.Storer,
 		Resolver:        o.Resolver,
 		Pss:             o.Pss,
+		Gsoc:            o.Gsoc,
 		FeedFactory:     o.Feeds,
 		Post:            o.Post,
 		AccessControl:   o.AccessControl,

pkg/api/envelope.go

@@ -59,7 +59,7 @@ func (s *Service) envelopePostHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	stamp, err := stamper.Stamp(paths.Address)
+	stamp, err := stamper.Stamp(paths.Address, paths.Address)
 	if err != nil {
 		logger.Debug("split write all failed", "error", err)
 		logger.Error(nil, "split write all failed")

pkg/api/gsoc.go

@@ -0,0 +1,120 @@
+// Copyright 2024 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 (
+	"net/http"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/jsonhttp"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+	"github.com/gorilla/mux"
+	"github.com/gorilla/websocket"
+)
+
+func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) {
+	logger := s.logger.WithName("gsoc_subscribe").Build()
+
+	paths := struct {
+		Address swarm.Address `map:"address,resolve" validate:"required"`
+	}{}
+
+	if response := s.mapStructure(mux.Vars(r), &paths); response != nil {
+		response("invalid path params", logger, w)
+		return
+	}
+
+	upgrader := websocket.Upgrader{
+		ReadBufferSize:  swarm.ChunkSize,
+		WriteBufferSize: swarm.ChunkSize,
+		CheckOrigin:     s.checkOrigin,
+	}
+
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		logger.Debug("upgrade failed", "error", err)
+		logger.Error(nil, "upgrade failed")
+		jsonhttp.InternalServerError(w, "upgrade failed")
+		return
+	}
+
+	s.wsWg.Add(1)
+	go s.gsocListeningWs(conn, paths.Address)
+}
+
+func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address) {
+	defer s.wsWg.Done()
+
+	var (
+		dataC  = make(chan []byte)
+		gone   = make(chan struct{})
+		ticker = time.NewTicker(s.WsPingPeriod)
+		err    error
+	)
+	defer func() {
+		ticker.Stop()
+		_ = conn.Close()
+	}()
+	cleanup := s.gsoc.Subscribe(socAddress, func(m []byte) {
+		select {
+		case dataC <- m:
+		case <-gone:
+			return
+		case <-s.quit:
+			return
+		}
+	})
+
+	defer cleanup()
+
+	conn.SetCloseHandler(func(code int, text string) error {
+		s.logger.Debug("gsoc ws: client gone", "code", code, "message", text)
+		close(gone)
+		return nil
+	})
+
+	for {
+		select {
+		case b := <-dataC:
+			err = conn.SetWriteDeadline(time.Now().Add(writeDeadline))
+			if err != nil {
+				s.logger.Debug("gsoc ws: set write deadline failed", "error", err)
+				return
+			}
+
+			err = conn.WriteMessage(websocket.BinaryMessage, b)
+			if err != nil {
+				s.logger.Debug("gsoc ws: write message failed", "error", err)
+				return
+			}
+
+		case <-s.quit:
+			// shutdown
+			err = conn.SetWriteDeadline(time.Now().Add(writeDeadline))
+			if err != nil {
+				s.logger.Debug("gsoc ws: set write deadline failed", "error", err)
+				return
+			}
+			err = conn.WriteMessage(websocket.CloseMessage, []byte{})
+			if err != nil {
+				s.logger.Debug("gsoc ws: write close message failed", "error", err)
+			}
+			return
+		case <-gone:
+			// client gone
+			return
+		case <-ticker.C:
+			err = conn.SetWriteDeadline(time.Now().Add(writeDeadline))
+			if err != nil {
+				s.logger.Debug("gsoc ws: set write deadline failed", "error", err)
+				return
+			}
+			if err = conn.WriteMessage(websocket.PingMessage, nil); err != nil {
+				// error encountered while pinging client. client probably gone
+				return
+			}
+		}
+	}
+}