<commit_before>module Vere.Ames (ames) where

import UrbitPrelude

import Arvo                      hiding (Fake)
import Network.Socket            hiding (recvFrom, sendTo)
import Network.Socket.ByteString
import Vere.Pier.Types

import qualified Data.ByteString as BS
import qualified Data.Map        as M
import qualified Urbit.Ob        as Ob
import qualified Urbit.Time      as Time

-- Types -----------------------------------------------------------------------

data AmesDrv = AmesDrv
  { aIsLive        :: IORef Bool
  , aTurfs         :: TVar [Turf]
  , aGalaxies      :: IORef (M.Map Galaxy (Async (), TQueue ByteString))
  , aSocketSend    :: Socket
  , aSocketRecv    :: Socket
  , aWakeTimer     :: Async ()
  , aListener      :: Async ()
  , aSendingQueue  :: TQueue (SockAddr, ByteString)
  , aSendingThread :: Async ()
  }

data NetworkMode = Fake | Real
  deriving (Eq, Ord, Show)

-- data GalaxyInfo = GalaxyInfo { ip :: Ipv4, age :: Time.Unix }
--   deriving (Eq, Ord, Show)


-- Utils -----------------------------------------------------------------------

galaxyPort :: NetworkMode -> Galaxy -> PortNumber
galaxyPort Fake (Galaxy g) = fromIntegral g + 31337
galaxyPort Real (Galaxy g) = fromIntegral g + 13337

listenPort :: NetworkMode -> Ship -> PortNumber
listenPort m s | s < 256 = galaxyPort m (fromIntegral s)
listenPort m _ = 0

localhost :: HostAddress
localhost = tupleToHostAddress (127,0,0,1)

okayFakeAddr :: AmesDest -> Bool
okayFakeAddr = \case
    ADGala _ _          -> True
    ADIpv4 _ p (Ipv4 a) -> a == localhost

destSockAddr :: NetworkMode -> AmesDest -> SockAddr
destSockAddr m = \case
    ADGala _ g   -> SockAddrInet (galaxyPort m g) localhost
    ADIpv4 _ p a -> SockAddrInet (fromIntegral p) (unIpv4 a)

barnEv :: KingId -> Ev
barnEv inst =
  EvBlip $ BlipEvNewt $ NewtEvBarn (fromIntegral inst, ()) ()

wakeEv :: Ev
wakeEv =
  EvBlip $ BlipEvAmes $ AmesEvWake () ()

hearEv :: Time.Wen -> PortNumber -> HostAddress -> ByteString -> Ev
hearEv w p a bs =
    EvBlip $ BlipEvAmes $ AmesEvHear () dest (MkBytes bs)
  where
    dest = ADIpv4 w (fromIntegral p) (Ipv4 a)

_turfText :: Turf -> Text
_turfText = intercalate "." . reverse . fmap unCord . unTurf


--------------------------------------------------------------------------------

{-
    inst      -- Process instance number.
    who       -- Which ship are we?
    enqueueEv -- Queue-event action.
    mPort     -- Explicit port override from command line arguments.

    We ignore the %turf arguments for now. We only have fake ships,
    so we don't implement the DNS stuff yet.

    TODO Handle socket exceptions in waitPacket

    4096 is a reasonable number for recvFrom. Packets of that size are
    not possible on the internet.

    TODO log when `sendTo` sent fewer bytes than requested.

    TODO verify that the KingIds match on effects.
-}
ames :: forall e. HasLogFunc e
     => KingId -> Ship -> Maybe Port -> QueueEv
     -> ([Ev], RAcquire e (EffCb e NewtEf))
