-- | Circuit Relay v2 server: manage reservations and bridge streams.
--
-- Protocols:
--   /libp2p/circuit/relay/0.2.0/hop (client ↔ relay)
--
-- Provides:
--   - Reservation management (with expiration and limits)
--   - Stream bridging between source and target
--   - Resource limits (max reservations, max circuits, data/duration limits)
module LibP2P.NAT.Relay
  ( -- * Types
    RelayConfig (..)
  , RelayState (..)
  , ActiveReservation (..)
  , HopContext (..)
    -- * Configuration
  , defaultRelayConfig
    -- * State management
  , newRelayState
    -- * Handlers
  , handleReserve
  , handleConnect
    -- * Stream bridging
  , bridgeStreams
    -- * Relay address helpers
  , buildRelayAddrBytes
  , isRelayedConnection
  , isRelayedAddr
    -- * Voucher constants
  , relayRsvpDomain
  , relayRsvpPayloadType
  ) where

import Data.ByteString (ByteString)
import qualified Data.ByteString as BS
import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (race_)
import Control.Concurrent.STM
import Control.Exception (finally)
import Data.IORef (IORef, newIORef, readIORef, modifyIORef')
import qualified Data.Map.Strict as Map
import Data.Time.Clock.POSIX (getPOSIXTime)
import Data.Word (Word32, Word64)
import LibP2P.NAT.Relay.Message
import LibP2P.Multiaddr (Multiaddr (..), fromBytes, toBytes)
import LibP2P.Multiaddr.Protocol (Protocol (..))
import LibP2P.MultistreamSelect.Negotiation (StreamIO (..))
import LibP2P.Crypto.Key (KeyPair (..))
import LibP2P.Crypto.PeerId (PeerId (..), peerIdBytes)
import LibP2P.Crypto.SignedEnvelope (createEnvelope, encodeSignedEnvelope)
import LibP2P.Core.Varint (encodeUvarint)

-- | Relay server configuration.
data RelayConfig = RelayConfig
  { RelayConfig -> Int
rcMaxReservations      :: !Int      -- ^ Max concurrent reservations
  , RelayConfig -> Int
rcMaxCircuits          :: !Int      -- ^ Max concurrent relayed circuits
  , RelayConfig -> Word64
rcReservationDuration  :: !Word64   -- ^ Reservation duration (seconds)
  , RelayConfig -> Word64
rcDefaultDataLimit     :: !Word64   -- ^ Default data limit per circuit (bytes)
  , RelayConfig -> Word32
rcDefaultDurationLimit :: !Word32   -- ^ Default duration limit per circuit (seconds)
  } deriving (Int -> RelayConfig -> ShowS
[RelayConfig] -> ShowS
RelayConfig -> String
(Int -> RelayConfig -> ShowS)
-> (RelayConfig -> String)
-> ([RelayConfig] -> ShowS)
-> Show RelayConfig
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> RelayConfig -> ShowS
showsPrec :: Int -> RelayConfig -> ShowS
$cshow :: RelayConfig -> String
show :: RelayConfig -> String
$cshowList :: [RelayConfig] -> ShowS
showList :: [RelayConfig] -> ShowS
Show, RelayConfig -> RelayConfig -> Bool
(RelayConfig -> RelayConfig -> Bool)
-> (RelayConfig -> RelayConfig -> Bool) -> Eq RelayConfig
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: RelayConfig -> RelayConfig -> Bool
== :: RelayConfig -> RelayConfig -> Bool
$c/= :: RelayConfig -> RelayConfig -> Bool
/= :: RelayConfig -> RelayConfig -> Bool
Eq)

-- | Default relay configuration.
defaultRelayConfig :: RelayConfig
defaultRelayConfig :: RelayConfig
defaultRelayConfig = RelayConfig
  { rcMaxReservations :: Int
rcMaxReservations      = Int
128
  , rcMaxCircuits :: Int
rcMaxCircuits          = Int
16
  , rcReservationDuration :: Word64
rcReservationDuration  = Word64
3600  -- 1 hour
  , rcDefaultDataLimit :: Word64
rcDefaultDataLimit     = Word64
131072  -- 128 KiB
  , rcDefaultDurationLimit :: Word32
rcDefaultDurationLimit = Word32
120  -- 2 minutes
  }

-- | An active reservation for a peer.
data ActiveReservation = ActiveReservation
  { ActiveReservation -> PeerId
arPeerId     :: !PeerId
  , ActiveReservation -> Word64
arExpiration :: !Word64   -- ^ Absolute expiration time (UTC Unix time, seconds)
  } deriving (Int -> ActiveReservation -> ShowS
[ActiveReservation] -> ShowS
ActiveReservation -> String
(Int -> ActiveReservation -> ShowS)
-> (ActiveReservation -> String)
-> ([ActiveReservation] -> ShowS)
-> Show ActiveReservation
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> ActiveReservation -> ShowS
showsPrec :: Int -> ActiveReservation -> ShowS
$cshow :: ActiveReservation -> String
show :: ActiveReservation -> String
$cshowList :: [ActiveReservation] -> ShowS
showList :: [ActiveReservation] -> ShowS
Show, ActiveReservation -> ActiveReservation -> Bool
(ActiveReservation -> ActiveReservation -> Bool)
-> (ActiveReservation -> ActiveReservation -> Bool)
-> Eq ActiveReservation
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: ActiveReservation -> ActiveReservation -> Bool
== :: ActiveReservation -> ActiveReservation -> Bool
$c/= :: ActiveReservation -> ActiveReservation -> Bool
/= :: ActiveReservation -> ActiveReservation -> Bool
Eq)

-- | Mutable relay server state.
data RelayState = RelayState
  { RelayState -> RelayConfig
rsConfig        :: !RelayConfig
  , RelayState -> TVar (Map PeerId ActiveReservation)
rsReservations  :: !(TVar (Map.Map PeerId ActiveReservation))
  , RelayState -> TVar (Map PeerId Int)
rsCircuitCounts :: !(TVar (Map.Map PeerId Int))
    -- ^ Active relayed circuits per peer (both initiator and target sides)
  }

-- | Create new relay state from configuration.
newRelayState :: RelayConfig -> IO RelayState
newRelayState :: RelayConfig -> IO RelayState
newRelayState RelayConfig
config = RelayConfig
-> TVar (Map PeerId ActiveReservation)
-> TVar (Map PeerId Int)
-> RelayState
RelayState RelayConfig
config
  (TVar (Map PeerId ActiveReservation)
 -> TVar (Map PeerId Int) -> RelayState)
-> IO (TVar (Map PeerId ActiveReservation))
-> IO (TVar (Map PeerId Int) -> RelayState)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Map PeerId ActiveReservation
-> IO (TVar (Map PeerId ActiveReservation))
forall a. a -> IO (TVar a)
newTVarIO Map PeerId ActiveReservation
forall k a. Map k a
Map.empty
  IO (TVar (Map PeerId Int) -> RelayState)
-> IO (TVar (Map PeerId Int)) -> IO RelayState
forall a b. IO (a -> b) -> IO a -> IO b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Map PeerId Int -> IO (TVar (Map PeerId Int))
forall a. a -> IO (TVar a)
newTVarIO Map PeerId Int
forall k a. Map k a
Map.empty

-- | Per-request context for the hop handlers: the relay's own identity
-- (used to sign reservation vouchers), its public addresses (advertised in
-- the reservation), and the address the requesting connection arrived over
-- (used to refuse requests over already-relayed connections).
data HopContext = HopContext
  { HopContext -> PeerId
hcRelayId    :: !PeerId       -- ^ Relay's own peer ID
  , HopContext -> KeyPair
hcRelayKey   :: !KeyPair      -- ^ Relay identity key, signs vouchers
  , HopContext -> [Multiaddr]
hcRelayAddrs :: ![Multiaddr]
    -- ^ Relay public addresses including the trailing @/p2p/\<relay\>@
    -- component but without @/p2p-circuit@, per circuit-v2 reservation addrs
  , HopContext -> Multiaddr
hcRemoteAddr :: !Multiaddr    -- ^ Address the request arrived over
  }

-- | Domain separation string for reservation voucher envelopes (circuit-v2).
relayRsvpDomain :: ByteString
relayRsvpDomain :: ByteString
relayRsvpDomain = ByteString
"libp2p-relay-rsvp"

-- | Multicodec payload type for reservation vouchers (0x0302).
relayRsvpPayloadType :: ByteString
relayRsvpPayloadType :: ByteString
relayRsvpPayloadType = [Word8] -> ByteString
BS.pack [Word8
0x03, Word8
0x02]

-- | Handle a RESERVE request from a peer.
handleReserve :: RelayState -> HopContext -> StreamIO -> PeerId -> IO ()
handleReserve :: RelayState -> HopContext -> StreamIO -> PeerId -> IO ()
handleReserve RelayState
state HopContext
ctx StreamIO
stream PeerId
peerId
  -- Per circuit-v2 spec, relays must not serve requests arriving over an
  -- already-relayed connection (no relay chains).
  | Multiaddr -> Bool
isRelayedAddr (HopContext -> Multiaddr
hcRemoteAddr HopContext
ctx) = StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
PermissionDenied
  | Bool
otherwise = do
      -- Per circuit-v2 spec, Reservation.expire is an absolute UTC Unix
      -- time in seconds, not a duration.
      now <- IO POSIXTime
getPOSIXTime
      let expiration = POSIXTime -> Word64
forall b. Integral b => POSIXTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor POSIXTime
now Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ RelayConfig -> Word64
rcReservationDuration (RelayState -> RelayConfig
rsConfig RelayState
state)
          reservation = ActiveReservation
            { arPeerId :: PeerId
arPeerId = PeerId
peerId
            , arExpiration :: Word64
arExpiration = Word64
expiration
            }
      granted <- atomically $ do
        reservations <- readTVar (rsReservations state)
        let limit = RelayConfig -> Int
rcMaxReservations (RelayState -> RelayConfig
rsConfig RelayState
state)
            isRefresh = PeerId -> Map PeerId ActiveReservation -> Bool
forall k a. Ord k => k -> Map k a -> Bool
Map.member PeerId
peerId Map PeerId ActiveReservation
reservations
        -- A refresh from an existing holder is not a new reservation: it
        -- must succeed even when the relay is at capacity.
        if not isRefresh && Map.size reservations >= limit
          then pure False
          else do
            modifyTVar' (rsReservations state) (Map.insert peerId reservation)
            pure True
      if not granted
        -- Per circuit-v2 spec, a reservation rejected because the relay is
        -- at capacity is RESERVATION_REFUSED (200); RESOURCE_LIMIT_EXCEEDED
        -- (201) is reserved for relayed-connection limits on CONNECT.
        then sendHopStatus stream ReservationRefused
        else do
          -- Send OK response with reservation info
          let resp = HopMessage
                { hopType :: Maybe HopMessageType
hopType = HopMessageType -> Maybe HopMessageType
forall a. a -> Maybe a
Just HopMessageType
HopStatus
                , hopPeer :: Maybe RelayPeer
hopPeer = Maybe RelayPeer
forall a. Maybe a
Nothing
                , hopReservation :: Maybe Reservation
hopReservation = Reservation -> Maybe Reservation
forall a. a -> Maybe a
Just Reservation
                    { rsvExpire :: Maybe Word64
rsvExpire = Word64 -> Maybe Word64
forall a. a -> Maybe a
Just Word64
expiration
                    , rsvAddrs :: [ByteString]
rsvAddrs = (Multiaddr -> ByteString) -> [Multiaddr] -> [ByteString]
forall a b. (a -> b) -> [a] -> [b]
map Multiaddr -> ByteString
toBytes (HopContext -> [Multiaddr]
hcRelayAddrs HopContext
ctx)
                    , rsvVoucher :: Maybe ByteString
rsvVoucher = HopContext -> PeerId -> Word64 -> Maybe ByteString
signVoucher HopContext
ctx PeerId
peerId Word64
expiration
                    }
                , hopLimit :: Maybe RelayLimit
hopLimit = RelayLimit -> Maybe RelayLimit
forall a. a -> Maybe a
Just RelayLimit
                    { rlDuration :: Maybe Word32
rlDuration = Word32 -> Maybe Word32
forall a. a -> Maybe a
Just (RelayConfig -> Word32
rcDefaultDurationLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                    , rlData :: Maybe Word64
rlData = Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (RelayConfig -> Word64
rcDefaultDataLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                    }
                , hopStatus :: Maybe RelayStatus
hopStatus = RelayStatus -> Maybe RelayStatus
forall a. a -> Maybe a
Just RelayStatus
RelayOK
                }
          writeHopMessage stream resp

-- | Sign a reservation voucher: a circuit-v2 Voucher payload wrapped in an
-- RFC 0002 signed envelope under the relay-rsvp domain.
signVoucher :: HopContext -> PeerId -> Word64 -> Maybe ByteString
signVoucher :: HopContext -> PeerId -> Word64 -> Maybe ByteString
signVoucher HopContext
ctx PeerId
peerId Word64
expiration =
  let payload :: ByteString
payload = Voucher -> ByteString
encodeVoucher Voucher
        { vRelay :: ByteString
vRelay = PeerId -> ByteString
peerIdBytes (HopContext -> PeerId
hcRelayId HopContext
ctx)
        , vPeer :: ByteString
vPeer = PeerId -> ByteString
peerIdBytes PeerId
peerId
        , vExpiration :: Word64
vExpiration = Word64
expiration
        }
      kp :: KeyPair
kp = HopContext -> KeyPair
hcRelayKey HopContext
ctx
  in case PrivateKey
-> PublicKey
-> ByteString
-> ByteString
-> ByteString
-> Either String SignedEnvelope
createEnvelope (KeyPair -> PrivateKey
kpPrivate KeyPair
kp) (KeyPair -> PublicKey
kpPublic KeyPair
kp) ByteString
relayRsvpDomain ByteString
relayRsvpPayloadType ByteString
payload of
       Left String
_    -> Maybe ByteString
forall a. Maybe a
Nothing  -- voucher is optional per spec; omit if signing fails
       Right SignedEnvelope
env -> ByteString -> Maybe ByteString
forall a. a -> Maybe a
Just (SignedEnvelope -> ByteString
encodeSignedEnvelope SignedEnvelope
env)

-- | Handle a CONNECT request from a peer.
-- The openStopStream callback is used to open a stop stream to the target.
handleConnect :: RelayState -> HopContext -> StreamIO -> PeerId -> HopMessage -> (PeerId -> IO (Maybe StreamIO)) -> IO ()
handleConnect :: RelayState
-> HopContext
-> StreamIO
-> PeerId
-> HopMessage
-> (PeerId -> IO (Maybe StreamIO))
-> IO ()
handleConnect RelayState
state HopContext
ctx StreamIO
stream PeerId
sourcePeerId HopMessage
msg PeerId -> IO (Maybe StreamIO)
openStopStream
  -- Per circuit-v2 spec, a CONNECT arriving over an already-relayed
  -- connection is refused: circuits cannot be chained through relays.
  | Multiaddr -> Bool
isRelayedAddr (HopContext -> Multiaddr
hcRemoteAddr HopContext
ctx) = StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
PermissionDenied
  | Bool
otherwise = case HopMessage -> Maybe RelayPeer
hopPeer HopMessage
msg of
    Maybe RelayPeer
Nothing -> StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
MalformedMessage
    Just RelayPeer
peer -> do
      let targetId :: PeerId
targetId = ByteString -> PeerId
PeerId (RelayPeer -> ByteString
rpId RelayPeer
peer)
      now <- IO POSIXTime
getPOSIXTime
      let nowSecs = POSIXTime -> Word64
forall b. Integral b => POSIXTime -> b
forall a b. (RealFrac a, Integral b) => a -> b
floor POSIXTime
now :: Word64
      -- Check target has an unexpired reservation; drop it if it has expired.
      hasReservation <- atomically $ do
        reservations <- readTVar (rsReservations state)
        case Map.lookup targetId reservations of
          Maybe ActiveReservation
Nothing -> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
          Just ActiveReservation
rsv
            | ActiveReservation -> Word64
arExpiration ActiveReservation
rsv Word64 -> Word64 -> Bool
forall a. Ord a => a -> a -> Bool
<= Word64
nowSecs -> do
                TVar (Map PeerId ActiveReservation)
-> (Map PeerId ActiveReservation -> Map PeerId ActiveReservation)
-> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (RelayState -> TVar (Map PeerId ActiveReservation)
rsReservations RelayState
state) (PeerId
-> Map PeerId ActiveReservation -> Map PeerId ActiveReservation
forall k a. Ord k => k -> Map k a -> Map k a
Map.delete PeerId
targetId)
                Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
            | Bool
otherwise -> Bool -> STM Bool
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
      if not hasReservation
        then sendHopStatus stream NoReservation
        else do
          -- Acquire a circuit slot for both peers, enforcing the advertised
          -- per-peer circuit limit atomically with the check.
          acquired <- atomically $ acquireCircuitSlots state sourcePeerId targetId
          if not acquired
            then sendHopStatus stream ResourceLimitExceeded
            else establishCircuit state stream sourcePeerId targetId openStopStream
                   `finally` atomically (releaseCircuitSlots state sourcePeerId targetId)

-- | Reserve one circuit slot for each of the two peers if neither is at the
-- configured per-peer limit. Returns False without modifying state otherwise.
acquireCircuitSlots :: RelayState -> PeerId -> PeerId -> STM Bool
acquireCircuitSlots :: RelayState -> PeerId -> PeerId -> STM Bool
acquireCircuitSlots RelayState
state PeerId
sourceId PeerId
targetId = do
  counts <- TVar (Map PeerId Int) -> STM (Map PeerId Int)
forall a. TVar a -> STM a
readTVar (RelayState -> TVar (Map PeerId Int)
rsCircuitCounts RelayState
state)
  let limit = RelayConfig -> Int
rcMaxCircuits (RelayState -> RelayConfig
rsConfig RelayState
state)
      countOf PeerId
pid = Int -> PeerId -> Map PeerId Int -> Int
forall k a. Ord k => a -> k -> Map k a -> a
Map.findWithDefault Int
0 PeerId
pid Map PeerId Int
counts
  if countOf sourceId >= limit || countOf targetId >= limit
    then pure False
    else do
      let bump = (Int -> Int -> Int)
-> PeerId -> Int -> Map PeerId Int -> Map PeerId Int
forall k a. Ord k => (a -> a -> a) -> k -> a -> Map k a -> Map k a
Map.insertWith Int -> Int -> Int
forall a. Num a => a -> a -> a
(+) PeerId
sourceId Int
1 (Map PeerId Int -> Map PeerId Int)
-> (Map PeerId Int -> Map PeerId Int)
-> Map PeerId Int
-> Map PeerId Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (Int -> Int -> Int)
-> PeerId -> Int -> Map PeerId Int -> Map PeerId Int
forall k a. Ord k => (a -> a -> a) -> k -> a -> Map k a -> Map k a
Map.insertWith Int -> Int -> Int
forall a. Num a => a -> a -> a
(+) PeerId
targetId Int
1
      writeTVar (rsCircuitCounts state) (bump counts)
      pure True

-- | Release the circuit slots acquired by 'acquireCircuitSlots'.
releaseCircuitSlots :: RelayState -> PeerId -> PeerId -> STM ()
releaseCircuitSlots :: RelayState -> PeerId -> PeerId -> STM ()
releaseCircuitSlots RelayState
state PeerId
sourceId PeerId
targetId =
  TVar (Map PeerId Int)
-> (Map PeerId Int -> Map PeerId Int) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
modifyTVar' (RelayState -> TVar (Map PeerId Int)
rsCircuitCounts RelayState
state) (PeerId -> Map PeerId Int -> Map PeerId Int
dropOne PeerId
sourceId (Map PeerId Int -> Map PeerId Int)
-> (Map PeerId Int -> Map PeerId Int)
-> Map PeerId Int
-> Map PeerId Int
forall b c a. (b -> c) -> (a -> b) -> a -> c
. PeerId -> Map PeerId Int -> Map PeerId Int
dropOne PeerId
targetId)
  where
    dropOne :: PeerId -> Map PeerId Int -> Map PeerId Int
dropOne = (Int -> Maybe Int) -> PeerId -> Map PeerId Int -> Map PeerId Int
forall k a. Ord k => (a -> Maybe a) -> k -> Map k a -> Map k a
Map.update (\Int
n -> if Int
n Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
<= Int
1 then Maybe Int
forall a. Maybe a
Nothing else Int -> Maybe Int
forall a. a -> Maybe a
Just (Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
- Int
1))

-- | Open a stop stream to the target, exchange CONNECT/STATUS, and bridge the
-- two streams until EOF or a limit is exceeded.
establishCircuit :: RelayState -> StreamIO -> PeerId -> PeerId -> (PeerId -> IO (Maybe StreamIO)) -> IO ()
establishCircuit :: RelayState
-> StreamIO
-> PeerId
-> PeerId
-> (PeerId -> IO (Maybe StreamIO))
-> IO ()
establishCircuit RelayState
state StreamIO
stream PeerId
sourcePeerId PeerId
targetId PeerId -> IO (Maybe StreamIO)
openStopStream = do
          -- Try to open stop stream to target
          mStopStream <- PeerId -> IO (Maybe StreamIO)
openStopStream PeerId
targetId
          case mStopStream of
            Maybe StreamIO
Nothing -> StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
ConnectionFailed
            Just StreamIO
stopStream -> do
              -- Send CONNECT to target via stop protocol
              let stopMsg :: StopMessage
stopMsg = StopMessage
                    { stopType :: Maybe StopMessageType
stopType = StopMessageType -> Maybe StopMessageType
forall a. a -> Maybe a
Just StopMessageType
StopConnect
                    , stopPeer :: Maybe RelayPeer
stopPeer = RelayPeer -> Maybe RelayPeer
forall a. a -> Maybe a
Just RelayPeer
                        { rpId :: ByteString
rpId = let PeerId ByteString
bs = PeerId
sourcePeerId in ByteString
bs
                        , rpAddrs :: [ByteString]
rpAddrs = []
                        }
                    , stopLimit :: Maybe RelayLimit
stopLimit = RelayLimit -> Maybe RelayLimit
forall a. a -> Maybe a
Just RelayLimit
                        { rlDuration :: Maybe Word32
rlDuration = Word32 -> Maybe Word32
forall a. a -> Maybe a
Just (RelayConfig -> Word32
rcDefaultDurationLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                        , rlData :: Maybe Word64
rlData = Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (RelayConfig -> Word64
rcDefaultDataLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                        }
                    , stopStatus :: Maybe RelayStatus
stopStatus = Maybe RelayStatus
forall a. Maybe a
Nothing
                    }
              StreamIO -> StopMessage -> IO ()
writeStopMessage StreamIO
stopStream StopMessage
stopMsg
              -- Wait for target's STATUS response
              targetResp <- StreamIO -> Int -> IO (Either String StopMessage)
readStopMessage StreamIO
stopStream Int
maxRelayMessageSize
              case targetResp of
                Right StopMessage
resp | StopMessage -> Maybe RelayStatus
stopStatus StopMessage
resp Maybe RelayStatus -> Maybe RelayStatus -> Bool
forall a. Eq a => a -> a -> Bool
== RelayStatus -> Maybe RelayStatus
forall a. a -> Maybe a
Just RelayStatus
RelayOK -> do
                  -- Notify source of success
                  let okResp :: HopMessage
okResp = HopMessage
                        { hopType :: Maybe HopMessageType
hopType = HopMessageType -> Maybe HopMessageType
forall a. a -> Maybe a
Just HopMessageType
HopStatus
                        , hopPeer :: Maybe RelayPeer
hopPeer = Maybe RelayPeer
forall a. Maybe a
Nothing
                        , hopReservation :: Maybe Reservation
hopReservation = Maybe Reservation
forall a. Maybe a
Nothing
                        , hopLimit :: Maybe RelayLimit
hopLimit = RelayLimit -> Maybe RelayLimit
forall a. a -> Maybe a
Just RelayLimit
                            { rlDuration :: Maybe Word32
rlDuration = Word32 -> Maybe Word32
forall a. a -> Maybe a
Just (RelayConfig -> Word32
rcDefaultDurationLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                            , rlData :: Maybe Word64
rlData = Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (RelayConfig -> Word64
rcDefaultDataLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                            }
                        , hopStatus :: Maybe RelayStatus
hopStatus = RelayStatus -> Maybe RelayStatus
forall a. a -> Maybe a
Just RelayStatus
RelayOK
                        }
                  StreamIO -> HopMessage -> IO ()
writeHopMessage StreamIO
stream HopMessage
okResp
                  -- Bridge the two streams
                  let limit :: Maybe RelayLimit
limit = RelayLimit -> Maybe RelayLimit
forall a. a -> Maybe a
Just RelayLimit
                        { rlDuration :: Maybe Word32
rlDuration = Word32 -> Maybe Word32
forall a. a -> Maybe a
Just (RelayConfig -> Word32
rcDefaultDurationLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                        , rlData :: Maybe Word64
rlData = Word64 -> Maybe Word64
forall a. a -> Maybe a
Just (RelayConfig -> Word64
rcDefaultDataLimit (RelayState -> RelayConfig
rsConfig RelayState
state))
                        }
                  Maybe RelayLimit -> StreamIO -> StreamIO -> IO ()
bridgeStreams Maybe RelayLimit
limit StreamIO
stream StreamIO
stopStream
                Either String StopMessage
_ -> StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
ConnectionFailed

-- | Send a simple HopMessage STATUS response.
sendHopStatus :: StreamIO -> RelayStatus -> IO ()
sendHopStatus :: StreamIO -> RelayStatus -> IO ()
sendHopStatus StreamIO
stream RelayStatus
status = StreamIO -> HopMessage -> IO ()
writeHopMessage StreamIO
stream HopMessage
  { hopType :: Maybe HopMessageType
hopType = HopMessageType -> Maybe HopMessageType
forall a. a -> Maybe a
Just HopMessageType
HopStatus
  , hopPeer :: Maybe RelayPeer
hopPeer = Maybe RelayPeer
forall a. Maybe a
Nothing
  , hopReservation :: Maybe Reservation
hopReservation = Maybe Reservation
forall a. Maybe a
Nothing
  , hopLimit :: Maybe RelayLimit
hopLimit = Maybe RelayLimit
forall a. Maybe a
Nothing
  , hopStatus :: Maybe RelayStatus
hopStatus = RelayStatus -> Maybe RelayStatus
forall a. a -> Maybe a
Just RelayStatus
status
  }

-- | Bridge two streams bidirectionally with optional data/duration limits.
-- Terminates when either direction closes or limits are exceeded, then closes
-- both streams so neither end is left with a half-open circuit.
bridgeStreams :: Maybe RelayLimit -> StreamIO -> StreamIO -> IO ()
bridgeStreams :: Maybe RelayLimit -> StreamIO -> StreamIO -> IO ()
bridgeStreams Maybe RelayLimit
mLimit StreamIO
streamA StreamIO
streamB = do
  let dataLimit :: Int
dataLimit = case Maybe RelayLimit
mLimit Maybe RelayLimit -> (RelayLimit -> Maybe Word64) -> Maybe Word64
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= RelayLimit -> Maybe Word64
rlData of
        Just Word64
n  -> Word64 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word64
n :: Int
        Maybe Word64
Nothing -> Int
forall a. Bounded a => a
maxBound
      -- Duration limit in microseconds for threadDelay
      mDurationMicros :: Maybe Int
mDurationMicros = case Maybe RelayLimit
mLimit Maybe RelayLimit -> (RelayLimit -> Maybe Word32) -> Maybe Word32
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= RelayLimit -> Maybe Word32
rlDuration of
        Just Word32
secs | Word32
secs Word32 -> Word32 -> Bool
forall a. Ord a => a -> a -> Bool
> Word32
0 -> Int -> Maybe Int
forall a. a -> Maybe a
Just (Word32 -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Word32
secs Int -> Int -> Int
forall a. Num a => a -> a -> a
* Int
1000000 :: Int)
        Maybe Word32
_                    -> Maybe Int
forall a. Maybe a
Nothing
  -- Track bytes transferred in each direction
  countAtoB <- Int -> IO (IORef Int)
forall a. a -> IO (IORef a)
newIORef (Int
0 :: Int)
  countBtoA <- newIORef (0 :: Int)
  -- Forward A→B and B→A concurrently; terminate when either finishes
  let forwarding = IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO ()
race_
        (StreamIO -> StreamIO -> IORef Int -> Int -> IO ()
forwardWithLimit StreamIO
streamA StreamIO
streamB IORef Int
countAtoB Int
dataLimit)
        (StreamIO -> StreamIO -> IORef Int -> Int -> IO ()
forwardWithLimit StreamIO
streamB StreamIO
streamA IORef Int
countBtoA Int
dataLimit)
  -- Enforce the advertised duration limit: close the circuit when it elapses
  (case mDurationMicros of
     Maybe Int
Nothing     -> IO ()
forwarding
     Just Int
micros -> IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO ()
race_ (Int -> IO ()
threadDelay Int
micros) IO ()
forwarding)
    `finally` do
      streamClose streamA
      streamClose streamB

-- | Forward bytes from source to destination with a byte limit.
-- The limit is checked before each read, so the circuit terminates as soon
-- as exactly @limit@ bytes have been forwarded — no byte beyond the limit
-- is consumed from the source.
forwardWithLimit :: StreamIO -> StreamIO -> IORef Int -> Int -> IO ()
forwardWithLimit :: StreamIO -> StreamIO -> IORef Int -> Int -> IO ()
forwardWithLimit StreamIO
src StreamIO
dst IORef Int
countRef Int
limit = IO ()
go
  where
    go :: IO ()
go = do
      count <- IORef Int -> IO Int
forall a. IORef a -> IO a
readIORef IORef Int
countRef
      if count >= limit
        then pure ()  -- limit reached, stop forwarding
        else do
          b <- streamReadByte src
          modifyIORef' countRef (+ 1)
          streamWrite dst (BS.singleton b)
          go

-- | Build a relay multiaddr in binary format.
-- Format: <relayAddr>/p2p/<relayId>/p2p-circuit/p2p/<targetId>
buildRelayAddrBytes :: ByteString -> ByteString -> ByteString -> ByteString
buildRelayAddrBytes :: ByteString -> ByteString -> ByteString -> ByteString
buildRelayAddrBytes ByteString
relayAddr ByteString
relayIdBytes ByteString
targetIdBytes =
  ByteString
relayAddr
  ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString -> ByteString
p2pProtocolBytes ByteString
relayIdBytes
  ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
p2pCircuitBytes
  ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString -> ByteString
p2pProtocolBytes ByteString
targetIdBytes
  where
    -- P2P protocol: code 421 (0xa503) + varint(len) + peer ID bytes
    p2pProtocolBytes :: ByteString -> ByteString
    p2pProtocolBytes :: ByteString -> ByteString
p2pProtocolBytes ByteString
pid = Word64 -> ByteString
encodeUvarint Word64
421 ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> Word64 -> ByteString
encodeUvarint (Int -> Word64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (ByteString -> Int
BS.length ByteString
pid)) ByteString -> ByteString -> ByteString
forall a. Semigroup a => a -> a -> a
<> ByteString
pid

    -- P2PCircuit protocol: code 290 (0xa202), no address
    p2pCircuitBytes :: ByteString
    p2pCircuitBytes :: ByteString
p2pCircuitBytes = Word64 -> ByteString
encodeUvarint Word64
290

-- | Check whether a multiaddr contains a p2p-circuit component.
isRelayedAddr :: Multiaddr -> Bool
isRelayedAddr :: Multiaddr -> Bool
isRelayedAddr (Multiaddr [Protocol]
ps) = Protocol
P2PCircuit Protocol -> [Protocol] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [Protocol]
ps

-- | Check whether raw multiaddr bytes describe a relayed connection.
-- Decodes the bytes structurally: the p2p-circuit byte pattern occurring
-- inside another component (e.g. a peer ID) does not count, unlike the
-- previous substring heuristic. Undecodable bytes are not relayed.
isRelayedConnection :: ByteString -> Bool
isRelayedConnection :: ByteString -> Bool
isRelayedConnection = (String -> Bool)
-> (Multiaddr -> Bool) -> Either String Multiaddr -> Bool
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Bool -> String -> Bool
forall a b. a -> b -> a
const Bool
False) Multiaddr -> Bool
isRelayedAddr (Either String Multiaddr -> Bool)
-> (ByteString -> Either String Multiaddr) -> ByteString -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ByteString -> Either String Multiaddr
fromBytes