From fea9660f80084c4c6ec31ea8500d0592ab272507 Mon Sep 17 00:00:00 2001
From: Trolli Schmittlauch <t.schmittlauch@orlives.de>
Date: Sat, 30 May 2020 01:09:17 +0200
Subject: [PATCH] WIP: start implementing incoming request handling

---
 src/Hash2Pub/DHTProtocol.hs | 28 ++++++++++++++++++++++++++++
 src/Hash2Pub/FediChord.hs   | 30 +++++++++++++++++++++---------
 src/Hash2Pub/Main.hs        |  2 +-
 3 files changed, 50 insertions(+), 10 deletions(-)

diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs
index 34423a4..fba3f20 100644
--- a/src/Hash2Pub/DHTProtocol.hs
+++ b/src/Hash2Pub/DHTProtocol.hs
@@ -4,6 +4,7 @@ module Hash2Pub.DHTProtocol
     , addCacheEntry
     , addCacheEntryPure
     , deleteCacheEntry
+    , deserialiseMessage
     , markCacheEntryAsVerified
     , RemoteCacheEntry(..)
     , toRemoteCacheEntry
@@ -19,6 +20,7 @@ module Hash2Pub.DHTProtocol
     , resolve
     , mkSendSocket
     , mkServerSocket
+    , handleIncomingRequest
     )
         where
 
@@ -31,6 +33,7 @@ import           Control.Monad                  (foldM, forM, forM_)
 import qualified Data.ByteString                as BS
 import           Data.Either                    (rights)
 import           Data.Foldable                  (foldl', foldr')
+import           Data.Functor.Identity
 import           Data.IORef
 import           Data.IP                        (IPv6, fromHostAddress6,
                                                  toHostAddress6)
@@ -141,8 +144,33 @@ markCacheEntryAsVerified timestamp = Map.adjust adjustFunc
         adjustFunc (ProxyEntry _ (Just entry)) = adjustFunc entry
         adjustFunc entry = entry
 
+
+-- | uses the successor and predecessor list of a node as an indicator for whether a
+-- node has properly joined the DHT
+isJoined_ :: LocalNodeState -> Bool
+isJoined_ ns = not . all null $ [successors ns, predecessors ns]
+
 -- ====== message send and receive operations ======
 
+handleIncomingRequest :: LocalNodeState                     -- ^ the handling node
+                      -> TQueue (BS.ByteString, SockAddr)   -- ^ send queue
+                      -> FediChordMessage                   -- ^ request to handle
+                      -> SockAddr                           -- ^ source address of the request
+                      -> IO ()
+handleIncomingRequest ns sendQ msg sourceAddr = do
+    -- add nodestate to cache
+    now <- getPOSIXTime
+    queueAddEntries (Identity . RemoteCacheEntry (sender msg) $ now) ns
+    -- distinguish on whether and how to respond
+    -- create and enqueue ACK
+    -- Idea: only respond with payload on last part (part == parts), problem: need to know partnumber of response from first part on
+    -- PLACEHOLDER
+    pure ()
+
+-- ....... response sending .......
+
+-- ....... request sending .......
+
 -- | send a join request and return the joined 'LocalNodeState' including neighbours
 requestJoin :: NodeState a => a             -- ^ currently responsible node to be contacted
             -> LocalNodeState               -- ^ joining NodeState
diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs
index fe6d0aa..8777cb8 100644
--- a/src/Hash2Pub/FediChord.hs
+++ b/src/Hash2Pub/FediChord.hs
@@ -53,10 +53,12 @@ import qualified Data.Map.Strict               as Map
 import           Data.Maybe                    (fromMaybe, isJust, mapMaybe)
 import qualified Data.Set                      as Set
 import           Data.Time.Clock.POSIX
-import           Network.Socket                hiding (recv, recvFrom, send, sendTo)
+import           Network.Socket                hiding (recv, recvFrom, send,
+                                                sendTo)
 import           Network.Socket.ByteString
 
 -- for hashing and ID conversion
+import           Control.Concurrent
 import           Control.Concurrent.Async
 import           Control.Concurrent.STM
 import           Control.Concurrent.STM.TQueue
@@ -213,12 +215,22 @@ fediMessageHandler :: TQueue (BS.ByteString, SockAddr)  -- ^ send queue
                    -> IO ()
 fediMessageHandler sendQ recvQ ns = forever $ do
     -- wait for incoming messages
---    newMsg <- deserialiseMessage <$> recvFrom sock
---    either (\_ ->
---        -- ignore invalid messages
---        pure ()
---           )
---           (\aMsg ->
---        case aMsg of
---          aRequest@Request{} -> handleRequest 
+    (rawMsg, sourceAddr) <- atomically $ readTQueue recvQ
+    let aMsg = deserialiseMessage rawMsg
+    -- handling multipart messages:
+    -- So far I handle the effects of each message part immedeiately, before making sure that and whether all parts have been received, based on the idea that even incomplete information is beneficial and handled idempotent.
+    -- If this turns out not to be the case, request parts can be insert into a map (key: (sender IP against spoofing, request ID), value: timestamp + set of message parts, handle all of them when size of set == parts) before being handled. This map needs to be purged periodically by a separate thread and can be protected by an MVar for fairness.
+    either (\_ ->
+        -- drop invalid messages
+        pure ()
+           )
+           (\validMsg ->
+        case validMsg of
+          aRequest@Request{} -> forkIO (handleIncomingRequest ns sendQ aRequest sourceAddr) >> pure ()
+          -- Responses should never arrive on the main server port, as they are always
+          -- responses to requests sent from dedicated sockets on another port
+          _ -> pure ()
+           )
+        aMsg
+
     pure ()
diff --git a/src/Hash2Pub/Main.hs b/src/Hash2Pub/Main.hs
index 5b7fad6..a837cc5 100644
--- a/src/Hash2Pub/Main.hs
+++ b/src/Hash2Pub/Main.hs
@@ -4,7 +4,7 @@ import           Control.Concurrent
 import           Control.Concurrent.Async
 import           Control.Exception
 import           Data.Either
-import           Data.IP            (IPv6, toHostAddress6)
+import           Data.IP                  (IPv6, toHostAddress6)
 import           System.Environment
 
 import           Hash2Pub.FediChord