Watch direct messages using local state diffs

Changelog: API: Replaced `watchReceivedDirectMessages` with `watchDirectMessageThreads`

4 files changed, 72 insertions, 41 deletions

diff --git a/main/Main.hs b/main/Main.hs
index 974038f..d95e766 100644
--- a/main/Main.hs
+++ b/main/Main.hs
@@ -369,16 +369,20 @@ interactiveLoop st opts = withTerminal commandCompletion $ \term -> do
 
     _ <- liftIO $ do
         tzone <- getCurrentTimeZone
-        watchReceivedDirectMessages erebosHead $ \smsg -> do
-            let msg = fromStored smsg
-            extPrintLn $ formatDirectMessage tzone msg
-            case optDmBotEcho opts of
-                Nothing -> return ()
-                Just prefix -> do
-                    res <- runExceptT $ flip runReaderT erebosHead $ sendDirectMessage (msgFrom msg) (prefix <> msgText msg)
-                    case res of
-                        Right reply -> extPrintLn $ formatDirectMessage tzone $ fromStored reply
-                        Left err -> extPrintLn $ "Failed to send dm echo: " <> err
+        let self = finalOwner $ headLocalIdentity erebosHead
+        watchDirectMessageThreads erebosHead $ \prev cur -> do
+            forM_ (reverse $ dmThreadToListSince prev cur) $ \msg -> do
+                extPrintLn $ formatDirectMessage tzone msg
+                case optDmBotEcho opts of
+                    Just prefix
+                        | not (msgFrom msg `sameIdentity` self)
+                        -> do
+                            void $ forkIO $ do
+                                res <- runExceptT $ flip runReaderT erebosHead $ sendDirectMessage (msgFrom msg) (prefix <> msgText msg)
+                                case res of
+                                    Right _ -> return ()
+                                    Left err -> extPrintLn $ "Failed to send dm echo: " <> err
+                    _ -> return ()
 
     peers <- liftIO $ newMVar []
     contextOptions <- liftIO $ newMVar []
@@ -682,11 +686,7 @@ cmdSend :: Command
 cmdSend = void $ do
     text <- asks ciLine
     conv <- getSelectedConversation
-    sendMessage conv (T.pack text) >>= \case
-        Just msg -> do
-            tzone <- liftIO $ getCurrentTimeZone
-            cmdPutStrLn $ formatMessage tzone msg
-        Nothing -> return ()
+    void $ sendMessage conv (T.pack text)
 
 cmdDelete :: Command
 cmdDelete = void $ do
diff --git a/main/Test.hs b/main/Test.hs
index fd6258d..323b240 100644
--- a/main/Test.hs
+++ b/main/Test.hs
@@ -232,14 +232,16 @@ discoveryAttributes = (defaultServiceAttributes Proxy)
     { discoveryProvideTunnel = \_ _ -> False
     }
 
-dmReceivedWatcher :: Output -> Stored DirectMessage -> IO ()
-dmReceivedWatcher out smsg = do
-    let msg = fromStored smsg
-    outLine out $ unwords
-        [ "dm-received"
-        , "from", maybe "<unnamed>" T.unpack $ idName $ msgFrom msg
-        , "text", T.unpack $ msgText msg
-        ]
+dmThreadWatcher :: ComposedIdentity -> Output -> DirectMessageThread -> DirectMessageThread -> IO ()
+dmThreadWatcher self out prev cur = do
+    forM_ (reverse $ dmThreadToListSince prev cur) $ \msg -> do
+        outLine out $ unwords
+            [ if sameIdentity self (msgFrom msg)
+                 then "dm-sent"
+                 else "dm-received"
+            , "from", maybe "<unnamed>" T.unpack $ idName $ msgFrom msg
+            , "text", T.unpack $ msgText msg
+            ]
 
 
 newtype CommandM a = CommandM (ReaderT TestInput (StateT TestState (ExceptT ErebosError IO)) a)