ames inst who mPort enqueueEv =
    (initialEvents, runAmes)
  where
    initialEvents :: [Ev]
    initialEvents = [barnEv inst]

    runAmes :: RAcquire e (EffCb e NewtEf)
    runAmes = do
        drv <- mkRAcquire start stop
        pure (handleEffect drv)

    start :: RIO e AmesDrv
    start = do
        aIsLive        <- newIORef False
        aTurfs         <- newTVarIO []
        aGalaxies      <- newIORef mempty
        aSocketSend    <- io $ bindSendSock
        aSocketRecv    <- bindRecvSock
        aWakeTimer     <- async runTimer
        aListener      <- async (waitPacket aSocketRecv)
        aSendingQueue  <- newTQueueIO
        aSendingThread <- async (sendingThread aSendingQueue aSocketSend)
        pure            $ AmesDrv{..}

    -- TODO: This switch needs to reflect the network mode flag; are we a fake
    -- network? Do we have all networking disabled?
    netMode :: NetworkMode
    netMode = Fake

    stop :: AmesDrv -> RIO e ()
    stop AmesDrv{..} = do
        galaxies <- readIORef aGalaxies
        mapM_ (cancel . fst) galaxies

        cancel aSendingThread
        cancel aWakeTimer
        cancel aListener
        io $ close' aSocketRecv

    runTimer :: RIO e ()
    runTimer = forever $ do
        threadDelay (300 * 1000000) -- 300 seconds
        atomically (enqueueEv wakeEv)

    bindSendSock :: IO Socket
    bindSendSock = socket AF_INET Datagram defaultProtocol

    bindRecvSock :: RIO e Socket
    bindRecvSock = do
        let ourPort = maybe (listenPort netMode who) fromIntegral mPort
        s  <- io $ socket AF_INET Datagram defaultProtocol
        logTrace $ displayShow ("(ames) Binding to port ", ourPort)
        () <- io $ bind s (SockAddrInet ourPort localhost)
        pure s

    waitPacket :: Socket -> RIO e ()
    waitPacket s = forever $ do
        (bs, addr) <- io $ recvFrom s 4096
        logTrace $ displayShow ("(ames) Received packet from ", addr)
        wen        <- io $ Time.now
        case addr of
            SockAddrInet p a -> atomically (enqueueEv $ hearEv wen p a bs)
            _                -> pure ()

    handleEffect :: AmesDrv -> NewtEf -> RIO e ()
    handleEffect drv@AmesDrv{..} = \case
      NewtEfTurf (_id, ()) turfs -> do
          writeIORef aIsLive True
          atomically $ writeTVar aTurfs turfs

      NewtEfSend (_id, ()) dest (MkBytes bs) -> do
          whenM (readIORef aIsLive) (sendPacket drv netMode dest bs)

    sendPacket :: AmesDrv -> NetworkMode -> AmesDest -> ByteString -> RIO e ()

    sendPacket AmesDrv{..} Fake dest bs = do<commit_msg>Use the same socket for sending and receiving.<commit_after>module Vere.Ames (ames) where

import UrbitPrelude

import Arvo                      hiding (Fake)
import Network.Socket            hiding (recvFrom, sendTo)
import Network.Socket.ByteString
import Vere.Pier.Types

import qualified Data.ByteString as BS
import qualified Data.Map        as M
import qualified Urbit.Ob        as Ob
import qualified Urbit.Time      as Time

-- Types -----------------------------------------------------------------------

data AmesDrv = AmesDrv
  { aIsLive        :: IORef Bool
  , aTurfs         :: TVar [Turf]
  , aGalaxies      :: IORef (M.Map Galaxy (Async (), TQueue ByteString))
  , aSocket        :: Socket
  , aWakeTimer     :: Async ()
  , aListener      :: Async ()
  , aSendingQueue  :: TQueue (SockAddr, ByteString)
  , aSendingThread :: Async ()
  }

data NetworkMode = Fake | Real
  deriving (Eq, Ord, Show)

-- data GalaxyInfo = GalaxyInfo { ip :: Ipv4, age :: Time.Unix }
--   deriving (Eq, Ord, Show)


-- Utils -----------------------------------------------------------------------

galaxyPort :: NetworkMode -> Galaxy -> PortNumber
galaxyPort Fake (Galaxy g) = fromIntegral g + 31337
galaxyPort Real (Galaxy g) = fromIntegral g + 13337

listenPort :: NetworkMode -> Ship -> PortNumber
listenPort m s | s < 256 = galaxyPort m (fromIntegral s)
listenPort m _ = 0

localhost :: HostAddress
localhost = tupleToHostAddress (127,0,0,1)

okayFakeAddr :: AmesDest -> Bool
okayFakeAddr = \case
    ADGala _ _          -> True
    ADIpv4 _ p (Ipv4 a) -> a == localhost

destSockAddr :: NetworkMode -> AmesDest -> SockAddr
destSockAddr m = \case
    -- As mentioned previously, "localhost" is wrong.
    ADGala _ g   -> SockAddrInet (galaxyPort m g) localhost
    ADIpv4 _ p a -> SockAddrInet (fromIntegral p) (unIpv4 a)

barnEv :: KingId -> Ev
barnEv inst =
  EvBlip $ BlipEvNewt $ NewtEvBarn (fromIntegral inst, ()) ()

