{-# LANGUAGE AllowAmbiguousTypes       #-}
{-# LANGUAGE ExistentialQuantification #-}
{-# LANGUAGE RecordWildCards           #-}

module Pos.Infra.Network.Types
       ( -- * Network configuration
         NetworkConfig (..)
       , NodeName (..)
       -- * Topology
       , StaticPeers(..)
       , Topology(..)
       -- ** Derived information
       , SubscriptionWorker(..)
       , topologyNodeType
       , topologySubscribers
       , topologyUnknownNodeType
       , topologySubscriptionWorker
       , topologyRunKademlia
       , topologyEnqueuePolicy
       , topologyDequeuePolicy
       , topologyFailurePolicy
       , topologyMaxBucketSize
       , topologyHealthStatus
       , topologyRoute53HealthCheckEnabled
       -- * Queue initialization
       , Bucket(..)
       , initQueue
       -- * Constructing peers
       , Valency
       , Fallbacks
       , choosePeers
       -- * DNS support
       , Resolver
       , resolveDnsDomains
       , initDnsOnUse
       -- * Helpers
       , HasNodeType (..)
       , getNodeTypeDefault
       -- * Re-exports
       -- ** from .DnsDomains
       , DnsDomains(..)
       -- ** from time-warp
       , NodeType (..)
       , MsgType (..)
       , Origin (..)
       -- ** other
       , NodeId (..)
       ) where

import           Universum

import           Data.Aeson
import           Data.Aeson.Encoding.Internal as A
import           Data.IP (IPv4)
import qualified Data.Set as Set (null)
import qualified Data.Text as T
import           Network.Broadcast.OutboundQueue (OutboundQ)
import qualified Network.Broadcast.OutboundQueue as OQ
import           Network.Broadcast.OutboundQueue.Types
import           Network.DNS (DNSError)
import qualified Network.DNS as DNS
import qualified Network.Transport.TCP as TCP
import           Node.Internal (NodeId (..))
import qualified Prelude
import qualified System.Metrics as Monitoring

import           Pos.Core.Metrics.Constants (cardanoNamespace)
import           Pos.Infra.Network.DnsDomains (DnsDomains (..), NodeAddr)
import qualified Pos.Infra.Network.DnsDomains as DnsDomains
import qualified Pos.Infra.Network.Policy as Policy
import           Pos.Infra.Reporting.Health.Types (HealthStatus (..))
import           Pos.Infra.Util.TimeWarp (addressToNodeId)
import           Pos.Util.Trace (Severity, Trace)
import           Pos.Util.Trace.Named (LogNamed, appendName, named)
import           Pos.Util.Util (HasLens', lensOf)

{-------------------------------------------------------------------------------
  Network configuration
-------------------------------------------------------------------------------}

newtype NodeName = NodeName Text
    deriving (Show, Generic, Ord, Eq, IsString)

instance ToJSON NodeName where
    toJSON (NodeName name) = String name

instance ToJSONKey NodeName where
  toJSONKey = ToJSONKeyText f g
    where f = T.pack . toString
          g = A.text . T.pack . toString

instance FromJSON NodeName where
    parseJSON = fmap NodeName . parseJSON

instance ToString NodeName where
    toString (NodeName txt) = toString txt

-- | Information about the network in which a node participates.
data NetworkConfig kademlia = NetworkConfig
    { ncTopology      :: !(Topology kademlia)
      -- ^ Network topology from the point of view of the current node
    , ncDefaultPort   :: !Word16
      -- ^ Port number to use when translating IP addresses to NodeIds
    , ncSelfName      :: !(Maybe NodeName)
      -- ^ Our node name (if known)
    , ncEnqueuePolicy :: !(OQ.EnqueuePolicy NodeId)
    , ncDequeuePolicy :: !OQ.DequeuePolicy
    , ncFailurePolicy :: !(OQ.FailurePolicy NodeId)
    , ncTcpAddr       :: !TCP.TCPAddr
      -- ^ External TCP address of the node.
      -- It encapsulates both bind address and address visible to other nodes.
    }

instance Show kademlia => Show (NetworkConfig kademlia) where
    show = show . showableNetworkConfig

data ShowableNetworkConfig kademlia = ShowableNetworkConfig
    { sncTopology    :: !(Topology kademlia)
    , sncDefaultPort :: !Word16
    , sncSelfName    :: !(Maybe NodeName)
    } deriving (Show)

showableNetworkConfig :: NetworkConfig kademlia -> ShowableNetworkConfig kademlia
showableNetworkConfig NetworkConfig {..} =
    let sncTopology    = ncTopology
        sncDefaultPort = ncDefaultPort
        sncSelfName    = ncSelfName
    in  ShowableNetworkConfig {..}

----------------------------------------------------------------------------
-- Topology
----------------------------------------------------------------------------

-- | Statically configured peers
--
-- Although the peers are statically configured, this is nonetheless stateful
-- because we re-read the file on SIGHUP.
data StaticPeers = StaticPeers {
      -- | Register a handler to be invoked whenever the static peers change
      --
      -- The handler will also be called on registration
      -- (with the current value).
      staticPeersOnChange   :: (Peers NodeId -> IO ()) -> IO ()
    , -- | Monitoring worker which is supposed to be started in a
      -- separate thread. This worker processes handlers registered by
      -- 'staticPeersOnChange'.
      staticPeersMonitoring :: IO ()
    }

instance Show StaticPeers where
    show _ = "<<StaticPeers>>"

-- | Topology of the network, from the point of view of the current node
data Topology kademlia =
    -- | All peers of the node have been statically configured
    --
    -- This is used for core and relay nodes
    TopologyCore {
        topologyStaticPeers :: !StaticPeers
      , topologyOptKademlia :: !(Maybe kademlia)
      }

  | TopologyRelay {
        -- We can use static peers or dynamic subscriptions.
        -- We typically use on or the other but not both.
        topologyStaticPeers :: !StaticPeers
      , topologyDnsDomains  :: !(DnsDomains DNS.Domain)
      , topologyValency     :: !Valency
      , topologyFallbacks   :: !Fallbacks
      , topologyOptKademlia :: !(Maybe kademlia)
      , topologyMaxSubscrs  :: !OQ.MaxBucketSize
      }

    -- | We discover our peers through DNS
    --
    -- This is used for behind-NAT nodes.
  | TopologyBehindNAT {
        topologyValency    :: !Valency
      , topologyFallbacks  :: !Fallbacks
      , topologyDnsDomains :: !(DnsDomains DNS.Domain)
      }

    -- | We discover our peers through Kademlia
  | TopologyP2P {
        topologyValency    :: !Valency
      , topologyFallbacks  :: !Fallbacks
      , topologyKademlia   :: !kademlia
      , topologyMaxSubscrs :: !OQ.MaxBucketSize
      }

    -- | We discover our peers through Kademlia, and every node in the network
    -- is a core node.
  | TopologyTraditional {
        topologyValency    :: !Valency
      , topologyFallbacks  :: !Fallbacks
      , topologyKademlia   :: !kademlia
      , topologyMaxSubscrs :: !OQ.MaxBucketSize
      }

    -- | Auxx simulates "real" edge nodes, but is configured with
    -- a static set of relays.
  | TopologyAuxx {
        topologyRelays :: ![NodeId]
      }
  deriving (Show)

----------------------------------------------------------------------------
-- Information derived from the topology
----------------------------------------------------------------------------

-- See the networking policy document for background to understand this
-- docs/network/policy.md

-- | Derive node type from its topology
topologyNodeType :: Topology kademlia -> NodeType
topologyNodeType TopologyCore{}        = NodeCore
topologyNodeType TopologyRelay{}       = NodeRelay
topologyNodeType TopologyBehindNAT{}   = NodeEdge
topologyNodeType TopologyP2P{}         = NodeEdge
topologyNodeType TopologyTraditional{} = NodeCore
topologyNodeType TopologyAuxx{}        = NodeEdge

-- | Type class which encapsulates something that has 'NodeType'.
class HasNodeType ctx where
    -- | Extract 'NodeType' from a context. It's not a 'Lens', because
    -- the intended usage is that context has topology and it's not
    -- possible to 'set' node type in topology.
    getNodeType :: ctx -> NodeType

-- | Implementation of 'getNodeType' for something that has
-- 'NetworkConfig' inside.
getNodeTypeDefault ::
       forall kademlia ctx. HasLens' ctx (NetworkConfig kademlia)
    => ctx
    -> NodeType
getNodeTypeDefault =
    topologyNodeType . ncTopology <$> view (lensOf @(NetworkConfig kademlia))

-- | Assumed type and maximum number of subscribers (if subscription is allowed)
--
-- Note that the 'TopologyRelay' case covers /both/ priviledged and
-- unpriviledged relays. See the networking policy document for full details of
-- why this makes sense or works.
topologySubscribers :: Topology kademlia -> Maybe (NodeType, OQ.MaxBucketSize)
topologySubscribers TopologyCore{}          = Nothing
topologySubscribers TopologyRelay{..}       = Just (NodeEdge, topologyMaxSubscrs)
topologySubscribers TopologyBehindNAT{}     = Nothing
topologySubscribers TopologyP2P{..}         = Just (NodeRelay, topologyMaxSubscrs)
topologySubscribers TopologyTraditional{..} = Just (NodeCore, topologyMaxSubscrs)
topologySubscribers TopologyAuxx{}          = Nothing

-- | Assumed type for unknown nodes
topologyUnknownNodeType :: Topology kademlia -> OQ.UnknownNodeType NodeId
topologyUnknownNodeType topology = OQ.UnknownNodeType $ go topology
  where
    go :: Topology kademlia -> NodeId -> NodeType
    go TopologyCore{..}      = const NodeRelay  -- to allow dynamic reconfig
    go TopologyRelay{..}     = const NodeEdge   -- since we don't trust anyone
    go TopologyTraditional{} = const NodeCore
    go TopologyP2P{}         = const NodeRelay  -- a fairly normal expected case
    go TopologyBehindNAT{}   = const NodeEdge   -- should never happen
    go TopologyAuxx{}        = const NodeEdge   -- should never happen

data SubscriptionWorker =
    SubscriptionWorkerBehindNAT (DnsDomains DNS.Domain)
  | SubscriptionWorkerKademlia NodeType Valency Fallbacks

-- | What kind of subscription worker do we run?
topologySubscriptionWorker :: Topology kademlia -> Maybe SubscriptionWorker
topologySubscriptionWorker = go
  where
    go TopologyCore{}          = Nothing
    go TopologyRelay{topologyDnsDomains = DnsDomains []}
                               = Nothing
    go TopologyRelay{..}       = Just $ SubscriptionWorkerBehindNAT
                                          topologyDnsDomains
    go TopologyBehindNAT{..}   = Just $ SubscriptionWorkerBehindNAT
                                          topologyDnsDomains
    go TopologyP2P{..}         = Just $ SubscriptionWorkerKademlia
                                          NodeRelay
                                          topologyValency
                                          topologyFallbacks
    go TopologyTraditional{..} = Just $ SubscriptionWorkerKademlia
                                          NodeCore
                                          topologyValency
                                          topologyFallbacks
    go TopologyAuxx{}   = Nothing

-- | Should we register to the Kademlia network? If so, is it essential that we
-- successfully join it (contact at least one existing peer)? Second component
-- is 'True' if yes.
topologyRunKademlia :: Topology kademlia -> Maybe (kademlia, Bool)
topologyRunKademlia = go
  where
    go TopologyCore{..}        = flip (,) False <$> topologyOptKademlia
    go TopologyRelay{..}       = flip (,) False <$> topologyOptKademlia
    go TopologyBehindNAT{}     = Nothing
    go TopologyP2P{..}         = Just (topologyKademlia, True)
    go TopologyTraditional{..} = Just (topologyKademlia, True)
    go TopologyAuxx{}          = Nothing

-- | Enqueue policy for the given topology
topologyEnqueuePolicy :: Topology kademia -> OQ.EnqueuePolicy NodeId
topologyEnqueuePolicy = go
  where
    go TopologyCore{}        = Policy.defaultEnqueuePolicyCore
    go TopologyRelay{}       = Policy.defaultEnqueuePolicyRelay
    go TopologyBehindNAT{..} = Policy.defaultEnqueuePolicyEdgeBehindNat
    go TopologyP2P{}         = Policy.defaultEnqueuePolicyEdgeP2P
    go TopologyTraditional{} = Policy.defaultEnqueuePolicyCore
    go TopologyAuxx{}        = Policy.defaultEnqueuePolicyAuxx

-- | Dequeue policy for the given topology
topologyDequeuePolicy :: Topology kademia -> OQ.DequeuePolicy
topologyDequeuePolicy = go
  where
    go TopologyCore{}        = Policy.defaultDequeuePolicyCore
    go TopologyRelay{}       = Policy.defaultDequeuePolicyRelay
    go TopologyBehindNAT{..} = Policy.defaultDequeuePolicyEdgeBehindNat
    go TopologyP2P{}         = Policy.defaultDequeuePolicyEdgeP2P
    go TopologyTraditional{} = Policy.defaultDequeuePolicyCore
    go TopologyAuxx{}        = Policy.defaultDequeuePolicyAuxx

-- | Failure policy for the given topology
topologyFailurePolicy :: Topology kademia -> OQ.FailurePolicy NodeId
topologyFailurePolicy = go
  where
    go TopologyAuxx{} = Policy.defaultFailurePolicyAuxx
    go topo           = Policy.defaultFailurePolicy . topologyNodeType $ topo

-- | Maximum bucket size
topologyMaxBucketSize :: Topology kademia -> Bucket -> Maybe OQ.MaxBucketSize
topologyMaxBucketSize topology bucket =
    case bucket of
      BucketSubscriptionListener ->
        case topologySubscribers topology of
          Just (_subscriberType, maxBucketSize) -> Just maxBucketSize
          Nothing                               -> Nothing -- subscription not allowed
      _otherBucket -> Just OQ.BucketSizeUnlimited

topologyHealthStatus :: MonadIO m => Topology kademlia -> OutboundQ msg nid Bucket -> m HealthStatus
topologyHealthStatus topology = case topology of
    TopologyCore{}        -> const (pure topologyHealthStatusCore)
    TopologyRelay{..}     -> topologyHealthStatusRelay topologyMaxSubscrs
    TopologyBehindNAT{}   -> topologyHealthStatusNAT
    TopologyP2P{}         -> topologyHealthStatusP2P
    TopologyTraditional{} -> topologyHealthStatusTraditional
    TopologyAuxx{}        -> topologyHealthStatusAuxx

-- | Core nodes are always healthy.
topologyHealthStatusCore :: HealthStatus
topologyHealthStatusCore = HSHealthy ""

-- | Health of a relay is determined by its spare capacity.
topologyHealthStatusRelay
    :: MonadIO m
    => OQ.MaxBucketSize
    -> OQ.OutboundQ msg nid Bucket
    -> m HealthStatus
topologyHealthStatusRelay mbs oq = do
    let maxCapacityText :: Text
        maxCapacityText = case mbs of
            OQ.BucketSizeUnlimited -> fromString "unlimited"
            OQ.BucketSizeMax x     -> fromString (show x)
    spareCapacity <- liftIO $ OQ.bucketSpareCapacity oq BucketSubscriptionListener
    pure $ case spareCapacity of
        OQ.SpareCapacity sc  | sc == 0 -> HSUnhealthy (fromString "0/" <> maxCapacityText)
        OQ.SpareCapacity sc  -> HSHealthy $ fromString (show sc) <> "/" <> maxCapacityText
        OQ.UnlimitedCapacity -> HSHealthy maxCapacityText

-- | Health of a behind-NAT node is good iff it is connected to some other node.
topologyHealthStatusNAT
    :: MonadIO m
    => OQ.OutboundQ msg nid bucket
    -> m HealthStatus
topologyHealthStatusNAT oq = do
    peers <- liftIO $ OQ.getAllPeers oq
    if (Set.null (peersSet peers))
    then pure $ HSUnhealthy "not connected"
    else pure $ HSHealthy "connected"

-- | Health status of a P2P node is the same as for behind NAT. Its capacity to
-- support new subscribers is ignored.
topologyHealthStatusP2P
    :: MonadIO m
    => OQ.OutboundQ msg nid bucket
    -> m HealthStatus
topologyHealthStatusP2P = topologyHealthStatusNAT

-- | Health status of a traditional node is the same as for behind NAT. Its
-- capacity to support new subscribers is ignored.
topologyHealthStatusTraditional
    :: MonadIO m
    => OQ.OutboundQ msg nid bucket
    -> m HealthStatus
topologyHealthStatusTraditional = topologyHealthStatusTraditional

-- | Auxx health status is the same as for behind-NAT.
topologyHealthStatusAuxx
    :: MonadIO m
    => OQ.OutboundQ msg nid bucket
    -> m HealthStatus
topologyHealthStatusAuxx = topologyHealthStatusNAT

-- | Whether or not we want to enable the health-check endpoint to be used by Route53
-- in determining if a relay is healthy (i.e. it can accept more subscriptions or not)
topologyRoute53HealthCheckEnabled :: Topology kademia -> Bool
topologyRoute53HealthCheckEnabled = go
  where
    go TopologyCore{}        = False
    go TopologyRelay{}       = True
    go TopologyBehindNAT{..} = False
    go TopologyP2P{}         = False
    go TopologyTraditional{} = False
    go TopologyAuxx{}        = False

{-------------------------------------------------------------------------------
  Queue initialization
-------------------------------------------------------------------------------}

-- | The various buckets we use for the outbound queue
data Bucket =
    -- | Bucket for nodes we add statically
    BucketStatic

    -- | Bucket for nodes added by the behind-NAT worker
  | BucketBehindNatWorker

    -- | Bucket for nodes added by the Kademlia worker
  | BucketKademliaWorker

    -- | Bucket for nodes added by the subscription listener
  | BucketSubscriptionListener
  deriving (Eq, Ord, Enum, Bounded, Show)

-- | Initialize the outbound queue based on the network configuration
--
-- We add all statically known peers to the queue, so that we know to send
-- messages to those peers. This is relevant only for core nodes and
-- auxx. In the former case, those core nodes will in turn add this
-- core node to /their/ outbound queue because this node would equally be
-- a statically known peer; in the latter case, auxx is not expected
-- to receive any messages so messages in the reverse direction don't matter.
--
-- For behind NAT nodes and Kademlia nodes (P2P or traditional) we start
-- (elsewhere) specialized workers that add peers to the queue and subscribe
-- to (some of) those peers.
--
-- This will use the log-warper trace for logging from the outbound queue.
-- You can choose what name to give it.
initQueue :: (FormatMsg msg)
          => NetworkConfig kademlia
          -> Trace IO (LogNamed (Severity, Text))
          -> Maybe Monitoring.Store -- ^ EKG store (if used)
          -> IO (OutboundQ msg NodeId Bucket)
initQueue NetworkConfig{..} logTrace mStore = liftIO $ do
    let NodeName selfName = fromMaybe (NodeName "self") ncSelfName
        oqTrace = named (appendName selfName logTrace)
    oq <- OQ.new oqTrace
                 ncEnqueuePolicy
                 ncDequeuePolicy
                 ncFailurePolicy
                 (fromMaybe (OQ.BucketSizeMax 0) . topologyMaxBucketSize ncTopology)
                 (topologyUnknownNodeType ncTopology)

    case mStore of
      Nothing    -> return () -- EKG store not used
      Just store -> OQ.registerQueueMetrics (Just (toString cardanoNamespace)) oq store

    case ncTopology of
      TopologyAuxx peers -> do
        let peers' = simplePeers $ map (NodeRelay, ) peers
        void $ OQ.updatePeersBucket oq BucketStatic (\_ -> peers')
      TopologyBehindNAT{} ->
        -- subscription worker is responsible for adding peers
        return ()
      TopologyP2P{} ->
        -- Kademlia worker is responsible for adding peers
        return ()
      TopologyTraditional{} ->
        -- Kademlia worker is responsible for adding peers
        return ()
      TopologyCore{topologyStaticPeers = StaticPeers{..}} ->
        staticPeersOnChange $ \peers -> do
          OQ.clearRecentFailures oq
          void $ OQ.updatePeersBucket oq BucketStatic (\_ -> peers)
      TopologyRelay{topologyStaticPeers = StaticPeers{..}} ->
        staticPeersOnChange $ \peers -> do
          OQ.clearRecentFailures oq
          void $ OQ.updatePeersBucket oq BucketStatic (\_ -> peers)

    return oq

----------------------------------------------------------------------------
-- Constructing peers
----------------------------------------------------------------------------

-- | The number of peers we want to send to
--
-- In other words, this should correspond to the length of the outermost lists
-- in the OutboundQueue's 'Peers' data structure.
type Valency = Int

-- | The number of fallbacks for each peer we want to send to
--
-- In other words, this should corresponding to one less than the length of the
-- innermost lists in the OutboundQueue's 'Peers' data structure.
type Fallbacks = Int

-- | Construct 'Peers' from a set of potential peer nodes.
choosePeers :: Valency -> Fallbacks -> NodeType -> [NodeId] -> Peers NodeId
choosePeers valency fallbacks peerType =
      peersFromList mempty
    . fmap ((,) peerType)
    . transpose
    . take (1 + fallbacks)
    . mkGroupsOf valency
  where
    mkGroupsOf :: Int -> [a] -> [[a]]
    mkGroupsOf _ []  = []
    mkGroupsOf n lst = case splitAt n lst of
                         (these, those) -> these : mkGroupsOf n those

----------------------------------------------------------------------------
-- DNS support
----------------------------------------------------------------------------

type Resolver = DNS.Domain -> IO (Either DNSError [IPv4])

-- | Variation on resolveDnsDomains that returns node IDs
--
-- This uses the network, and so may throw exceptions in case of, for instance,
-- and unreachable network.
resolveDnsDomains :: Word16
                  -> [NodeAddr DNS.Domain]
                  -> IO [Either DNSError [NodeId]]
resolveDnsDomains defaultPort dnsDomains =
    initDnsOnUse $ \resolve -> (fmap . fmap . fmap . fmap) addressToNodeId $
        DnsDomains.resolveDnsDomains resolve
                                     defaultPort
                                     dnsDomains
{-# ANN resolveDnsDomains ("HLint: ignore Use <$>" :: String) #-}


-- | Initialize the DNS library whenever it's used
--
-- This isn't great for performance but it means that we do not initialize it
-- when we need it; initializing it once only on demand is possible but requires
-- jumping through too many hoops.
initDnsOnUse :: (Resolver -> IO a) -> IO a
initDnsOnUse k = k $ \dom -> do
    resolvSeed <- DNS.makeResolvSeed DNS.defaultResolvConf
    DNS.withResolver resolvSeed $ \resolver ->
      DNS.lookupA resolver dom