@@ -456,7 +458,8 @@ cmdHeadUnwatch = do
 
 initTestHead :: Head LocalState -> Command
 initTestHead h = do
-    _ <- liftIO . watchReceivedDirectMessages h . dmReceivedWatcher =<< asks tiOutput
+    let self = finalOwner $ headLocalIdentity h
+    _ <- liftIO . watchDirectMessageThreads h . dmThreadWatcher self =<< asks tiOutput
     modify $ \s -> s { tsHead = Just h }
 
 loadTestHead :: CommandM (Head LocalState)
diff --git a/src/Erebos/DirectMessage.hs b/src/Erebos/DirectMessage.hs
index 8ce3184..2dd0b06 100644
--- a/src/Erebos/DirectMessage.hs
+++ b/src/Erebos/DirectMessage.hs
@@ -9,22 +9,24 @@ module Erebos.DirectMessage (
     dmThreadList,
 
     DirectMessageThread(..),
-    dmThreadToList,
+    dmThreadToList, dmThreadToListSince,
     dmThreadView,
 
-    watchReceivedDirectMessages,
+    watchDirectMessageThreads,
     formatDirectMessage,
 ) where
 
+import Control.Concurrent.MVar
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
 
 import Data.List
 import Data.Ord
-import qualified Data.Set as S
+import Data.Set (Set)
+import Data.Set qualified as S
 import Data.Text (Text)
-import qualified Data.Text as T
+import Data.Text qualified as T
 import Data.Time.Format
 import Data.Time.LocalTime
 
@@ -230,12 +232,18 @@ data DirectMessageThread = DirectMessageThread
     }
 
 dmThreadToList :: DirectMessageThread -> [ DirectMessage ]
-dmThreadToList thread = helper S.empty $ msgHead thread
-    where helper seen msgs
-              | msg : msgs' <- filter (`S.notMember` seen) $ reverse $ sortBy (comparing cmpView) msgs =
-                  fromStored msg : helper (S.insert msg seen) (msgs' ++ msgPrev (fromStored msg))
-              | otherwise = []
-          cmpView msg = (zonedTimeToUTC $ msgTime $ fromStored msg, msg)
+dmThreadToList thread = threadToListHelper S.empty $ msgHead thread
+
+dmThreadToListSince :: DirectMessageThread -> DirectMessageThread -> [ DirectMessage ]
+dmThreadToListSince since thread = threadToListHelper (S.fromAscList $ msgHead since) (msgHead thread)
+
+threadToListHelper :: Set (Stored DirectMessage) -> [ Stored DirectMessage ] -> [ DirectMessage ]
+threadToListHelper seen msgs
+    | msg : msgs' <- filter (`S.notMember` seen) $ reverse $ sortBy (comparing cmpView) msgs =
+        fromStored msg : threadToListHelper (S.insert msg seen) (msgs' ++ msgPrev (fromStored msg))
+    | otherwise = []
+  where
+    cmpView msg = (zonedTimeToUTC $ msgTime $ fromStored msg, msg)
 
 dmThreadView :: [ Stored MessageState ] -> [ DirectMessageThread ]
 dmThreadView = helper []
@@ -264,12 +272,24 @@ messageThreadFor peer mss =
          }
 
 
-watchReceivedDirectMessages :: Head LocalState -> (Stored DirectMessage -> IO ()) -> IO WatchedHead
-watchReceivedDirectMessages h f = do
-    let self = finalOwner $ localIdentity $ headObject h
+watchDirectMessageThreads :: Head LocalState -> (DirectMessageThread -> DirectMessageThread -> IO ()) -> IO WatchedHead
+watchDirectMessageThreads h f = do
+    prevVar <- newMVar Nothing
     watchHeadWith h (lookupSharedValue . lsShared . headObject) $ \(DirectMessageThreads sms _) -> do
-        forM_ (map fromStored sms) $ \ms -> do
-            mapM_ f $ filter (not . sameIdentity self . msgFrom . fromStored) $ msReceived ms
+        modifyMVar_ prevVar $ \case
+            Just prev -> do
+                let addPeer (p : ps) p'
+                        | p `sameIdentity` p' = p : ps
+                        | otherwise = p : addPeer ps p'
+                    addPeer [] p' = [ p' ]
+
+                let peers = foldl' addPeer [] $ map (msPeer . fromStored) $ storedDifference prev sms
+                forM_ peers $ \peer -> do
+                    f (messageThreadFor peer prev) (messageThreadFor peer sms)
+                return (Just sms)
+
+            Nothing -> do
+                return (Just sms)
 
 formatDirectMessage :: TimeZone -> DirectMessage -> String
 formatDirectMessage tzone msg = concat
