-- | GossipSub Switch integration handler (Phase 10b).
--
-- Bridges the GossipSub Router with the Switch by:
-- 1. Registering a StreamHandler for inbound /meshsub/1.1.0 streams
-- 2. Providing a sendRPC callback that opens/reuses outbound streams
-- 3. Managing lifecycle (heartbeat start/stop)
--
-- GossipSub maintains persistent bidirectional RPC streams, unlike
-- Identify/Ping which are one-shot. Each peer has at most one cached
-- outbound stream.
module LibP2P.Protocol.GossipSub.Handler
  ( -- * Types
    GossipSubNode (..)
    -- * Construction
  , newGossipSubNode
    -- * Stream handling
  , handleGossipSubStream
  , sendCurrentSubscriptions
    -- * Lifecycle
  , startGossipSub
  , stopGossipSub
    -- * Convenience API
  , gossipJoin
  , gossipLeave
  , gossipPublish
    -- * Constants
  , gossipSubProtocolId
  , gossipSubProtocolIdV10
  , floodSubProtocolId
  ) where

import Control.Concurrent.Async (Async, async, cancel)
import Control.Concurrent.STM
  ( TVar
  , atomically
  , newTVarIO
  , readTVar
  , writeTVar
  , modifyTVar'
  )
import Control.Exception (SomeException, catch)
import Data.ByteString (ByteString)
import qualified Data.Map.Strict as Map
import Data.Time.Clock (getCurrentTime)
import LibP2P.Crypto.PeerId (PeerId)
import LibP2P.MultistreamSelect.Negotiation
  ( NegotiationResult (..)
  , ProtocolId
  , StreamIO (..)
  , negotiateInitiator
  )
import LibP2P.Protocol.GossipSub.Heartbeat (runHeartbeat)
import LibP2P.Protocol.GossipSub.Message (readRPCMessage, writeRPCMessage)
import LibP2P.Core.Binary (word32BE)
import LibP2P.Multiaddr (protocols)
import LibP2P.Multiaddr.Protocol (Protocol (..))
import LibP2P.Protocol.GossipSub.Router
  ( addPeer
  , handleRPC
  , join
  , leave
  , newRouter
  , publish
  , removePeer
  , setPeerIP
  , setSignedPeerRecord
  )
import LibP2P.Protocol.Identify.Message (IdentifyInfo (..))
import qualified Data.Set as Set
import LibP2P.Protocol.GossipSub.Types
  ( GossipSubParams
  , GossipSubRouter (..)
  , PeerProtocol (..)
  , RPC (..)
  , SubOpts (..)
  , Topic
  , emptyRPC
  , maxRPCSize
  )
import LibP2P.Switch.ConnPool (lookupConn)
import LibP2P.Switch (removeStreamHandler, setStreamHandler)
import LibP2P.Switch.Types
  ( Connection (..)
  , MuxerSession (..)
  , Switch (..)
  )

-- | GossipSub v1.1 protocol ID (preferred).
gossipSubProtocolId :: ProtocolId
gossipSubProtocolId :: ProtocolId
gossipSubProtocolId = ProtocolId
"/meshsub/1.1.0"

