Automatic discovery of peers for pending messages

Changelog: Automatic discovery of peers for pending direct messages

4 files changed, 34 insertions, 14 deletions

diff --git a/src/Erebos/Conversation.hs b/src/Erebos/Conversation.hs
index dee6faa..187fddd 100644
--- a/src/Erebos/Conversation.hs
+++ b/src/Erebos/Conversation.hs
@@ -71,7 +71,7 @@ directMessageConversation :: MonadHead LocalState m => ComposedIdentity -> m Con
 directMessageConversation peer = do
     (find (sameIdentity peer . msgPeer) . toThreadList . lookupSharedValue . lsShared . fromStored <$> getLocalHead) >>= \case
         Just thread -> return $ DirectMessageConversation thread
-        Nothing -> return $ DirectMessageConversation $ DirectMessageThread peer [] [] []
+        Nothing -> return $ DirectMessageConversation $ DirectMessageThread peer [] [] [] []
 
 chatroomConversation :: MonadHead LocalState m => ChatroomState -> m (Maybe Conversation)
 chatroomConversation rstate = chatroomConversationByStateData (head $ roomStateData rstate)
diff --git a/src/Erebos/DirectMessage.hs b/src/Erebos/DirectMessage.hs
index 05da865..dc6724c 100644
--- a/src/Erebos/DirectMessage.hs
+++ b/src/Erebos/DirectMessage.hs
@@ -17,6 +17,7 @@ module Erebos.DirectMessage (
 ) where
 
 import Control.Monad
+import Control.Monad.Except
 import Control.Monad.Reader
 
 import Data.List
@@ -27,8 +28,10 @@ import qualified Data.Text as T
 import Data.Time.Format
 import Data.Time.LocalTime
 
+import Erebos.Discovery
 import Erebos.Identity
 import Erebos.Network
+import Erebos.Object
 import Erebos.Service
 import Erebos.State
 import Erebos.Storable
@@ -102,8 +105,10 @@ instance Service DirectMessage where
 
     serviceNewPeer = syncDirectMessageToPeer . lookupSharedValue . lsShared . fromStored =<< svcGetLocal
 
-    serviceStorageWatchers _ = (:[]) $
-        SomeStorageWatcher (lookupSharedValue . lsShared . fromStored) syncDirectMessageToPeer
+    serviceStorageWatchers _ =
+        [ SomeStorageWatcher (lookupSharedValue . lsShared . fromStored) syncDirectMessageToPeer
+        , GlobalStorageWatcher (lookupSharedValue . lsShared . fromStored) findMissingPeers
+        ]
 
 
 data MessageState = MessageState
@@ -209,12 +214,19 @@ syncDirectMessageToPeer (DirectMessageThreads mss _) = do
               else do
                 return unchanged
 
+findMissingPeers :: Server -> DirectMessageThreads -> ExceptT ErebosError IO ()
+findMissingPeers server threads = do
+    forM_ (toThreadList threads) $ \thread -> do
+        when (msgHead thread /= msgReceived thread) $ do
+            mapM_ (discoverySearch server) $ map (refDigest . storedRef) $ idDataF $ msgPeer thread
+
 
 data DirectMessageThread = DirectMessageThread
     { msgPeer :: ComposedIdentity
-    , msgHead :: [Stored DirectMessage]
-    , msgSent :: [Stored DirectMessage]
-    , msgSeen :: [Stored DirectMessage]
+    , msgHead :: [ Stored DirectMessage ]
+    , msgSent :: [ Stored DirectMessage ]
+    , msgSeen :: [ Stored DirectMessage ]
+    , msgReceived :: [ Stored DirectMessage ]
     }
 
 threadToList :: DirectMessageThread -> [DirectMessage]
@@ -248,6 +260,7 @@ messageThreadFor peer mss =
          , msgHead = filterAncestors $ ready ++ received
          , msgSent = filterAncestors $ sent ++ received
          , msgSeen = filterAncestors $ ready ++ seen
+         , msgReceived = filterAncestors $ received
          }
 
 
diff --git a/src/Erebos/Network.hs b/src/Erebos/Network.hs
index b341974..8da4c8d 100644
--- a/src/Erebos/Network.hs
+++ b/src/Erebos/Network.hs
@@ -327,13 +327,18 @@ startServer serverOptions serverOrigHead logd' serverServices = do
                     announceUpdate idt
 
             forM_ serverServices $ \(SomeService service _) -> do
-                forM_ (serviceStorageWatchers service) $ \(SomeStorageWatcher sel act) -> do
-                    watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
-                        withMVar serverPeers $ mapM_ $ \peer -> atomically $ do
-                            readTVar (peerIdentityVar peer) >>= \case
-                                PeerIdentityFull _ -> writeTQueue serverIOActions $ do
-                                    runPeerService peer $ act x
-                                _ -> return ()
+                forM_ (serviceStorageWatchers service) $ \case
+                    SomeStorageWatcher sel act -> do
+                        watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
+                            withMVar serverPeers $ mapM_ $ \peer -> atomically $ do
+                                readTVar (peerIdentityVar peer) >>= \case
+                                    PeerIdentityFull _ -> writeTQueue serverIOActions $ do
+                                        runPeerService peer $ act x
+                                    _ -> return ()
+                    GlobalStorageWatcher sel act -> do
+                        watchHeadWith serverOrigHead (sel . headStoredObject) $ \x -> do
+                            atomically $ writeTQueue serverIOActions $ do
+                                act server x
 
             forkServerThread server $ forever $ do
                 (msg, saddr) <- S.recvFrom sock 4096
diff --git a/src/Erebos/Service.hs b/src/Erebos/Service.hs
index fefc503..4499ef9 100644
--- a/src/Erebos/Service.hs
+++ b/src/Erebos/Service.hs
@@ -104,7 +104,9 @@ someServiceEmptyGlobalState :: SomeService -> SomeServiceGlobalState
 someServiceEmptyGlobalState (SomeService p _) = SomeServiceGlobalState p (emptyServiceGlobalState p)
 
 
-data SomeStorageWatcher s = forall a. Eq a => SomeStorageWatcher (Stored LocalState -> a) (a -> ServiceHandler s ())
+data SomeStorageWatcher s
+    = forall a. Eq a => SomeStorageWatcher (Stored LocalState -> a) (a -> ServiceHandler s ())
+    | forall a. Eq a => GlobalStorageWatcher (Stored LocalState -> a) (Server -> a -> ExceptT ErebosError IO ())
 
 
 mkServiceID :: String -> ServiceID