diff --git a/test/message.et b/test/message.et
index 2990d0f..acdfc27 100644
--- a/test/message.et
+++ b/test/message.et
@@ -28,16 +28,20 @@ test DirectMessage:
 
 	for i in [1..2]:
 		send "dm-send-peer $peer1_2 hello$i" to p1
+		expect /dm-sent from Owner1 text hello$i/ from p1
 		expect /dm-received from Owner1 text hello$i/ from p2
 
 	for i in [1..2]:
 		send "dm-send-peer $peer2_1 hi$i" to p2
+		expect /dm-sent from Owner2 text hi$i/ from p2
 		expect /dm-received from Owner2 text hi$i/ from p1
 
 	for i in [3..4]:
 		send "dm-send-peer $peer1_2 hello$i" to p1
+		expect /dm-sent from Owner1 text hello$i/ from p1
 		expect /dm-received from Owner1 text hello$i/ from p2
 		send "dm-send-peer $peer2_1 hi$i" to p2
+		expect /dm-sent from Owner2 text hi$i/ from p2
 		expect /dm-received from Owner2 text hi$i/ from p1
 
 	# Create contacts
@@ -67,16 +71,20 @@ test DirectMessage:
 
 	for i in [1..2]:
 		send "dm-send-contact $c1_2 hello_c_$i" to p1
+		expect /dm-sent from Owner1 text hello_c_$i/ from p1
 		expect /dm-received from Owner1 text hello_c_$i/ from p2
 
 	for i in [1..2]:
 		send "dm-send-contact $c2_1 hi_c_$i" to p2
+		expect /dm-sent from Owner2 text hi_c_$i/ from p2
 		expect /dm-received from Owner2 text hi_c_$i/ from p1
 
 	for i in [3..4]:
 		send "dm-send-contact $c1_2 hello_c_$i" to p1
+		expect /dm-sent from Owner1 text hello_c_$i/ from p1
 		expect /dm-received from Owner1 text hello_c_$i/ from p2
 		send "dm-send-contact $c2_1 hi_c_$i" to p2
+		expect /dm-sent from Owner2 text hi_c_$i/ from p2
 		expect /dm-received from Owner2 text hi_c_$i/ from p1
 
 	send "dm-list-contact $c1_2" to p1
@@ -135,6 +143,7 @@ test DirectMessage:
 	send "start-server services $services" to p2
 
 	send "dm-send-contact $c1_2 while_offline" to p1
+	expect /dm-sent from Owner1 text while_offline/ from p1
 	send "start-server services $services" to p1
 
 	expect /dm-received from Owner1 text while_offline/ from p2
@@ -148,8 +157,7 @@ test DirectMessage:
 	send "start-server services $services" to p1
 
 	send "dm-send-contact $c1_2 while_peer_offline" to p1
-	# TODO: sync from p1 on peer p2 discovery not ensured without addition wait
-	#wait
+	expect /dm-sent from Owner1 text while_peer_offline/ from p1
 	send "start-server services $services" to p2
 
 	expect /dm-received from Owner1 text while_peer_offline/ from p2