-- | GossipSub v1.0 protocol ID, advertised alongside v1.1 so that
-- v1.0-only peers still get a pubsub stream (#157).
gossipSubProtocolIdV10 :: ProtocolId
gossipSubProtocolIdV10 :: ProtocolId
gossipSubProtocolIdV10 = ProtocolId
"/meshsub/1.0.0"

-- | FloodSub protocol ID, advertised alongside the meshsub protocols so
-- that floodsub-only peers still get a pubsub stream (#157,
-- gossipsub-v1.0.md "Compatibility with FloodSub").
floodSubProtocolId :: ProtocolId
floodSubProtocolId :: ProtocolId
floodSubProtocolId = ProtocolId
"/floodsub/1.0.0"

-- | All protocol IDs we register and offer, preferred first.
gossipSubProtocolIds :: [ProtocolId]
gossipSubProtocolIds :: [ProtocolId]
gossipSubProtocolIds =
  [ProtocolId
gossipSubProtocolId, ProtocolId
gossipSubProtocolIdV10, ProtocolId
floodSubProtocolId]

-- | Map a negotiated protocol ID to the peer's protocol version.
protocolFor :: ProtocolId -> PeerProtocol
protocolFor :: ProtocolId -> PeerProtocol
protocolFor ProtocolId
proto
  | ProtocolId
proto ProtocolId -> ProtocolId -> Bool
forall a. Eq a => a -> a -> Bool
== ProtocolId
gossipSubProtocolIdV10 = PeerProtocol
GossipSubV10Peer
  | ProtocolId
proto ProtocolId -> ProtocolId -> Bool
forall a. Eq a => a -> a -> Bool
== ProtocolId
floodSubProtocolId     = PeerProtocol
FloodSubPeer
  | Bool
otherwise                       = PeerProtocol
GossipSubPeer

-- | A GossipSub node: Router + Switch integration.
data GossipSubNode = GossipSubNode
  { GossipSubNode -> GossipSubRouter
gsnRouter    :: !GossipSubRouter
  , GossipSubNode -> Switch
gsnSwitch    :: !Switch
  , GossipSubNode -> TVar (Maybe (Async ()))
gsnHeartbeat :: !(TVar (Maybe (Async ())))
  , GossipSubNode -> TVar (Map PeerId StreamIO)
gsnStreams   :: !(TVar (Map.Map PeerId StreamIO))  -- ^ Cached outbound streams per peer
  }

-- | Create a new GossipSub node with a Router wired to the Switch.
--
-- The Router's gsSendRPC callback opens/reuses outbound streams to peers
-- via the Switch's connection pool.
newGossipSubNode :: Switch -> GossipSubParams -> IO GossipSubNode
newGossipSubNode :: Switch -> GossipSubParams -> IO GossipSubNode
newGossipSubNode Switch
sw GossipSubParams
params = do
  streamsVar <- Map PeerId StreamIO -> IO (TVar (Map PeerId StreamIO))
forall a. a -> IO (TVar a)
newTVarIO Map PeerId StreamIO
forall k a. Map k a
Map.empty
  hbVar <- newTVarIO Nothing
  -- Create router with real sendRPC that uses the Switch
  let localPid = Switch -> PeerId
swLocalPeerId Switch
sw
  router <- newRouter params localPid (sendRPCviaSwitch sw streamsVar) getCurrentTime
  pure GossipSubNode
    { gsnRouter    = router
    , gsnSwitch    = sw
    , gsnHeartbeat = hbVar
    , gsnStreams   = streamsVar
    }

-- | Send an RPC to a peer via cached or newly opened stream.
sendRPCviaSwitch :: Switch -> TVar (Map.Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
sendRPCviaSwitch :: Switch -> TVar (Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
sendRPCviaSwitch Switch
sw TVar (Map PeerId StreamIO)
streamsVar PeerId
pid RPC
rpc = do
  -- Try to use cached stream
  mCached <- STM (Maybe StreamIO) -> IO (Maybe StreamIO)
forall a. STM a -> IO a
atomically (STM (Maybe StreamIO) -> IO (Maybe StreamIO))
-> STM (Maybe StreamIO) -> IO (Maybe StreamIO)
forall a b. (a -> b) -> a -> b
$ PeerId -> Map PeerId StreamIO -> Maybe StreamIO
forall k a. Ord k => k -> Map k a -> Maybe a
Map.lookup PeerId
pid (Map PeerId StreamIO -> Maybe StreamIO)
-> STM (Map PeerId StreamIO) -> STM (Maybe StreamIO)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar (Map PeerId StreamIO) -> STM (Map PeerId StreamIO)
forall a. TVar a -> STM a
readTVar TVar (Map PeerId StreamIO)
streamsVar
  case mCached of
    Just StreamIO
stream -> do
      -- Try sending on cached stream; reopen on failure
      sendResult <- StreamIO -> RPC -> IO (Either () ())
trySend StreamIO
stream RPC
rpc
      case sendResult of
        Right () -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Left ()
_ -> do
          STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId StreamIO)
-> (Map PeerId StreamIO -> Map PeerId StreamIO) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar (Map PeerId StreamIO)
streamsVar (PeerId -> Map PeerId StreamIO -> Map PeerId StreamIO
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
pid)
          Switch -> TVar (Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
openAndSend Switch
sw TVar (Map PeerId StreamIO)
streamsVar PeerId
pid RPC
rpc
    Maybe StreamIO
Nothing -> Switch -> TVar (Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
openAndSend Switch
sw TVar (Map PeerId StreamIO)
streamsVar PeerId
pid RPC
rpc

-- | Open a new outbound stream to a peer and send an RPC.
openAndSend :: Switch -> TVar (Map.Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
openAndSend :: Switch -> TVar (Map PeerId StreamIO) -> PeerId -> RPC -> IO ()
openAndSend Switch
sw TVar (Map PeerId StreamIO)
streamsVar PeerId
pid RPC
rpc = do
  mStream <- Switch -> PeerId -> IO (Maybe StreamIO)
openStreamToPeer Switch
sw PeerId
pid
  case mStream of
    Maybe StreamIO
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()  -- No connection to peer; fire-and-forget
    Just StreamIO
stream -> do
      STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId StreamIO)
-> (Map PeerId StreamIO -> Map PeerId StreamIO) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' TVar (Map PeerId StreamIO)
streamsVar (PeerId -> StreamIO -> Map PeerId StreamIO -> Map PeerId StreamIO
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
pid StreamIO
stream)
      _ <- StreamIO -> RPC -> IO (Either () ())
trySend StreamIO
stream RPC
rpc
      pure ()

-- | Open a new mux stream to a peer and negotiate GossipSub protocol.
openStreamToPeer :: Switch -> PeerId -> IO (Maybe StreamIO)
openStreamToPeer :: Switch -> PeerId -> IO (Maybe StreamIO)
openStreamToPeer Switch
sw PeerId
pid = do
  mConn <- STM (Maybe Connection) -> IO (Maybe Connection)
forall a. STM a -> IO a
atomically (STM (Maybe Connection) -> IO (Maybe Connection))
-> STM (Maybe Connection) -> IO (Maybe Connection)
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId [Connection]) -> PeerId -> STM (Maybe Connection)
lookupConn (Switch -> TVar (Map PeerId [Connection])
swConnPool Switch
sw) PeerId
pid
  case mConn of
    Maybe Connection
Nothing -> Maybe StreamIO -> IO (Maybe StreamIO)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe StreamIO
forall a. Maybe a
Nothing
    Just Connection
conn -> do
      result <- (Maybe (StreamIO, PeerProtocol)
-> Either () (Maybe (StreamIO, PeerProtocol))
forall a b. b -> Either a b
Right (Maybe (StreamIO, PeerProtocol)
 -> Either () (Maybe (StreamIO, PeerProtocol)))
-> IO (Maybe (StreamIO, PeerProtocol))
-> IO (Either () (Maybe (StreamIO, PeerProtocol)))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Connection -> IO (Maybe (StreamIO, PeerProtocol))
openAndNegotiate Connection
conn) IO (Either () (Maybe (StreamIO, PeerProtocol)))
-> (SomeException
    -> IO (Either () (Maybe (StreamIO, PeerProtocol))))
-> IO (Either () (Maybe (StreamIO, PeerProtocol)))
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch`
                  (\(SomeException
_ :: SomeException) -> Either () (Maybe (StreamIO, PeerProtocol))
-> IO (Either () (Maybe (StreamIO, PeerProtocol)))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either () (Maybe (StreamIO, PeerProtocol))
forall a b. a -> Either a b
Left ()))
      case result of
        Left () -> Maybe StreamIO -> IO (Maybe StreamIO)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe StreamIO
forall a. Maybe a
Nothing
        Right Maybe (StreamIO, PeerProtocol)
mStream -> Maybe StreamIO -> IO (Maybe StreamIO)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ((StreamIO, PeerProtocol) -> StreamIO
forall a b. (a, b) -> a
fst ((StreamIO, PeerProtocol) -> StreamIO)
-> Maybe (StreamIO, PeerProtocol) -> Maybe StreamIO
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Maybe (StreamIO, PeerProtocol)
mStream)

-- | Open a mux stream and negotiate a GossipSub protocol, preferring
-- /meshsub/1.1.0 and falling back to /meshsub/1.0.0 (#157).
openAndNegotiate :: Connection -> IO (Maybe (StreamIO, PeerProtocol))
openAndNegotiate :: Connection -> IO (Maybe (StreamIO, PeerProtocol))
openAndNegotiate Connection
conn = do
  stream <- MuxerSession -> IO StreamIO
muxOpenStream (Connection -> MuxerSession
connSession Connection
conn)
  negResult <- negotiateInitiator stream gossipSubProtocolIds
  case negResult of
    Accepted ProtocolId
proto -> Maybe (StreamIO, PeerProtocol)
-> IO (Maybe (StreamIO, PeerProtocol))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ((StreamIO, PeerProtocol) -> Maybe (StreamIO, PeerProtocol)
forall a. a -> Maybe a
Just (StreamIO
stream, ProtocolId -> PeerProtocol
protocolFor ProtocolId
proto))
    NegotiationResult
NoProtocol -> Maybe (StreamIO, PeerProtocol)
-> IO (Maybe (StreamIO, PeerProtocol))
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Maybe (StreamIO, PeerProtocol)
forall a. Maybe a
Nothing

-- | Extract the remote IP bytes (4 for IPv4, 16 for IPv6) from a
-- connection's multiaddr, for P6 IP colocation scoring.
remoteIPBytes :: Connection -> Maybe ByteString
remoteIPBytes :: Connection -> Maybe ByteString
remoteIPBytes Connection
conn = [Protocol] -> Maybe ByteString
go (Multiaddr -> [Protocol]
protocols (Connection -> Multiaddr
connRemoteAddr Connection
conn))
  where
    go :: [Protocol] -> Maybe ByteString
go (IP4 Word32
w  : [Protocol]
_)   = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just (Word32 -> ByteString
word32BE Word32
w)
    go (IP6 ByteString
bs : [Protocol]
_)   = ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just ByteString
bs
    go (Protocol
_      : [Protocol]
ps)  = [Protocol] -> Maybe ByteString
go [Protocol]
ps
    go []             = Maybe ByteString
forall a. Maybe a
Nothing

-- | Try to send an RPC on a stream, catching exceptions.
trySend :: StreamIO -> RPC -> IO (Either () ())
trySend :: StreamIO -> RPC -> IO (Either () ())
trySend StreamIO
stream RPC
rpc =
  (StreamIO -> RPC -> IO ()
writeRPCMessage StreamIO
stream RPC
rpc IO () -> IO (Either () ()) -> IO (Either () ())
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> Either () () -> IO (Either () ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either () ()
forall a b. b -> Either a b
Right ()))
    IO (Either () ())
-> (SomeException -> IO (Either () ())) -> IO (Either () ())
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch` (\(SomeException
_ :: SomeException) -> Either () () -> IO (Either () ())
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> Either () ()
forall a b. a -> Either a b
Left ()))

-- | Handle an inbound GossipSub stream.
--
-- Reads framed RPCs in a loop and dispatches each to the Router's handleRPC.
-- The peer's negotiated protocol version gates v1.1 control extensions.
-- On error or EOF, cleans up the peer's cached stream and removes the peer.
handleGossipSubStream :: GossipSubNode -> StreamIO -> PeerId -> PeerProtocol
                      -> Maybe ByteString -> IO ()
handleGossipSubStream :: GossipSubNode
-> StreamIO -> PeerId -> PeerProtocol -> Maybe ByteString -> IO ()
handleGossipSubStream GossipSubNode
node StreamIO
stream PeerId
pid PeerProtocol
proto Maybe ByteString
mIP = do
  -- Register peer with router (IP feeds P6 colocation scoring)
  now <- IO UTCTime
getCurrentTime
  addPeer (gsnRouter node) pid proto False now
  mapM_ (setPeerIP (gsnRouter node) pid) mIP
  syncSignedPeerRecord node pid
  -- Read loop
  readLoop
  -- Cleanup on disconnect
  removePeer (gsnRouter node) pid
  atomically $ modifyTVar' (gsnStreams node) (Map.delete pid)
  where
    readLoop :: IO ()
readLoop = do
      result <- StreamIO -> Int -> IO (Either String RPC)
readRPCMessage StreamIO
stream Int
maxRPCSize
      case result of
        Left String
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()  -- Error/EOF: stop loop
        Right RPC
rpc -> do
          GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node) PeerId
pid RPC
rpc
          IO ()
readLoop

-- | Feed the peer's signed peer record (obtained via identify, already
-- verified against the authenticated peer id on receipt) from the
-- Switch's peer store into the router, so PRUNE-with-PX can attach it
-- when advertising this peer (#230).
syncSignedPeerRecord :: GossipSubNode -> PeerId -> IO ()
syncSignedPeerRecord :: GossipSubNode -> PeerId -> IO ()
syncSignedPeerRecord GossipSubNode
node PeerId
pid = do
  store <- STM (Map PeerId IdentifyInfo) -> IO (Map PeerId IdentifyInfo)
forall a. STM a -> IO a
atomically (STM (Map PeerId IdentifyInfo) -> IO (Map PeerId IdentifyInfo))
-> STM (Map PeerId IdentifyInfo) -> IO (Map PeerId IdentifyInfo)
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId IdentifyInfo) -> STM (Map PeerId IdentifyInfo)
forall a. TVar a -> STM a
readTVar (Switch -> TVar (Map PeerId IdentifyInfo)
swPeerStore (GossipSubNode -> Switch
gsnSwitch GossipSubNode
node))
  mapM_ (setSignedPeerRecord (gsnRouter node) pid)
    (Map.lookup pid store >>= idSignedPeerRecord)

