summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--main/Test.hs9
-rw-r--r--src/Erebos/Conversation.hs2
-rw-r--r--src/Erebos/DirectMessage.hs23
-rw-r--r--src/Erebos/Network.hs19
-rw-r--r--src/Erebos/Service.hs4
-rw-r--r--test/common.test3
-rw-r--r--test/message.test114
7 files changed, 160 insertions, 14 deletions
diff --git a/main/Test.hs b/main/Test.hs
index c563291..fa8501e 100644
--- a/main/Test.hs
+++ b/main/Test.hs
@@ -300,6 +300,7 @@ commands = map (T.pack *** id)
, ("contact-set-name", cmdContactSetName)
, ("dm-send-peer", cmdDmSendPeer)
, ("dm-send-contact", cmdDmSendContact)
+ , ("dm-send-identity", cmdDmSendIdentity)
, ("dm-list-peer", cmdDmListPeer)
, ("dm-list-contact", cmdDmListContact)
, ("chatroom-create", cmdChatroomCreate)
@@ -816,6 +817,14 @@ cmdDmSendContact = do
Just to <- contactIdentity <$> getContact cid
void $ sendDirectMessage to msg
+cmdDmSendIdentity :: Command
+cmdDmSendIdentity = do
+ st <- asks tiStorage
+ [ tid, msg ] <- asks tiParams
+ Just ref <- liftIO $ readRef st $ encodeUtf8 tid
+ Just to <- return $ validateExtendedIdentity $ wrappedLoad ref
+ void $ sendDirectMessage to msg
+
dmList :: Foldable f => Identity f -> Command
dmList peer = do
threads <- toThreadList . lookupSharedValue . lsShared . headObject <$> getHead
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
diff --git a/test/common.test b/test/common.test
new file mode 100644
index 0000000..89941f0
--- /dev/null
+++ b/test/common.test
@@ -0,0 +1,3 @@
+module common
+
+export def refpat = /blake2#[0-9a-f]*/
diff --git a/test/message.test b/test/message.test
index c0e251b..2990d0f 100644
--- a/test/message.test
+++ b/test/message.test
@@ -1,3 +1,7 @@
+module message
+
+import common
+
test DirectMessage:
let services = "contact,dm"
@@ -149,3 +153,113 @@ test DirectMessage:
send "start-server services $services" to p2
expect /dm-received from Owner1 text while_peer_offline/ from p2
+
+
+test DirectMessageDiscovery:
+ let services = "dm,discovery"
+
+ subnet sd
+ subnet s1
+ subnet s2
+ subnet s3
+ subnet s4
+
+ spawn on sd as pd
+ spawn on s1 as p1
+ spawn on s2 as p2
+ spawn on s3 as p3
+ spawn on s4 as p4
+
+ send "create-identity Discovery" to pd
+
+ send "create-identity Device1 Owner1" to p1
+ expect /create-identity-done ref ($refpat)/ from p1 capture p1_id
+ send "identity-info $p1_id" to p1
+ expect /identity-info ref $p1_id base ($refpat) owner ($refpat).*/ from p1 capture p1_base, p1_owner
+
+ send "create-identity Device2 Owner2" to p2
+ expect /create-identity-done ref ($refpat)/ from p2 capture p2_id
+ send "identity-info $p2_id" to p2
+ expect /identity-info ref $p2_id base ($refpat) owner ($refpat).*/ from p2 capture p2_base, p2_owner
+ send "identity-info $p2_owner" to p2
+ expect /identity-info ref $p2_owner base ($refpat).*/ from p2 capture p2_obase
+
+ send "create-identity Device3 Owner3" to p3
+ expect /create-identity-done ref ($refpat)/ from p3 capture p3_id
+ send "identity-info $p3_id" to p3
+ expect /identity-info ref $p3_id base ($refpat) owner ($refpat).*/ from p3 capture p3_base, p3_owner
+
+ send "create-identity Device4 Owner4" to p4
+ expect /create-identity-done ref ($refpat)/ from p4 capture p4_id
+ send "identity-info $p4_id" to p4
+ expect /identity-info ref $p4_id base ($refpat) owner ($refpat).*/ from p4 capture p4_base, p4_owner
+
+
+ for p in [ p1, p2, p3, p4 ]:
+ with p:
+ send "start-server services $services"
+
+ for p in [ p2, p3, p4 ]:
+ with p1:
+ send "peer-add ${p.node.ip}"
+ expect:
+ /peer [0-9]+ addr ${p.node.ip} 29665/
+ /peer [0-9]+ id Device. Owner./
+ expect from p:
+ /peer 1 addr ${p1.node.ip} 29665/
+ /peer 1 id Device1 Owner1/
+
+ # Make sure p1 has other identities in storage:
+ for i in [ 1 .. 3 ]:
+ send "dm-send-peer $i init1" to p1
+ for p in [ p2, p3, p4 ]:
+ expect /dm-received from Owner1 text init1/ from p
+ send "dm-send-identity $p1_owner init2" to p
+ expect /dm-received from Owner. text init2/ from p1
+
+ # Restart servers to remove peers:
+ for p in [ p1, p2, p3, p4 ]:
+ with p:
+ send "stop-server"
+ for p in [ p1, p2, p3, p4 ]:
+ with p:
+ expect /stop-server-done/
+
+ # Prepare message before peers connect to discovery
+ send "dm-send-identity $p4_owner hello_to_p4" to p1
+
+ for p in [ p1, p2, p3, p4, pd ]:
+ with p:
+ send "start-server services $services"
+
+ for p in [ p2, p3, p4, p1 ]:
+ with p:
+ send "peer-add ${pd.node.ip}"
+ expect:
+ /peer 1 addr ${pd.node.ip} 29665/
+ /peer 1 id Discovery/
+ expect from pd:
+ /peer [0-9]+ addr ${p.node.ip} 29665/
+ /peer [0-9]+ id Device. Owner./
+
+ multiply_timeout by 2.0
+
+ # Connect via discovery manually, then send message
+ send "discovery-connect $p2_obase" to p1
+ expect from p1:
+ /peer [0-9]+ addr ${p2.node.ip} 29665/
+ /peer [0-9]+ id Device2 Owner2/
+ send "dm-send-identity $p2_owner hello_to_p2" to p1
+ expect /dm-received from Owner1 text hello_to_p2/ from p2
+
+ # Send message, expect automatic discovery
+ send "dm-send-identity $p3_owner hello_to_p3" to p1
+ expect /dm-received from Owner1 text hello_to_p3/ from p3
+
+ # Verify the first message
+ expect /dm-received from Owner1 text hello_to_p4/ from p4
+
+ for p in [ p1, p2, p3, p4, pd ]:
+ send "stop-server" to p
+ for p in [ p1, p2, p3, p4, pd ]:
+ expect /stop-server-done/ from p
generated by cgit v1.2.3 (git 2.25.1) at 2025-06-29 15:06:57 +0000