feat: mic moc

openapi/Swarm.yaml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 
 info:
-  version: 8.1.0
+  version: 8.2.0
   title: Bee API
   description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management"
 
@@ -846,6 +846,48 @@ paths:
         default:
           description: Default response
 
+  "/moc/subscribe/{id}":
+    get:
+      summary: Subscribe to MOC payloads
+      description: "Subscribe to Single Owner Chunks by their identifier, regardless of owner (Mined Owner Chunk)."
+      tags:
+        - MOC
+      parameters:
+        - in: path
+          name: id
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/HexString"
+          required: true
+          description: "Single Owner Chunk identifier (32 bytes, hex encoded)"
+      responses:
+        "200":
+          description: Establishes a WebSocket subscription for incoming messages on the Single Owner Chunk identifier
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"
+        default:
+          description: Default response
+
+  "/mic/subscribe/{owner}":
+    get:
+      summary: Subscribe to MIC payloads
+      description: "Subscribe to Single Owner Chunks by their owner ethereum address, regardless of identifier (Mined ID Chunk)."
+      tags:
+        - MIC
+      parameters:
+        - in: path
+          name: owner
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/EthereumAddress"
+          required: true
+          description: "Single Owner Chunk owner ethereum address (20 bytes, hex encoded)"
+      responses:
+        "200":
+          description: Establishes a WebSocket subscription for incoming messages on the Single Owner Chunk owner
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"
+        default:
+          description: Default response
+
   "/soc/{owner}/{id}":
     post:
       summary: Upload a Single Owner Chunk

pkg/api/api.go