wakeEv :: Ev
wakeEv =
  EvBlip $ BlipEvAmes $ AmesEvWake () ()

hearEv :: Time.Wen -> PortNumber -> HostAddress -> ByteString -> Ev
hearEv w p a bs =
    EvBlip $ BlipEvAmes $ AmesEvHear () dest (MkBytes bs)
  where
    dest = ADIpv4 w (fromIntegral p) (Ipv4 a)

_turfText :: Turf -> Text
_turfText = intercalate "." . reverse . fmap unCord . unTurf


--------------------------------------------------------------------------------

{-
    inst      -- Process instance number.
    who       -- Which ship are we?
    enqueueEv -- Queue-event action.
    mPort     -- Explicit port override from command line arguments.

    We ignore the %turf arguments for now. We only have fake ships,
    so we don't implement the DNS stuff yet.

    TODO Handle socket exceptions in waitPacket

    4096 is a reasonable number for recvFrom. Packets of that size are
    not possible on the internet.

    TODO log when `sendTo` sent fewer bytes than requested.

    TODO verify that the KingIds match on effects.
-}
ames :: forall e. HasLogFunc e
     => KingId -> Ship -> Maybe Port -> QueueEv
     -> ([Ev], RAcquire e (EffCb e NewtEf))
ames inst who mPort enqueueEv =
    (initialEvents, runAmes)
  where
    initialEvents :: [Ev]
    initialEvents = [barnEv inst]

    runAmes :: RAcquire e (EffCb e NewtEf)
    runAmes = do
        drv <- mkRAcquire start stop
        pure (handleEffect drv)

    start :: RIO e AmesDrv
    start = do
        aIsLive        <- newIORef False
        aTurfs         <- newTVarIO []
        aGalaxies      <- newIORef mempty
        aSocket        <- bindSock
        aWakeTimer     <- async runTimer
        aListener      <- async (waitPacket aSocket)
        aSendingQueue  <- newTQueueIO
        aSendingThread <- async (sendingThread aSendingQueue aSocket)
        pure            $ AmesDrv{..}

    -- TODO: This switch needs to reflect the network mode flag; are we a fake
    -- network? Do we have all networking disabled?
    netMode :: NetworkMode
    netMode = Fake

    stop :: AmesDrv -> RIO e ()
    stop AmesDrv{..} = do
        galaxies <- readIORef aGalaxies
        mapM_ (cancel . fst) galaxies

        cancel aSendingThread
        cancel aWakeTimer
        cancel aListener
        io $ close' aSocket

    runTimer :: RIO e ()
    runTimer = forever $ do
        threadDelay (300 * 1000000) -- 300 seconds
        atomically (enqueueEv wakeEv)

    bindSock :: RIO e Socket
    bindSock = do
        let ourPort = maybe (listenPort netMode who) fromIntegral mPort
        s  <- io $ socket AF_INET Datagram defaultProtocol
        logTrace $ displayShow ("(ames) Binding to port ", ourPort)
        -- localhost may be wrong here; ames.c switches between INADDR_ANY and
        -- INADDR_LOOPBACK based on networking flags.
        () <- io $ bind s (SockAddrInet ourPort localhost)
        pure s

    waitPacket :: Socket -> RIO e ()
    waitPacket s = forever $ do
        (bs, addr) <- io $ recvFrom s 4096
        logTrace $ displayShow ("(ames) Received packet from ", addr)
        wen        <- io $ Time.now
        case addr of
            SockAddrInet p a -> atomically (enqueueEv $ hearEv wen p a bs)
            _                -> pure ()

    handleEffect :: AmesDrv -> NewtEf -> RIO e ()
    handleEffect drv@AmesDrv{..} = \case
      NewtEfTurf (_id, ()) turfs -> do
          writeIORef aIsLive True
          atomically $ writeTVar aTurfs turfs

      NewtEfSend (_id, ()) dest (MkBytes bs) -> do
          whenM (readIORef aIsLive) (sendPacket drv netMode dest bs)

    sendPacket :: AmesDrv -> NetworkMode -> AmesDest -> ByteString -> RIO e ()

    sendPacket AmesDrv{..} Fake dest bs = do
{'fraction_non_alphanumeric': 0.1257743064907083, 'fraction_numerical': 0.008618367896579585, 'mean_word_length': 2.036249659307713, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 71, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5363610', 'n_tokens_mistral': 3589, 'n_tokens_neox': 3397, 'n_words': 1138}