-- | Start the GossipSub node: register stream handler, notifier, and start heartbeat.
startGossipSub :: GossipSubNode -> IO ()
startGossipSub :: GossipSubNode -> IO ()
startGossipSub GossipSubNode
node = do
  -- Register inbound stream handlers for both protocol versions (#157)
  (ProtocolId -> IO ()) -> [ProtocolId] -> IO ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
(a -> m b) -> t a -> m ()
mapM_ (\ProtocolId
protoId ->
      Switch -> ProtocolId -> StreamHandler -> IO ()
setStreamHandler (GossipSubNode -> Switch
gsnSwitch GossipSubNode
node) ProtocolId
protoId
        (\Connection
conn StreamIO
stream ->
          GossipSubNode
-> StreamIO -> PeerId -> PeerProtocol -> Maybe ByteString -> IO ()
handleGossipSubStream GossipSubNode
node StreamIO
stream (Connection -> PeerId
connPeerId Connection
conn)
            (ProtocolId -> PeerProtocol
protocolFor ProtocolId
protoId) (Connection -> Maybe ByteString
remoteIPBytes Connection
conn)))
    [ProtocolId]
gossipSubProtocolIds
  -- Register connection notifier to auto-open GossipSub streams to new peers
  STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar [Connection -> IO ()]
-> ([Connection -> IO ()] -> [Connection -> IO ()]) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (Switch -> TVar [Connection -> IO ()]
swNotifiers (GossipSubNode -> Switch
gsnSwitch GossipSubNode
node))
    (GossipSubNode -> Connection -> IO ()
onNewConnection GossipSubNode
node (Connection -> IO ())
-> [Connection -> IO ()] -> [Connection -> IO ()]
forall a. a -> [a] -> [a]
:)
  -- Start heartbeat background thread
  hbAsync <- GossipSubRouter -> IO (Async ())
runHeartbeat (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node)
  atomically $ writeTVar (gsnHeartbeat node) (Just hbAsync)

-- | Called on new connection: open a GossipSub stream to the peer.
-- Caches the stream for outbound writes and starts a read loop
-- on it to receive RPCs sent back by the remote peer (e.g. subscriptions).
onNewConnection :: GossipSubNode -> Connection -> IO ()
onNewConnection :: GossipSubNode -> Connection -> IO ()
onNewConnection GossipSubNode
node Connection
conn = do
  let pid :: PeerId
pid = Connection -> PeerId
connPeerId Connection
conn
  -- Open a mux stream and negotiate GossipSub protocol
  mStream <- Connection -> IO (Maybe (StreamIO, PeerProtocol))
openAndNegotiate Connection
conn
  case mStream of
    Maybe (StreamIO, PeerProtocol)
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()  -- Peer doesn't support GossipSub
    Just (StreamIO
stream, PeerProtocol
proto) -> do
      -- Cache the outbound stream
      STM () -> IO ()
forall a. STM a -> IO a
atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ TVar (Map PeerId StreamIO)
-> (Map PeerId StreamIO -> Map PeerId StreamIO) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (GossipSubNode -> TVar (Map PeerId StreamIO)
gsnStreams GossipSubNode
node) (PeerId -> StreamIO -> Map PeerId StreamIO -> Map PeerId StreamIO
forall k a. Ord k => k -> a -> Map k a -> Map k a
Map.insert PeerId
pid StreamIO
stream)
      -- Register peer with its negotiated protocol version
      -- (IP feeds P6 colocation scoring)
      now <- IO UTCTime
getCurrentTime
      addPeer (gsnRouter node) pid proto True now
      mapM_ (setPeerIP (gsnRouter node) pid) (remoteIPBytes conn)
      syncSignedPeerRecord node pid
      -- Send current subscriptions to the new peer
      sendCurrentSubscriptions node stream
      -- Start read loop on this stream to receive RPCs from the peer
      -- (e.g. subscription announcements sent back on the same yamux stream)
      _ <- async $ outboundReadLoop node stream pid
      pure ()

-- | Send current topic subscriptions to a newly connected peer.
-- This ensures peers joining after we've already subscribed still learn
-- about our subscriptions (standard GossipSub behavior).
-- Writes directly to the stream to avoid any routing issues.
sendCurrentSubscriptions :: GossipSubNode -> StreamIO -> IO ()
sendCurrentSubscriptions :: GossipSubNode -> StreamIO -> IO ()
sendCurrentSubscriptions GossipSubNode
node StreamIO
stream = do
  let router :: GossipSubRouter
router = GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node
  -- Read the subscription set, not mesh keys: a topic joined before any
  -- peer was known has no mesh entry but must still be announced (#155).
  subs <- STM (Set ProtocolId) -> IO (Set ProtocolId)
forall a. STM a -> IO a
atomically (STM (Set ProtocolId) -> IO (Set ProtocolId))
-> STM (Set ProtocolId) -> IO (Set ProtocolId)
forall a b. (a -> b) -> a -> b
$ TVar (Set ProtocolId) -> STM (Set ProtocolId)
forall a. TVar a -> STM a
readTVar (GossipSubRouter -> TVar (Set ProtocolId)
gsSubscriptions GossipSubRouter
router)
  let topics = Set ProtocolId -> [ProtocolId]
forall a. Set a -> [a]
Set.toList Set ProtocolId
subs
  if null topics
    then pure ()
    else do
      let subRPC = RPC
emptyRPC
            { rpcSubscriptions = map (\ProtocolId
t -> Bool -> ProtocolId -> SubOpts
SubOpts Bool
True ProtocolId
t) topics }
      _ <- trySend stream subRPC
      pure ()

-- | Read loop on the outbound stream.
-- Handles RPCs sent back by the remote peer on the same yamux stream
-- (e.g. subscription announcements). Does NOT remove the peer on
-- EOF since the inbound handler or another mechanism manages peer lifecycle.
outboundReadLoop :: GossipSubNode -> StreamIO -> PeerId -> IO ()
outboundReadLoop :: GossipSubNode -> StreamIO -> PeerId -> IO ()
outboundReadLoop GossipSubNode
node StreamIO
stream PeerId
pid = IO ()
loop
  where
    loop :: IO ()
loop = do
      result <- StreamIO -> Int -> IO (Either String RPC)
readRPCMessage StreamIO
stream Int
maxRPCSize
      case result of
        Left String
_ -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()  -- EOF or error: stop
        Right RPC
rpc -> do
          GossipSubRouter -> PeerId -> RPC -> IO ()
handleRPC (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node) PeerId
pid RPC
rpc
          IO ()
loop

-- | Stop the GossipSub node: cancel heartbeat and unregister handler.
stopGossipSub :: GossipSubNode -> IO ()
stopGossipSub :: GossipSubNode -> IO ()
stopGossipSub GossipSubNode
node = do
  -- Cancel heartbeat
  mHb <- STM (Maybe (Async ())) -> IO (Maybe (Async ()))
forall a. STM a -> IO a
atomically (STM (Maybe (Async ())) -> IO (Maybe (Async ())))
-> STM (Maybe (Async ())) -> IO (Maybe (Async ()))
forall a b. (a -> b) -> a -> b
$ do
    hb <- TVar (Maybe (Async ())) -> STM (Maybe (Async ()))
forall a. TVar a -> STM a
readTVar (GossipSubNode -> TVar (Maybe (Async ()))
gsnHeartbeat GossipSubNode
node)
    writeTVar (gsnHeartbeat node) Nothing
    pure hb
  case mHb of
    Just Async ()
hbAsync -> Async () -> IO ()
forall a. Async a -> IO ()
cancel Async ()
hbAsync IO () -> (SomeException -> IO ()) -> IO ()
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
`catch` (\(SomeException
_ :: SomeException) -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
    Maybe (Async ())
Nothing -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  -- Unregister stream handlers for both protocol versions
  mapM_ (removeStreamHandler (gsnSwitch node)) gossipSubProtocolIds

-- | Subscribe to a topic.
gossipJoin :: GossipSubNode -> Topic -> IO ()
gossipJoin :: GossipSubNode -> ProtocolId -> IO ()
gossipJoin GossipSubNode
node ProtocolId
topic = GossipSubRouter -> ProtocolId -> IO ()
join (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node) ProtocolId
topic

-- | Unsubscribe from a topic.
gossipLeave :: GossipSubNode -> Topic -> IO ()
gossipLeave :: GossipSubNode -> ProtocolId -> IO ()
gossipLeave GossipSubNode
node ProtocolId
topic = GossipSubRouter -> ProtocolId -> IO ()
leave (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node) ProtocolId
topic

-- | Publish a message to a topic (signed with the Switch's identity key).
gossipPublish :: GossipSubNode -> Topic -> ByteString -> IO ()
gossipPublish :: GossipSubNode -> ProtocolId -> ByteString -> IO ()
gossipPublish GossipSubNode
node ProtocolId
topic ByteString
payload =
  GossipSubRouter
-> ProtocolId -> ByteString -> Maybe KeyPair -> IO ()
publish (GossipSubNode -> GossipSubRouter
gsnRouter GossipSubNode
node) ProtocolId
topic ByteString
payload (KeyPair -> Maybe KeyPair
forall a. a -> Maybe a
Just (Switch -> KeyPair
swIdentityKey (GossipSubNode -> Switch
gsnSwitch GossipSubNode
node)))