@@ -35,6 +35,8 @@ import (
 	"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/mic"
+	"github.com/ethersphere/bee/v2/pkg/moc"
 	"github.com/ethersphere/bee/v2/pkg/p2p"
 	"github.com/ethersphere/bee/v2/pkg/pingpong"
 	"github.com/ethersphere/bee/v2/pkg/postage"
@@ -153,6 +155,8 @@ type Service struct {
 	resolver        resolver.Interface
 	pss             pss.Interface
 	gsoc            gsoc.Listener
+	moc             moc.Listener
+	mic             mic.Listener
 	steward         steward.Interface
 	logger          log.Logger
 	loggerV1        log.Logger
@@ -259,6 +263,8 @@ type ExtraOptions struct {
 	Resolver        resolver.Interface
 	Pss             pss.Interface
 	Gsoc            gsoc.Listener
+	Moc             moc.Listener
+	Mic             mic.Listener
 	FeedFactory     feeds.Factory
 	Post            postage.Service
 	AccessControl   accesscontrol.Controller
@@ -340,6 +346,8 @@ func (s *Service) Configure(signer crypto.Signer, tracer *tracing.Tracer, o Opti
 	s.resolver = e.Resolver
 	s.pss = e.Pss
 	s.gsoc = e.Gsoc
+	s.moc = e.Moc
+	s.mic = e.Mic
 	s.feedFactory = e.FeedFactory
 	s.post = e.Post
 	s.accesscontrol = e.AccessControl

pkg/api/api_test.go

@@ -34,6 +34,8 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/gsoc"
 	"github.com/ethersphere/bee/v2/pkg/jsonhttp/jsonhttptest"
 	"github.com/ethersphere/bee/v2/pkg/log"
+	"github.com/ethersphere/bee/v2/pkg/mic"
+	"github.com/ethersphere/bee/v2/pkg/moc"
 	p2pmock "github.com/ethersphere/bee/v2/pkg/p2p/mock"
 	"github.com/ethersphere/bee/v2/pkg/pingpong"
 	"github.com/ethersphere/bee/v2/pkg/postage"
@@ -94,6 +96,8 @@ type testServerOptions struct {
 	Resolver           resolver.Interface
 	Pss                pss.Interface
 	Gsoc               gsoc.Listener
+	Moc                moc.Listener
+	Mic                mic.Listener
 	WsPath             string
 	WsPingPeriod       time.Duration
 	Logger             log.Logger
@@ -201,6 +205,8 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 		Resolver:        o.Resolver,
 		Pss:             o.Pss,
 		Gsoc:            o.Gsoc,
+		Moc:             o.Moc,
+		Mic:             o.Mic,
 		FeedFactory:     o.Feeds,
 		Post:            o.Post,
 		AccessControl:   o.AccessControl,

pkg/api/mic.go

@@ -0,0 +1,35 @@
+// 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 (
+	"net/http"
+
+	"github.com/ethersphere/bee/v2/pkg/mic"
+	"github.com/gorilla/mux"
+)
+
+func (s *Service) micWsHandler(w http.ResponseWriter, r *http.Request) {
+	logger := s.logger.WithName("mic_subscribe").Build()
+
+	paths := struct {
+		Owner []byte `map:"owner" validate:"required"`
+	}{}
+
+	if response := s.mapStructure(mux.Vars(r), &paths); response != nil {
+		response("invalid path params", logger, w)
+		return
+	}
+
+	conn, ok := s.wsUpgrade(w, r, logger)
+	if !ok {
+		return
+	}
+
+	s.wsWg.Add(1)
+	go s.socSubscribeWs("mic", conn, func(handler func([]byte)) func() {
+		return s.mic.Subscribe(paths.Owner, mic.Handler(handler))
+	})
+}

pkg/api/mic_test.go

@@ -0,0 +1,87 @@
+// 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_test
+
+import (
+	"encoding/hex"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/cac"
+	"github.com/ethersphere/bee/v2/pkg/crypto"
+	"github.com/ethersphere/bee/v2/pkg/log"
+	"github.com/ethersphere/bee/v2/pkg/mic"
+	mockbatchstore "github.com/ethersphere/bee/v2/pkg/postage/batchstore/mock"
+	"github.com/ethersphere/bee/v2/pkg/soc"
+	mockstorer "github.com/ethersphere/bee/v2/pkg/storer/mock"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+	"github.com/ethersphere/bee/v2/pkg/util/testutil"
+	"github.com/gorilla/websocket"
+)
+
+// TestMicWebsocketSingleHandler subscribes on a SOC owner and receives a message.
+func TestMicWebsocketSingleHandler(t *testing.T) {
+	t.Parallel()
+
+	var (
+		id            = make([]byte, 32)
+		m, cl, signer = newMicTest(t, 0)
+		respC         = make(chan error, 1)
+		payload       = []byte("hello there!")
+	)
+
+	err := cl.SetReadDeadline(time.Now().Add(longTimeout))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cl.SetReadLimit(swarm.ChunkSize)
+
+	ch, _ := cac.New(payload)
+	socCh := soc.New(id, ch)
+	ch, _ = socCh.Sign(signer)
+	socCh, _ = soc.FromChunk(ch)
+	m.Handle(socCh)
+
+	go expectMessage(t, cl, respC, payload)
+	if err := <-respC; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func newMicTest(t *testing.T, pingPeriod time.Duration) (mic.Listener, *websocket.Conn, crypto.Signer) {
+	t.Helper()
+	if pingPeriod == 0 {
+		pingPeriod = 10 * time.Second
+	}
+	var (
+		batchStore = mockbatchstore.New()
+		storer     = mockstorer.New()
+	)
+
+	privKey, err := crypto.GenerateSecp256k1Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	signer := crypto.NewDefaultSigner(privKey)
+	owner, err := signer.EthereumAddress()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	micService := mic.New(log.NewLogger("test"))
+	testutil.CleanupCloser(t, micService)
+
+	_, cl, _, _ := newTestServer(t, testServerOptions{
+		Mic:          micService,
+		WsPath:       fmt.Sprintf("/mic/subscribe/%s", hex.EncodeToString(owner.Bytes())),
+		Storer:       storer,
+		BatchStore:   batchStore,
+		Logger:       log.Noop,
+		WsPingPeriod: pingPeriod,
+	})
+
+	return micService, cl, signer
+}

pkg/api/moc.go

@@ -0,0 +1,35 @@
+// 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 (
+	"net/http"
+
+	"github.com/ethersphere/bee/v2/pkg/moc"
+	"github.com/gorilla/mux"
+)
+
+func (s *Service) mocWsHandler(w http.ResponseWriter, r *http.Request) {
+	logger := s.logger.WithName("moc_subscribe").Build()
+
+	paths := struct {
+		ID []byte `map:"id" validate:"required"`
+	}{}
+
+	if response := s.mapStructure(mux.Vars(r), &paths); response != nil {
+		response("invalid path params", logger, w)
+		return
+	}
+
+	conn, ok := s.wsUpgrade(w, r, logger)
+	if !ok {
+		return
+	}
+
+	s.wsWg.Add(1)
+	go s.socSubscribeWs("moc", conn, func(handler func([]byte)) func() {
+		return s.moc.Subscribe(paths.ID, moc.Handler(handler))
+	})
+}

pkg/api/moc_test.go

@@ -0,0 +1,83 @@
+// 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_test
+
+import (
+	"encoding/hex"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/ethersphere/bee/v2/pkg/cac"
+	"github.com/ethersphere/bee/v2/pkg/crypto"
+	"github.com/ethersphere/bee/v2/pkg/log"
+	"github.com/ethersphere/bee/v2/pkg/moc"
+	mockbatchstore "github.com/ethersphere/bee/v2/pkg/postage/batchstore/mock"
+	"github.com/ethersphere/bee/v2/pkg/soc"
+	mockstorer "github.com/ethersphere/bee/v2/pkg/storer/mock"
+	"github.com/ethersphere/bee/v2/pkg/swarm"
+	"github.com/ethersphere/bee/v2/pkg/util/testutil"
+	"github.com/gorilla/websocket"
+)
+
+// TestMocWebsocketSingleHandler subscribes on a SOC id and receives a message.
+func TestMocWebsocketSingleHandler(t *testing.T) {
+	t.Parallel()
+
+	var (
+		id            = make([]byte, 32)
+		m, cl, signer = newMocTest(t, id, 0)
+		respC         = make(chan error, 1)
+		payload       = []byte("hello there!")
+	)
+
+	err := cl.SetReadDeadline(time.Now().Add(longTimeout))
+	if err != nil {
+		t.Fatal(err)
+	}
+	cl.SetReadLimit(swarm.ChunkSize)
+
+	ch, _ := cac.New(payload)
+	socCh := soc.New(id, ch)
+	ch, _ = socCh.Sign(signer)
+	socCh, _ = soc.FromChunk(ch)
+	m.Handle(socCh)
+
+	go expectMessage(t, cl, respC, payload)
+	if err := <-respC; err != nil {
+		t.Fatal(err)
+	}
+}
+
+func newMocTest(t *testing.T, socId []byte, pingPeriod time.Duration) (moc.Listener, *websocket.Conn, crypto.Signer) {
+	t.Helper()
+	if pingPeriod == 0 {
+		pingPeriod = 10 * time.Second
+	}
+	var (
+		batchStore = mockbatchstore.New()
+		storer     = mockstorer.New()
+	)
+
+	privKey, err := crypto.GenerateSecp256k1Key()
+	if err != nil {
+		t.Fatal(err)
+	}
+	signer := crypto.NewDefaultSigner(privKey)
+
+	mocService := moc.New(log.NewLogger("test"))
+	testutil.CleanupCloser(t, mocService)
+
+	_, cl, _, _ := newTestServer(t, testServerOptions{
+		Moc:          mocService,
+		WsPath:       fmt.Sprintf("/moc/subscribe/%s", hex.EncodeToString(socId)),
+		Storer:       storer,
+		BatchStore:   batchStore,
+		Logger:       log.Noop,
+		WsPingPeriod: pingPeriod,
+	})
+
+	return mocService, cl, signer
+}

pkg/api/router.go

@@ -376,6 +376,18 @@ func (s *Service) mountAPI() {
 		),
 	})
 
+	handle("/moc/subscribe/{id}", jsonhttp.MethodHandler{
+		"GET": web.ChainHandlers(
+			web.FinalHandlerFunc(s.mocWsHandler),
+		),
+	})
+
+	handle("/mic/subscribe/{owner}", jsonhttp.MethodHandler{
+		"GET": web.ChainHandlers(
+			web.FinalHandlerFunc(s.micWsHandler),
+		),
+	})
+
 	handle("/tags", jsonhttp.MethodHandler{
 		"GET": http.HandlerFunc(s.listTagsHandler),
 		"POST": web.ChainHandlers(