erebos 0.1.3 → 0.1.4

raw patch · 12 files changed

+748/−199 lines, 12 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +7/−0
• README.md +51/−7
• erebos.cabal +2/−1
• main/Main.hs +197/−60
• main/Test.hs +47/−5
• src/Erebos/Chatroom.hs +125/−6
• src/Erebos/Conversation.hs +75/−0
• src/Erebos/Discovery.hs +1/−0
• src/Erebos/Message.hs +7/−2
• src/Erebos/Network.hs +115/−51
• src/Erebos/Network/Protocol.hs +97/−64
• src/Erebos/Storage.hs +24/−3

@@ -1,5 +1,12 @@ # Revision history for erebos +## 0.1.4 -- 2024-06-11++* Added `/conversations` command to list and select conversations+* Added `/details` command for info about selected conversation+* Handle peer reconnection after its restart+* Support non-interactive mode without tty+ ## 0.1.3 -- 2024-05-05  * Enable/disable network services by command-line parameters

@@ -36,10 +36,38 @@  After the initial setup, the erebos tool presents interactive prompt for messages and commands. All commands start with the slash (`/`) character,-followed by command name and parameters (if any) separated by spaces. When a-peer or contact is selected, message to send him can be entered directly on the-command prompt.+followed by command name and parameters (if any) separated by spaces. When+a conversation is selected, message to send there is entered directly on+the command prompt. +The session can be terminated either by end-of-input (typically `Ctrl-d`) or+using the `/quit` command.++### Example++Start `erebos` CLI and create new identity:+```+Name: Some Name+Device: First device+Some Name / First device+> +```++Add public peer:+```+> /peer-add-public+[1] PEER NEW <unnamed> [37.221.243.57 29665]+[1] PEER UPD discovery1.erebosprotocol.net [37.221.243.57 29665]+```++Select the peer and send it a message, the public server just responds with automatic echo message:+```+> /1+discovery1.erebosprotocol.net> hello+[18:55] Some Name: hello+[18:55] discovery1.erebosprotocol.net: Echo: hello+```+ ### Messaging  `/peers`  @@ -49,16 +77,22 @@ `/contacts`   List known contacts (see below). +`/conversations`  +List started conversations with contacts or other peers.+ `/<number>`  -Select contact or peer `<number>` based on previous `/contacts` or `/peers`-output list.+Select conversation, contact or peer `<number>` based on the last+`/conversations`, `/contacts` or `/peers` output list.  `<message>`  -Send `<message>` to selected contact.+Send `<message>` to selected conversation.  `/history`  -Show message history for selected contact or peer.+Show message history of the selected conversation. +`/details`  +Show information about the selected conversations, contact or peer.+ ### Add contacts  To ensure the identity of the contact and prevent man-in-the-middle attack,@@ -119,8 +153,18 @@ `/peer-add <host> [<port>]`   Manually add network peer with given hostname or IP address. +`/peer-add-public`  +Add known public network peer(s).++`/peer-drop`  +Drop the currently selected peer. Afterwards, the connection can be+re-established by either side.+ `/update-identity`   Interactively update current identity information++`/quit`  +Quit the erebos tool.   Storage

@@ -1,7 +1,7 @@ Cabal-Version:       2.2  Name:                erebos-Version:             0.1.3+Version:             0.1.4 Synopsis:            Decentralized messaging and synchronization Description:     Library and simple CLI interface implementing the Erebos identity@@ -97,6 +97,7 @@         Erebos.Channel         Erebos.Chatroom         Erebos.Contact+        Erebos.Conversation         Erebos.Identity         Erebos.Message         Erebos.Network

@@ -20,9 +20,10 @@ import Data.List import Data.Maybe import Data.Ord-import qualified Data.Text as T-import qualified Data.Text.Encoding as T-import qualified Data.Text.IO as T+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as T+import Data.Text.IO qualified as T import Data.Time.LocalTime import Data.Typeable @@ -36,12 +37,13 @@  import Erebos.Attach import Erebos.Contact+import Erebos.Conversation #ifdef ENABLE_ICE_SUPPORT import Erebos.Discovery import Erebos.ICE #endif import Erebos.Identity-import Erebos.Message+import Erebos.Message hiding (formatMessage) import Erebos.Network import Erebos.PubKey import Erebos.Service@@ -57,6 +59,7 @@ data Options = Options     { optServer :: ServerOptions     , optServices :: [ServiceOption]+    , optDmBotEcho :: Maybe Text     , optShowHelp :: Bool     , optShowVersion :: Bool     }@@ -72,6 +75,7 @@ defaultOptions = Options     { optServer = defaultServerOptions     , optServices = availableServices+    , optDmBotEcho = Nothing     , optShowHelp = False     , optShowVersion = False     }@@ -100,6 +104,9 @@     , Option ['s'] ["silent"]         (NoArg (so $ \opts -> opts { serverLocalDiscovery = False }))         "do not send announce packets for local discovery"+    , Option [] ["dm-bot-echo"]+        (ReqArg (\prefix -> \opts -> opts { optDmBotEcho = Just (T.pack prefix) }) "<prefix>")+        "automatically reply to direct messages with the same text prefixed with <prefix>"     , Option ['h'] ["help"]         (NoArg $ \opts -> opts { optShowHelp = True })         "show this help and exit"@@ -184,7 +191,7 @@                     serviceDesc ServiceOption {..} = padService ("  " <> soptName) <> soptDescription                      padTo n str = str <> replicate (n - length str) ' '-                    padOpt = padTo 28+                    padOpt = padTo 37                     padService = padTo 16                  if | optShowHelp opts -> putStr $ usageInfo header options <> unlines@@ -213,16 +220,66 @@     erebosHead <- liftIO $ loadLocalStateHead st     outputStrLn $ T.unpack $ displayIdentity $ headLocalIdentity erebosHead -    haveTerminalUI >>= \case True -> return ()-                             False -> error "Requires terminal"+    tui <- haveTerminalUI     extPrint <- getExternalPrint     let extPrintLn str = extPrint $ case reverse str of ('\n':_) -> str                                                         _ -> str ++ "\n"; +    let getInputLinesTui eprompt = do+            prompt <- case eprompt of+                Left cstate -> do+                    pname <- case csContext cstate of+                        NoContext -> return ""+                        SelectedPeer peer -> peerIdentity peer >>= return . \case+                            PeerIdentityFull pid -> maybe "<unnamed>" T.unpack $ idName $ finalOwner pid+                            PeerIdentityRef wref _ -> "<" ++ BC.unpack (showRefDigest $ wrDigest wref) ++ ">"+                            PeerIdentityUnknown _  -> "<unknown>"+                        SelectedContact contact -> return $ T.unpack $ contactName contact+                        SelectedConversation conv -> return $ T.unpack $ conversationName conv+                    return $ pname ++ "> "+                Right prompt -> return prompt+            Just input <- lift $ getInputLine prompt+            case reverse input of+                 _ | all isSpace input -> getInputLinesTui eprompt+                 '\\':rest -> (reverse ('\n':rest) ++) <$> getInputLinesTui (Right ">> ")+                 _         -> return input++        getInputCommandTui cstate = do+            input <- getInputLinesTui cstate+            let (CommandM cmd, line) = case input of+                    '/':rest -> let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') rest+                                 in if not (null scmd) && all isDigit scmd+                                       then (cmdSelectContext, scmd)+                                       else (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args)+                    _        -> (cmdSend, input)+            return (cmd, line)++        getInputLinesPipe = do+            lift (getInputLine "") >>= \case+                Just input -> return input+                Nothing -> liftIO $ forever $ threadDelay 100000000++        getInputCommandPipe _ = do+            input <- getInputLinesPipe+            let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') input+            let (CommandM cmd, line) = (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args)+            return (cmd, line)++    let getInputCommand = if tui then getInputCommandTui . Left+                                 else getInputCommandPipe+     _ <- liftIO $ do         tzone <- getCurrentTimeZone-        watchReceivedMessages erebosHead $-            extPrintLn . formatMessage tzone . fromStored+        watchReceivedMessages 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      server <- liftIO $ do         startServer (optServer opts) erebosHead extPrintLn $@@ -235,58 +292,47 @@         peer <- getNextPeerChange server         peerIdentity peer >>= \case             pid@(PeerIdentityFull _) -> do+                dropped <- isPeerDropped peer                 let shown = showPeer pid $ peerAddress peer-                let update [] = ([(peer, shown)], Nothing)-                    update ((p,s):ps) | p == peer = ((peer, shown) : ps, Just s)-                                      | otherwise = first ((p,s):) $ update ps+                let update [] = ([(peer, shown)], (Nothing, "NEW"))+                    update ((p,s):ps)+                        | p == peer && dropped = (ps, (Nothing, "DEL"))+                        | p == peer = ((peer, shown) : ps, (Just s, "UPD"))+                        | otherwise = first ((p,s):) $ update ps                 let ctxUpdate n [] = ([SelectedPeer peer], n)                     ctxUpdate n (ctx:ctxs)                         | SelectedPeer p <- ctx, p == peer = (ctx:ctxs, n)                         | otherwise = first (ctx:) $ ctxUpdate (n + 1) ctxs-                op <- modifyMVar peers (return . update)+                (op, updateType) <- modifyMVar peers (return . update)+                let updateType' = if dropped then "DEL" else updateType                 idx <- modifyMVar contextOptions (return . ctxUpdate (1 :: Int))-                when (Just shown /= op) $ extPrint $ "[" <> show idx <> "] PEER " <> shown+                when (Just shown /= op) $ extPrintLn $ "[" <> show idx <> "] PEER " <> updateType' <> " " <> shown             _ -> return () -    let getInputLines prompt = do-            Just input <- lift $ getInputLine prompt-            case reverse input of-                 _ | all isSpace input -> getInputLines prompt-                 '\\':rest -> (reverse ('\n':rest) ++) <$> getInputLines ">> "-                 _         -> return input-     let process :: CommandState -> MaybeT (InputT IO) CommandState         process cstate = do-            pname <- case csContext cstate of-                        NoContext -> return ""-                        SelectedPeer peer -> peerIdentity peer >>= return . \case-                            PeerIdentityFull pid -> maybe "<unnamed>" T.unpack $ idName $ finalOwner pid-                            PeerIdentityRef wref _ -> "<" ++ BC.unpack (showRefDigest $ wrDigest wref) ++ ">"-                            PeerIdentityUnknown _  -> "<unknown>"-                        SelectedContact contact -> return $ T.unpack $ contactName contact-            input <- getInputLines $ pname ++ "> "-            let (CommandM cmd, line) = case input of-                    '/':rest -> let (scmd, args) = dropWhile isSpace <$> span (\c -> isAlphaNum c || c == '-') rest-                                 in if not (null scmd) && all isDigit scmd-                                       then (cmdSelectContext $ read scmd, args)-                                       else (fromMaybe (cmdUnknown scmd) $ lookup scmd commands, args)-                    _        -> (cmdSend, input)+            (cmd, line) <- getInputCommand cstate             h <- liftIO (reloadHead $ csHead cstate) >>= \case                 Just h  -> return h-                Nothing -> do lift $ lift $ extPrint "current head deleted"+                Nothing -> do lift $ lift $ extPrintLn "current head deleted"                               mzero             res <- liftIO $ runExceptT $ flip execStateT cstate { csHead = h } $ runReaderT cmd CommandInput                 { ciServer = server                 , ciLine = line                 , ciPrint = extPrintLn-                , ciPeers = liftIO $ readMVar peers+                , ciPeers = liftIO $ modifyMVar peers $ \ps -> do+                    ps' <- filterM (fmap not . isPeerDropped . fst) ps+                    return (ps', ps')                 , ciContextOptions = liftIO $ readMVar contextOptions                 , ciSetContextOptions = \ctxs -> liftIO $ modifyMVar_ contextOptions $ const $ return ctxs                 }             case res of-                 Right cstate' -> return cstate'-                 Left err -> do lift $ lift $ extPrint $ "Error: " ++ err-                                return cstate+                Right cstate'+                    | csQuit cstate' -> mzero+                    | otherwise      -> return cstate'+                Left err -> do+                    lift $ lift $ extPrintLn $ "Error: " ++ err+                    return cstate      let loop (Just cstate) = runMaybeT (process cstate) >>= loop         loop Nothing = return ()@@ -297,6 +343,7 @@         , csIceSessions = [] #endif         , csIcePeer = Nothing+        , csQuit = False         }  @@ -316,11 +363,13 @@     , csIceSessions :: [IceSession] #endif     , csIcePeer :: Maybe Peer+    , csQuit :: Bool     }  data CommandContext = NoContext                     | SelectedPeer Peer                     | SelectedContact Contact+                    | SelectedConversation Conversation  newtype CommandM a = CommandM (ReaderT CommandInput (StateT CommandState (ExceptT String IO)) a)     deriving (Functor, Applicative, Monad, MonadReader CommandInput, MonadState CommandState, MonadError String)@@ -353,21 +402,24 @@     SelectedPeer peer -> return peer     _ -> throwError "no peer selected" -getSelectedIdentity :: CommandM ComposedIdentity-getSelectedIdentity = gets csContext >>= \case+getSelectedConversation :: CommandM Conversation+getSelectedConversation = gets csContext >>= \case     SelectedPeer peer -> peerIdentity peer >>= \case-        PeerIdentityFull pid -> return $ toComposedIdentity pid+        PeerIdentityFull pid -> directMessageConversation $ finalOwner pid         _ -> throwError "incomplete peer identity"     SelectedContact contact -> case contactIdentity contact of-        Just cid -> return cid+        Just cid -> directMessageConversation cid         Nothing -> throwError "contact without erebos identity"-    _ -> throwError "no contact or peer selected"+    SelectedConversation conv -> reloadConversation conv+    _ -> throwError "no contact, peer or conversation selected"  commands :: [(String, Command)] commands =     [ ("history", cmdHistory)     , ("peers", cmdPeers)     , ("peer-add", cmdPeerAdd)+    , ("peer-add-public", cmdPeerAddPublic)+    , ("peer-drop", cmdPeerDrop)     , ("send", cmdSend)     , ("update-identity", cmdUpdateIdentity)     , ("attach", cmdAttach)@@ -377,6 +429,8 @@     , ("contact-add", cmdContactAdd)     , ("contact-accept", cmdContactAccept)     , ("contact-reject", cmdContactReject)+    , ("conversations", cmdConversations)+    , ("details", cmdDetails) #ifdef ENABLE_ICE_SUPPORT     , ("discovery-init", cmdDiscoveryInit)     , ("discovery", cmdDiscovery)@@ -386,6 +440,8 @@     , ("ice-connect", cmdIceConnect)     , ("ice-send", cmdIceSend) #endif+    , ("select", cmdSelectContext)+    , ("quit", cmdQuit)     ]  commandCompletion :: CompletionFunc IO@@ -417,6 +473,17 @@     addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just hostname) (Just port)     liftIO $ serverPeer server (addrAddress addr) +cmdPeerAddPublic :: Command+cmdPeerAddPublic = do+    server <- asks ciServer+    addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just "discovery1.erebosprotocol.net") (Just (show discoveryPort))+    void $ liftIO $ serverPeer server (addrAddress addr)++cmdPeerDrop :: Command+cmdPeerDrop = do+    dropPeer =<< getSelectedPeer+    modify $ \s -> s { csContext = NoContext }+ showPeer :: PeerIdentity -> PeerAddress -> String showPeer pidentity paddr =     let name = case pidentity of@@ -425,30 +492,29 @@                     PeerIdentityFull pid   -> T.unpack $ displayIdentity pid      in name ++ " [" ++ show paddr ++ "]" -cmdSelectContext :: Int -> Command-cmdSelectContext n = join (asks ciContextOptions) >>= \ctxs -> if-    | n > 0, (ctx : _) <- drop (n - 1) ctxs -> modify $ \s -> s { csContext = ctx }-    | otherwise -> throwError "invalid index"+cmdSelectContext :: Command+cmdSelectContext = do+    n <- read <$> asks ciLine+    join (asks ciContextOptions) >>= \ctxs -> if+        | n > 0, (ctx : _) <- drop (n - 1) ctxs -> modify $ \s -> s { csContext = ctx }+        | otherwise -> throwError "invalid index"  cmdSend :: Command cmdSend = void $ do     text <- asks ciLine-    powner <- finalOwner <$> getSelectedIdentity-    smsg <- sendDirectMessage powner $ T.pack text+    conv <- getSelectedConversation+    msg <- sendMessage conv $ T.pack text     tzone <- liftIO $ getCurrentTimeZone-    liftIO $ putStrLn $ formatMessage tzone $ fromStored smsg+    liftIO $ putStrLn $ formatMessage tzone msg  cmdHistory :: Command cmdHistory = void $ do-    ehead <- gets csHead-    powner <- finalOwner <$> getSelectedIdentity--    case find (sameIdentity powner . msgPeer) $-            toThreadList $ lookupSharedValue $ lsShared $ headObject ehead of-        Just thread -> do+    conv <- getSelectedConversation+    case conversationHistory conv of+        thread@(_:_) -> do             tzone <- liftIO $ getCurrentTimeZone-            liftIO $ mapM_ (putStrLn . formatMessage tzone) $ reverse $ take 50 $ threadToList thread-        Nothing -> do+            liftIO $ mapM_ (putStrLn . formatMessage tzone) $ reverse $ take 50 thread+        [] -> do             liftIO $ putStrLn $ "<empty history>"  cmdUpdateIdentity :: Command@@ -493,6 +559,74 @@ cmdContactReject :: Command cmdContactReject = contactReject =<< getSelectedPeer +cmdConversations :: Command+cmdConversations = do+    conversations <- lookupConversations+    set <- asks ciSetContextOptions+    set $ map SelectedConversation conversations+    forM_ (zip [1..] conversations) $ \(i :: Int, conv) -> do+        liftIO $ putStrLn $ "[" ++ show i ++ "] " ++ T.unpack (conversationName conv)++cmdDetails :: Command+cmdDetails = do+    gets csContext >>= \case+        SelectedPeer peer -> do+            liftIO $ putStr $ unlines+                [ "Network peer:"+                , "  " <> show (peerAddress peer)+                ]+            peerIdentity peer >>= \case+                PeerIdentityUnknown _ -> liftIO $ do+                    putStrLn $ "unknown identity"+                PeerIdentityRef wref _ -> liftIO $ do+                    putStrLn $ "Identity ref:"+                    putStrLn $ "  " <> BC.unpack (showRefDigest $ wrDigest wref)+                PeerIdentityFull pid -> printContactOrIdentityDetails pid++        SelectedContact contact -> do+            printContactDetails contact++        SelectedConversation conv -> do+            case conversationPeer conv of+                Just pid -> printContactOrIdentityDetails pid+                Nothing -> liftIO $ putStrLn $ "(conversation without peer)"++        NoContext -> liftIO $ putStrLn "nothing selected"+  where+    printContactOrIdentityDetails cid = do+        contacts <- fromSetBy (comparing contactName) . lookupSharedValue . lsShared . fromStored <$> getLocalHead+        case find (maybe False (sameIdentity cid) . contactIdentity) contacts of+            Just contact -> printContactDetails contact+            Nothing -> printIdentityDetails cid++    printContactDetails contact = liftIO $ do+        putStrLn $ "Contact:"+        prefix <- case contactCustomName contact of+            Just name -> do+                putStrLn $ "  " <> T.unpack name+                return $ Just "alias of"+            Nothing -> do+                return $ Nothing++        case contactIdentity contact of+            Just cid -> do+                printIdentityDetailsBody prefix cid+            Nothing -> do+                putStrLn $ "  (without erebos identity)"++    printIdentityDetails identity = liftIO $ do+        putStrLn $ "Identity:"+        printIdentityDetailsBody Nothing identity++    printIdentityDetailsBody prefix identity = do+        forM_ (zip (False : repeat True) $ unfoldOwners identity) $ \(owned, cpid) -> do+            putStrLn $ unwords $ concat+                [ [ "  " ]+                , if owned then [ "owned by" ] else maybeToList prefix+                , [ maybe "<unnamed>" T.unpack (idName cpid) ]+                , map (BC.unpack . showRefDigest . refDigest . storedRef) $ idExtDataF cpid+                ]+ #ifdef ENABLE_ICE_SUPPORT  cmdDiscoveryInit :: Command@@ -566,3 +700,6 @@     liftIO $ serverPeerIce server s  #endif++cmdQuit :: Command+cmdQuit = modify $ \s -> s { csQuit = True }

@@ -246,6 +246,8 @@     , ("start-server", cmdStartServer)     , ("stop-server", cmdStopServer)     , ("peer-add", cmdPeerAdd)+    , ("peer-drop", cmdPeerDrop)+    , ("peer-list", cmdPeerList)     , ("test-message-send", cmdTestMessageSend)     , ("shared-state-get", cmdSharedStateGet)     , ("shared-state-wait", cmdSharedStateWait)@@ -269,6 +271,7 @@     , ("chatroom-list-local", cmdChatroomListLocal)     , ("chatroom-watch-local", cmdChatroomWatchLocal)     , ("chatroom-set-name", cmdChatroomSetName)+    , ("chatroom-message-send", cmdChatroomMessageSend)     ]  cmdStore :: Command@@ -409,6 +412,29 @@     addr:_ <- liftIO $ getAddrInfo (Just $ defaultHints { addrSocketType = Datagram }) (Just host) (Just port)     void $ liftIO $ serverPeer rsServer (addrAddress addr) +cmdPeerDrop :: Command+cmdPeerDrop = do+    [spidx] <- asks tiParams+    peer <- getPeer spidx+    liftIO $ dropPeer peer++cmdPeerList :: Command+cmdPeerList = do+    Just RunningServer {..} <- gets tsServer+    peers <- liftIO $ getCurrentPeerList rsServer+    tpeers <- liftIO $ readMVar rsPeers+    forM_ peers $ \peer -> do+        Just (n, _) <- return $ find ((peer==).snd) . snd $ tpeers+        mbpid <- peerIdentity peer+        cmdOut $ unwords $ concat+            [ [ "peer-list-item", show n ]+            , [ "addr", show (peerAddress peer) ]+            , case mbpid of PeerIdentityFull pid -> ("id":) $ map (maybe "<unnamed>" T.unpack . idName) (unfoldOwners pid)+                            _ -> []+            ]+    cmdOut "peer-list-done"++ cmdTestMessageSend :: Command cmdTestMessageSend = do     [spidx, tref] <- asks tiParams@@ -628,14 +654,30 @@         Just diff -> forM_ diff $ \case             AddedChatroom room -> outLine out $ unwords $ "chatroom-watched-added" : chatroomInfo room             RemovedChatroom room -> outLine out $ unwords $ "chatroom-watched-removed" : chatroomInfo room-            UpdatedChatroom oldroom room -> outLine out $ unwords $ concat-                [ [ "chatroom-watched-updated" ], chatroomInfo room-                , [ "old" ], map (show . refDigest . storedRef) (roomStateData oldroom)-                , [ "new" ], map (show . refDigest . storedRef) (roomStateData room)-                ]+            UpdatedChatroom oldroom room -> do+                when (any (not . null . rsdRoom . fromStored) (roomStateData room)) $ do+                    outLine out $ unwords $ concat+                        [ [ "chatroom-watched-updated" ], chatroomInfo room+                        , [ "old" ], map (show . refDigest . storedRef) (roomStateData oldroom)+                        , [ "new" ], map (show . refDigest . storedRef) (roomStateData room)+                        ]+                when (any (not . null . rsdMessages . fromStored) (roomStateData room)) $ do+                    forM_ (getMessagesSinceState room oldroom) $ \msg -> do+                        outLine out $ unwords $ concat+                            [ [ "chatroom-message-new" ]+                            , [ show . refDigest . storedRef . head . filterAncestors . concatMap storedRoots . toComponents $ room ]+                            , [ "from", maybe "<unnamed>" T.unpack $ idName $ cmsgFrom msg ]+                            , maybe [] (("text":) . (:[]) . T.unpack) $ cmsgText msg+                            ]  chatroomInfo :: ChatroomState -> [String] chatroomInfo room =     [ show . refDigest . storedRef . head . filterAncestors . concatMap storedRoots . toComponents $ room     , maybe "<unnamed>" T.unpack $ roomName =<< roomStateRoom room     ]++cmdChatroomMessageSend :: Command+cmdChatroomMessageSend = do+    [cid, msg] <- asks tiParams+    to <- getChatroomStateData cid+    void $ chatroomMessageByStateData to msg

@@ -10,10 +10,15 @@     listChatrooms,     findChatroomByRoomData,     findChatroomByStateData,+    getMessagesSinceState,      ChatroomSetChange(..),     watchChatrooms, +    ChatMessage, cmsgFrom, cmsgReplyTo, cmsgTime, cmsgText, cmsgLeave,+    ChatMessageData(..),+    chatroomMessageByStateData,+     ChatroomService(..), ) where @@ -27,8 +32,11 @@ import Data.Maybe import Data.Monoid import Data.Ord+import Data.Set qualified as S import Data.Text (Text)+import Data.Time +import Erebos.Identity import Erebos.PubKey import Erebos.Service import Erebos.Set@@ -87,34 +95,129 @@          in All $ all (fromStored sdata `isSignedBy`) required  +data ChatMessageData = ChatMessageData+    { mdPrev :: [Stored (Signed ChatMessageData)]+    , mdRoom :: [Stored (Signed ChatroomData)]+    , mdFrom :: ComposedIdentity+    , mdReplyTo :: Maybe (Stored (Signed ChatMessageData))+    , mdTime :: ZonedTime+    , mdText :: Maybe Text+    , mdLeave :: Bool+    }++data ChatMessage = ChatMessage+    { cmsgData :: Stored (Signed ChatMessageData)+    }++cmsgFrom :: ChatMessage -> ComposedIdentity+cmsgFrom = mdFrom . fromSigned . cmsgData++cmsgReplyTo :: ChatMessage -> Maybe ChatMessage+cmsgReplyTo = fmap ChatMessage . mdReplyTo . fromSigned . cmsgData++cmsgTime :: ChatMessage -> ZonedTime+cmsgTime = mdTime . fromSigned . cmsgData++cmsgText :: ChatMessage -> Maybe Text+cmsgText = mdText . fromSigned . cmsgData++cmsgLeave :: ChatMessage -> Bool+cmsgLeave = mdLeave . fromSigned . cmsgData++instance Storable ChatMessageData where+    store' ChatMessageData {..} = storeRec $ do+        mapM_ (storeRef "SPREV") mdPrev+        mapM_ (storeRef "room") mdRoom+        mapM_ (storeRef "from") $ idExtDataF mdFrom+        storeMbRef "reply-to" mdReplyTo+        storeDate "time" mdTime+        storeMbText "text" mdText+        when mdLeave $ storeEmpty "leave"++    load' = loadRec $ do+        mdPrev <- loadRefs "SPREV"+        mdRoom <- loadRefs "room"+        mdFrom <- loadIdentity "from"+        mdReplyTo <- loadMbRef "reply-to"+        mdTime <- loadDate "time"+        mdText <- loadMbText "text"+        mdLeave <- isJust <$> loadMbEmpty "leave"+        return ChatMessageData {..}++threadToList :: [Stored (Signed ChatMessageData)] -> [ChatMessage]+threadToList thread = helper S.empty $ thread+  where+    helper :: S.Set (Stored (Signed ChatMessageData)) -> [Stored (Signed ChatMessageData)] -> [ChatMessage]+    helper seen msgs+        | msg : msgs' <- filter (`S.notMember` seen) $ reverse $ sortBy (comparing cmpView) msgs =+            messageFromData msg : helper (S.insert msg seen) (msgs' ++ mdPrev (fromSigned msg))+        | otherwise = []+    cmpView msg = (zonedTimeToUTC $ mdTime $ fromSigned msg, msg)++    messageFromData :: Stored (Signed ChatMessageData) -> ChatMessage+    messageFromData sdata = ChatMessage { cmsgData = sdata }++chatroomMessageByStateData+    :: (MonadStorage m, MonadHead LocalState m, MonadError String m)+    => Stored ChatroomStateData -> Text -> m ()+chatroomMessageByStateData lookupData msg = void $ findAndUpdateChatroomState $ \cstate -> do+    guard $ any (lookupData `precedesOrEquals`) $ roomStateData cstate+    Just $ do+        self <- finalOwner . localIdentity . fromStored <$> getLocalHead+        secret <- loadKey $ idKeyMessage self+        time <- liftIO getZonedTime+        mdata <- mstore =<< sign secret =<< mstore ChatMessageData+            { mdPrev = roomStateMessageData cstate+            , mdRoom = []+            , mdFrom = self+            , mdReplyTo = Nothing+            , mdTime = time+            , mdText = Just msg+            , mdLeave = False+            }+        mergeSorted . (:[]) <$> mstore ChatroomStateData+            { rsdPrev = roomStateData cstate+            , rsdRoom = []+            , rsdMessages = [ mdata ]+            }++ data ChatroomStateData = ChatroomStateData     { rsdPrev :: [Stored ChatroomStateData]     , rsdRoom :: [Stored (Signed ChatroomData)]+    , rsdMessages :: [Stored (Signed ChatMessageData)]     }  data ChatroomState = ChatroomState     { roomStateData :: [Stored ChatroomStateData]     , roomStateRoom :: Maybe Chatroom+    , roomStateMessageData :: [Stored (Signed ChatMessageData)]+    , roomStateMessages :: [ChatMessage]     }  instance Storable ChatroomStateData where     store' ChatroomStateData {..} = storeRec $ do         forM_ rsdPrev $ storeRef "PREV"         forM_ rsdRoom $ storeRef "room"+        forM_ rsdMessages $ storeRef "msg"      load' = loadRec $ do         rsdPrev <- loadRefs "PREV"         rsdRoom <- loadRefs "room"+        rsdMessages <- loadRefs "msg"         return ChatroomStateData {..}  instance Mergeable ChatroomState where     type Component ChatroomState = ChatroomStateData -    mergeSorted cdata = ChatroomState-        { roomStateData = cdata-        , roomStateRoom = either (const Nothing) Just $ runExcept $-            validateChatroom $ concat $ findProperty ((\case [] -> Nothing; xs -> Just xs) . rsdRoom) cdata-        }+    mergeSorted roomStateData =+        let roomStateRoom = either (const Nothing) Just $ runExcept $+                validateChatroom $ concat $ findProperty ((\case [] -> Nothing; xs -> Just xs) . rsdRoom) roomStateData+            roomStateMessageData = filterAncestors $ concat $ flip findProperty roomStateData $ \case+                ChatroomStateData {..} | null rsdMessages -> Nothing+                                       | otherwise        -> Just rsdMessages+            roomStateMessages = threadToList $ concatMap (rsdMessages . fromStored) roomStateData+         in ChatroomState {..}      toComponents = roomStateData @@ -129,6 +232,7 @@     cstate <- mergeSorted . (:[]) <$> mstore ChatroomStateData         { rsdPrev = []         , rsdRoom = [ rdata ]+        , rsdMessages = []         }      updateLocalHead $ updateSharedState $ \rooms -> do@@ -174,6 +278,7 @@         mergeSorted . (:[]) <$> mstore ChatroomStateData             { rsdPrev = roomStateData cstate             , rsdRoom = [ rdata ]+            , rsdMessages = []             }  @@ -193,7 +298,13 @@ findChatroomByStateData :: MonadHead LocalState m => Stored ChatroomStateData -> m (Maybe ChatroomState) findChatroomByStateData cdata = findChatroom $ any (cdata `precedesOrEquals`) . roomStateData +getMessagesSinceState :: ChatroomState -> ChatroomState -> [ChatMessage]+getMessagesSinceState cur old = takeWhile notOld (roomStateMessages cur)+  where+    notOld msg = cmsgData msg `notElem` roomStateMessageData old+    -- TODO: parallel message threads + data ChatroomSetChange = AddedChatroom ChatroomState                        | RemovedChatroom ChatroomState                        | UpdatedChatroom ChatroomState ChatroomState@@ -231,22 +342,26 @@ data ChatroomService = ChatroomService     { chatRoomQuery :: Bool     , chatRoomInfo :: [Stored (Signed ChatroomData)]+    , chatRoomMessage :: [Stored (Signed ChatMessageData)]     }  emptyPacket :: ChatroomService emptyPacket = ChatroomService     { chatRoomQuery = False     , chatRoomInfo = []+    , chatRoomMessage = []     }  instance Storable ChatroomService where     store' ChatroomService {..} = storeRec $ do         when  chatRoomQuery $ storeEmpty "room-query"         forM_ chatRoomInfo $ storeRef "room-info"+        forM_ chatRoomMessage $ storeRef "room-message"      load' = loadRec $ do         chatRoomQuery <- isJust <$> loadMbEmpty "room-query"         chatRoomInfo <- loadRefs "room-info"+        chatRoomMessage <- loadRefs "room-message"         return ChatroomService {..}  data PeerState = PeerState@@ -288,7 +403,11 @@                         -- update local state only if we got roomInfo not present there                         if roomInfo `notElem` prevRoom && roomInfo `elem` room                           then do-                            sdata <- mstore ChatroomStateData { rsdPrev = prev, rsdRoom = room }+                            sdata <- mstore ChatroomStateData+                                { rsdPrev = prev+                                , rsdRoom = room+                                , rsdMessages = []+                                }                             storeSetAddComponent sdata set                           else return set                 foldM upd roomSet chatRoomInfo

@@ -0,0 +1,75 @@+module Erebos.Conversation (+    Message,+    messageFrom,+    messageText,+    messageUnread,+    formatMessage,++    Conversation,+    directMessageConversation,+    reloadConversation,+    lookupConversations,++    conversationName,+    conversationPeer,+    conversationHistory,++    sendMessage,+) where++import Control.Monad.Except++import Data.List+import Data.Maybe+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time.LocalTime++import Erebos.Identity+import Erebos.Message hiding (formatMessage)+import Erebos.State+import Erebos.Storage+++data Message = DirectMessageMessage DirectMessage Bool++messageFrom :: Message -> ComposedIdentity+messageFrom (DirectMessageMessage msg _) = msgFrom msg++messageText :: Message -> Maybe Text+messageText (DirectMessageMessage msg _) = Just $ msgText msg++messageUnread :: Message -> Bool+messageUnread (DirectMessageMessage _ unread) = unread++formatMessage :: TimeZone -> Message -> String+formatMessage tzone (DirectMessageMessage msg _) = formatDirectMessage tzone msg+++data Conversation = DirectMessageConversation DirectMessageThread++directMessageConversation :: MonadHead LocalState m => ComposedIdentity -> m Conversation+directMessageConversation peer = do+    (find (sameIdentity peer . msgPeer) . toThreadList . lookupSharedValue . lsShared . fromStored <$> getLocalHead) >>= \case+        Just thread -> return $ DirectMessageConversation thread+        Nothing -> return $ DirectMessageConversation $ DirectMessageThread peer [] [] []++reloadConversation :: MonadHead LocalState m => Conversation -> m Conversation+reloadConversation (DirectMessageConversation thread) = directMessageConversation (msgPeer thread)++lookupConversations :: MonadHead LocalState m => m [Conversation]+lookupConversations = map DirectMessageConversation . toThreadList . lookupSharedValue . lsShared . fromStored <$> getLocalHead+++conversationName :: Conversation -> Text+conversationName (DirectMessageConversation thread) = fromMaybe (T.pack "<unnamed>") $ idName $ msgPeer thread++conversationPeer :: Conversation -> Maybe ComposedIdentity+conversationPeer (DirectMessageConversation thread) = Just $ msgPeer thread++conversationHistory :: Conversation -> [Message]+conversationHistory (DirectMessageConversation thread) = map (\msg -> DirectMessageMessage msg False) $ threadToList thread+++sendMessage :: (MonadHead LocalState m, MonadError String m) => Conversation -> Text -> m Message+sendMessage (DirectMessageConversation thread) text = DirectMessageMessage <$> (fromStored <$> sendDirectMessage (msgPeer thread) text) <*> pure False

@@ -4,6 +4,7 @@ ) where  import Control.Concurrent+import Control.Monad import Control.Monad.Except import Control.Monad.Reader 

@@ -14,6 +14,7 @@      watchReceivedMessages,     formatMessage,+    formatDirectMessage, ) where  import Control.Monad@@ -126,7 +127,7 @@ instance Storable MessageState where     store' MessageState {..} = storeRec $ do         mapM_ (storeRef "PREV") msPrev-        mapM_ (storeRef "peer") $ idDataF msPeer+        mapM_ (storeRef "peer") $ idExtDataF msPeer         mapM_ (storeRef "ready") msReady         mapM_ (storeRef "sent") msSent         mapM_ (storeRef "received") msReceived@@ -258,8 +259,12 @@         forM_ (map fromStored sms) $ \ms -> do             mapM_ f $ filter (not . sameIdentity self . msgFrom . fromStored) $ msReceived ms +{-# DEPRECATED formatMessage "use formatDirectMessage instead" #-} formatMessage :: TimeZone -> DirectMessage -> String-formatMessage tzone msg = concat+formatMessage = formatDirectMessage++formatDirectMessage :: TimeZone -> DirectMessage -> String+formatDirectMessage tzone msg = concat     [ formatTime defaultTimeLocale "[%H:%M] " $ utcToLocalTime tzone $ zonedTimeToUTC $ msgTime msg     , maybe "<unnamed>" T.unpack $ idName $ msgFrom msg     , ": "

@@ -4,6 +4,7 @@     Server,     startServer,     stopServer,+    getCurrentPeerList,     getNextPeerChange,     ServerOptions(..), serverIdentity, defaultServerOptions, @@ -16,6 +17,8 @@ #ifdef ENABLE_ICE_SUPPORT     serverPeerIce, #endif+    dropPeer,+    isPeerDropped,     sendToPeer, sendToPeerStored, sendToPeerWith,     runPeerService, @@ -90,6 +93,9 @@ serverIdentity :: Server -> IO UnifiedIdentity serverIdentity = readMVar . serverIdentity_ +getCurrentPeerList :: Server -> IO [Peer]+getCurrentPeerList = fmap M.elems . readMVar . serverPeers+ getNextPeerChange :: Server -> IO Peer getNextPeerChange = atomically . readTChan . serverChanPeer @@ -108,7 +114,7 @@ data Peer = Peer     { peerAddress :: PeerAddress     , peerServer_ :: Server-    , peerConnection :: TVar (Either [(Bool, TransportPacket Ref, [TransportHeaderItem])] (Connection PeerAddress))+    , peerState :: TVar PeerState     , peerIdentityVar :: TVar PeerIdentity     , peerStorage_ :: Storage     , peerInStorage :: PartialStorage@@ -123,13 +129,18 @@ peerStorage = peerStorage_  getPeerChannel :: Peer -> STM ChannelState-getPeerChannel Peer {..} = either (const $ return ChannelNone) connGetChannel =<< readTVar peerConnection+getPeerChannel Peer {..} =+    readTVar peerState >>= \case+        PeerInit _ -> return ChannelNone+        PeerConnected conn -> connGetChannel conn+        PeerDropped -> return ChannelClosed  setPeerChannel :: Peer -> ChannelState -> STM () setPeerChannel Peer {..} ch = do-    readTVar peerConnection >>= \case-        Left _ -> retry-        Right conn -> connSetChannel conn ch+    readTVar peerState >>= \case+        PeerInit _ -> retry+        PeerConnected conn -> connSetChannel conn ch+        PeerDropped -> return ()  instance Eq Peer where     (==) = (==) `on` peerIdentityVar@@ -175,6 +186,11 @@ peerIdentity = liftIO . atomically . readTVar . peerIdentityVar  +data PeerState = PeerInit [(SecurityRequirement, TransportPacket Ref, [TransportHeaderItem])]+               | PeerConnected (Connection PeerAddress)+               | PeerDropped++ lookupServiceType :: [TransportHeaderItem] -> Maybe ServiceID lookupServiceType (ServiceType stype : _) = Just stype lookupServiceType (_ : hs) = lookupServiceType hs@@ -196,8 +212,13 @@   forkServerThread :: Server -> IO () -> IO ()-forkServerThread server act = modifyMVar_ (serverThreads server) $ \ts -> do-    (:ts) <$> forkIO act+forkServerThread server act = do+    modifyMVar_ (serverThreads server) $ \ts -> do+        t <- forkIO $ do+            t <- myThreadId+            act+            modifyMVar_ (serverThreads server) $ return . filter (/=t)+        return (t:ts)  startServer :: ServerOptions -> Head LocalState -> (String -> IO ()) -> [SomeService] -> IO Server startServer opt serverOrigHead logd' serverServices = do@@ -299,10 +320,14 @@                          forkServerThread server $ do                             atomically $ do-                                readTVar (peerConnection peer) >>= \case-                                    Left packets -> writeFlowBulk (connData conn) $ reverse packets-                                    Right _ -> return ()-                                writeTVar (peerConnection peer) (Right conn)+                                readTVar (peerState peer) >>= \case+                                    PeerInit packets -> do+                                        writeFlowBulk (connData conn) $ reverse packets+                                        writeTVar (peerState peer) (PeerConnected conn)+                                    PeerConnected _ -> do+                                        writeTVar (peerState peer) (PeerConnected conn)+                                    PeerDropped -> do+                                        connClose conn                              case mbpid of                                 Just dgst -> do@@ -317,12 +342,17 @@                                             _ -> return ()                                 Nothing -> return () -                            forever $ do-                                (secure, TransportPacket header objs) <- readFlowIO $ connData conn-                                prefs <- forM objs $ storeObject $ peerInStorage peer-                                identity <- readMVar serverIdentity_-                                let svcs = map someServiceID serverServices-                                handlePacket identity secure peer chanSvc svcs header prefs+                            let peerLoop = readFlowIO (connData conn) >>= \case+                                    Just (secure, TransportPacket header objs) -> do+                                        prefs <- forM objs $ storeObject $ peerInStorage peer+                                        identity <- readMVar serverIdentity_+                                        let svcs = map someServiceID serverServices+                                        handlePacket identity secure peer chanSvc svcs header prefs+                                        peerLoop+                                    Nothing -> do+                                        dropPeer peer+                                        atomically $ writeTChan serverChanPeer peer+                            peerLoop                      ReceivedAnnounce addr _ -> do                         void $ serverPeer' server addr@@ -391,13 +421,17 @@ runPacketHandler :: Bool -> Peer -> PacketHandler () -> STM () runPacketHandler secure peer@Peer {..} act = do     let logd = writeTQueue $ serverErrorLog peerServer_-    runExceptT (flip execStateT (PacketHandlerState peer [] [] []) $ unPacketHandler act) >>= \case+    runExceptT (flip execStateT (PacketHandlerState peer [] [] [] False) $ unPacketHandler act) >>= \case         Left err -> do             logd $ "Error in handling packet from " ++ show peerAddress ++ ": " ++ err         Right ph -> do             when (not $ null $ phHead ph) $ do                 let packet = TransportPacket (TransportHeader $ phHead ph) (phBody ph)-                sendToPeerS' secure peer (phAckedBy ph) packet+                    secreq = case (secure, phPlaintextReply ph) of+                                  (True, _) -> EncryptedOnly+                                  (False, False) -> PlaintextAllowed+                                  (False, True) -> PlaintextOnly+                sendToPeerS' secreq peer (phAckedBy ph) packet  liftSTM :: STM a -> PacketHandler a liftSTM = PacketHandler . lift . lift@@ -416,6 +450,7 @@     , phHead :: [TransportHeaderItem]     , phAckedBy :: [TransportHeaderItem]     , phBody :: [Ref]+    , phPlaintextReply :: Bool     }  addHeader :: TransportHeaderItem -> PacketHandler ()@@ -427,12 +462,15 @@ addBody :: Ref -> PacketHandler () addBody r = modify $ \ph -> ph { phBody = r `appendDistinct` phBody ph } +keepPlaintextReply :: PacketHandler ()+keepPlaintextReply = modify $ \ph -> ph { phPlaintextReply = True }+ openStream :: PacketHandler RawStreamWriter openStream = do     Peer {..} <- gets phPeer-    conn <- readTVarP peerConnection >>= \case-        Right conn -> return conn-        _          -> throwError "can't open stream without established connection"+    conn <- readTVarP peerState >>= \case+        PeerConnected conn -> return conn+        _                  -> throwError "can't open stream without established connection"     (hdr, writer, handler) <- liftSTM (connAddWriteStream conn) >>= \case         Right res -> return res         Left err -> throwError err@@ -444,9 +482,9 @@ acceptStream :: Word8 -> PacketHandler RawStreamReader acceptStream streamNumber = do     Peer {..} <- gets phPeer-    conn <- readTVarP peerConnection >>= \case-        Right conn -> return conn-        _          -> throwError "can't accept stream without established connection"+    conn <- readTVarP peerState >>= \case+        PeerConnected conn -> return conn+        _                  -> throwError "can't accept stream without established connection"     liftSTM $ connAddReadStream conn streamNumber  appendDistinct :: Eq a => a -> [a] -> [a]@@ -465,9 +503,9 @@             [ [ storedRef sidentity ]             , map storedRef $ idUpdates identity             , case ochannel of-                   ChannelOurRequest _ req  -> [ storedRef req ]-                   ChannelOurAccept _ acc _ -> [ storedRef acc ]-                   _                        -> []+                   ChannelOurRequest req  -> [ storedRef req ]+                   ChannelOurAccept acc _ -> [ storedRef acc ]+                   _                      -> []             ]      runPacketHandler secure peer $ do@@ -475,7 +513,7 @@         forM_ headers $ \case             Acknowledged dgst -> do                 liftSTM (getPeerChannel peer) >>= \case-                    ChannelOurAccept _ acc ch | refDigest (storedRef acc) == dgst -> do+                    ChannelOurAccept acc ch | refDigest (storedRef acc) == dgst -> do                         liftSTM $ finalizedChannel peer ch identity                     _ -> return () @@ -489,6 +527,11 @@                         partialRefFromDigest (peerInStorage peer) dgst                     addHeader $ DataResponse dgst                     addAckedBy [ Acknowledged dgst, Rejected dgst ]++                    -- Plaintext request may indicate the peer has restarted/changed or+                    -- otherwise lost the channel, so keep the reply plaintext as well.+                    when (not secure) keepPlaintextReply+                     let bytes = lazyLoadBytes mref                     -- TODO: MTU                     if (secure && BL.length bytes > 500)@@ -539,22 +582,23 @@                     _ -> return ()              TrChannelRequest dgst -> do-                let process cookie = do+                let process = do                         addHeader $ Acknowledged dgst                         wref <- newWaitingRef dgst $ handleChannelRequest peer identity-                        liftSTM $ setPeerChannel peer $ ChannelPeerRequest cookie wref+                        liftSTM $ setPeerChannel peer $ ChannelPeerRequest wref                     reject = addHeader $ Rejected dgst                  liftSTM (getPeerChannel peer) >>= \case                     ChannelNone {} -> return ()                     ChannelCookieWait {} -> return ()-                    ChannelCookieReceived cookie -> process $ Just cookie-                    ChannelCookieConfirmed cookie -> process $ Just cookie-                    ChannelOurRequest mbcookie our | dgst < refDigest (storedRef our) -> process mbcookie-                                                   | otherwise -> reject-                    ChannelPeerRequest mbcookie _ -> process mbcookie+                    ChannelCookieReceived {} -> process+                    ChannelCookieConfirmed {} -> process+                    ChannelOurRequest our | dgst < refDigest (storedRef our) -> process+                                          | otherwise -> reject+                    ChannelPeerRequest {} -> process                     ChannelOurAccept {} -> reject-                    ChannelEstablished {} -> process Nothing+                    ChannelEstablished {} -> process+                    ChannelClosed {} -> return ()              TrChannelAccept dgst -> do                 let process = do@@ -567,9 +611,10 @@                     ChannelCookieConfirmed {} -> reject                     ChannelOurRequest {} -> process                     ChannelPeerRequest {} -> process-                    ChannelOurAccept _ our _ | dgst < refDigest (storedRef our) -> process-                                             | otherwise -> addHeader $ Rejected dgst+                    ChannelOurAccept our _ | dgst < refDigest (storedRef our) -> process+                                           | otherwise -> addHeader $ Rejected dgst                     ChannelEstablished {} -> process+                    ChannelClosed {} -> return ()              ServiceType _ -> return ()             ServiceRef dgst@@ -604,10 +649,10 @@             ]     liftIO $ atomically $ do         getPeerChannel peer >>= \case-            ChannelCookieConfirmed cookie -> do+            ChannelCookieConfirmed -> do                 sendToPeerPlain peer [ Acknowledged reqref, Rejected reqref ] $                     TransportPacket (TransportHeader hitems) [storedRef req]-                setPeerChannel peer $ ChannelOurRequest (Just cookie) req+                setPeerChannel peer $ ChannelOurRequest req             _ -> return ()  handleChannelRequest :: Peer -> UnifiedIdentity -> Ref -> WaitingRefCallback@@ -616,8 +661,8 @@         (acc, ch) <- flip runReaderT (peerStorage peer) $ acceptChannelRequest identity upid (wrappedLoad req)         liftIO $ atomically $ do             getPeerChannel peer >>= \case-                ChannelPeerRequest mbcookie wr | wrDigest wr == refDigest req -> do-                    setPeerChannel peer $ ChannelOurAccept mbcookie acc ch+                ChannelPeerRequest wr | wrDigest wr == refDigest req -> do+                    setPeerChannel peer $ ChannelOurAccept acc ch                     let accref = refDigest $ storedRef acc                         header = TrChannelAccept accref                         ackedBy = [ Acknowledged accref, Rejected accref ]@@ -708,7 +753,7 @@  mkPeer :: Server -> PeerAddress -> IO Peer mkPeer peerServer_ peerAddress = do-    peerConnection <- newTVarIO (Left [])+    peerState <- newTVarIO (PeerInit [])     peerIdentityVar <- newTVarIO . PeerIdentityUnknown =<< newTVarIO []     peerStorage_ <- deriveEphemeralStorage $ serverStorage peerServer_     peerInStorage <- derivePartialStorage peerStorage_@@ -718,7 +763,11 @@  serverPeer :: Server -> SockAddr -> IO Peer serverPeer server paddr = do-    serverPeer' server (DatagramAddress paddr)+    let paddr' = case IP.fromSockAddr paddr of+            Just (IP.IPv4 ipv4, port)+                -> IP.toSockAddr (IP.IPv6 $ IP.toIPv6w (0, 0, 0xffff, IP.fromIPv4w ipv4), port)+            _   -> paddr+    serverPeer' server (DatagramAddress paddr')  #ifdef ENABLE_ICE_SUPPORT serverPeerIce :: Server -> IceSession -> IO Peer@@ -741,7 +790,21 @@         writeFlow (serverControlFlow server) (RequestConnection paddr)     return peer +dropPeer :: MonadIO m => Peer -> m ()+dropPeer peer = liftIO $ do+    modifyMVar_ (serverPeers $ peerServer peer) $ \pvalue -> do+        atomically $ do+            readTVar (peerState peer) >>= \case+                PeerConnected conn -> connClose conn+                _ -> return()+            writeTVar (peerState peer) PeerDropped+        return $ M.delete (peerAddress peer) pvalue +isPeerDropped :: MonadIO m => Peer -> m Bool+isPeerDropped peer = liftIO $ atomically $ readTVar (peerState peer) >>= \case+    PeerDropped -> return True+    _           -> return False+ sendToPeer :: (Service s, MonadIO m) => Peer -> s -> m () sendToPeer peer packet = sendToPeerList peer [ServiceReply (Left packet) True] @@ -762,17 +825,18 @@         ackedBy = concat [[ Acknowledged r, Rejected r, DataRequest r ] | r <- dgsts ]     liftIO $ atomically $ sendToPeerS peer ackedBy packet -sendToPeerS' :: Bool -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()+sendToPeerS' :: SecurityRequirement -> Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM () sendToPeerS' secure Peer {..} ackedBy packet = do-    readTVar peerConnection >>= \case-        Left xs -> writeTVar peerConnection $ Left $ (secure, packet, ackedBy) : xs-        Right conn -> writeFlow (connData conn) (secure, packet, ackedBy)+    readTVar peerState >>= \case+        PeerInit xs -> writeTVar peerState $ PeerInit $ (secure, packet, ackedBy) : xs+        PeerConnected conn -> writeFlow (connData conn) (secure, packet, ackedBy)+        PeerDropped -> return ()  sendToPeerS :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()-sendToPeerS = sendToPeerS' True+sendToPeerS = sendToPeerS' EncryptedOnly  sendToPeerPlain :: Peer -> [TransportHeaderItem] -> TransportPacket Ref -> STM ()-sendToPeerPlain = sendToPeerS' False+sendToPeerPlain = sendToPeerS' PlaintextAllowed  sendToPeerWith :: forall s m. (Service s, MonadIO m, MonadError String m) => Peer -> (ServiceState s -> ExceptT String IO (Maybe s, ServiceState s)) -> m () sendToPeerWith peer fobj = do

@@ -3,6 +3,7 @@     transportToObject,     TransportHeader(..),     TransportHeaderItem(..),+    SecurityRequirement(..),      WaitingRef(..),     WaitingRefCallback,@@ -19,6 +20,7 @@     connData,     connGetChannel,     connSetChannel,+    connClose,      RawStreamReader, RawStreamWriter,     connAddWriteStream,@@ -43,6 +45,7 @@ import Data.ByteString qualified as B import Data.ByteString.Char8 qualified as BC import Data.ByteString.Lazy qualified as BL+import Data.Function import Data.List import Data.Maybe import Data.Text (Text)@@ -93,6 +96,11 @@ newtype Cookie = Cookie ByteString     deriving (Eq, Show) +data SecurityRequirement = PlaintextOnly+                         | PlaintextAllowed+                         | EncryptedOnly+    deriving (Eq, Ord)+ isHeaderItemAcknowledged :: TransportHeaderItem -> Bool isHeaderItemAcknowledged = \case     Acknowledged {} -> False@@ -165,10 +173,15 @@ data Connection addr = Connection     { cGlobalState :: GlobalState addr     , cAddress :: addr-    , cDataUp :: Flow (Bool, TransportPacket PartialObject) (Bool, TransportPacket Ref, [TransportHeaderItem])-    , cDataInternal :: Flow (Bool, TransportPacket Ref, [TransportHeaderItem]) (Bool, TransportPacket PartialObject)+    , cDataUp :: Flow+        (Maybe (Bool, TransportPacket PartialObject))+        (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem])+    , cDataInternal :: Flow+        (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem])+        (Maybe (Bool, TransportPacket PartialObject))     , cChannel :: TVar ChannelState-    , cSecureOutQueue :: TQueue (Bool, TransportPacket Ref, [TransportHeaderItem])+    , cCookie :: TVar (Maybe Cookie)+    , cSecureOutQueue :: TQueue (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem])     , cMaxInFlightPackets :: TVar Int     , cReservedPackets :: TVar Int     , cSentPackets :: TVar [SentPacket]@@ -177,10 +190,15 @@     , cOutStreams :: TVar [(Word8, Stream)]     } +instance Eq (Connection addr) where+    (==) = (==) `on` cChannel+ connAddress :: Connection addr -> addr connAddress = cAddress -connData :: Connection addr -> Flow (Bool, TransportPacket PartialObject) (Bool, TransportPacket Ref, [TransportHeaderItem])+connData :: Connection addr -> Flow+    (Maybe (Bool, TransportPacket PartialObject))+    (SecurityRequirement, TransportPacket Ref, [TransportHeaderItem]) connData = cDataUp  connGetChannel :: Connection addr -> STM ChannelState@@ -190,6 +208,16 @@ connSetChannel Connection {..} ch = do     writeTVar cChannel ch +connClose :: Connection addr -> STM ()+connClose conn@Connection {..} = do+    let GlobalState {..} = cGlobalState+    readTVar cChannel >>= \case+        ChannelClosed -> return ()+        _ -> do+            writeTVar cChannel ChannelClosed+            writeTVar gConnections . filter (/=conn) =<< readTVar gConnections+            writeFlow cDataInternal Nothing+ connAddWriteStream :: Connection addr -> STM (Either String (TransportHeaderItem, RawStreamWriter, IO ())) connAddWriteStream conn@Connection {..} = do     outStreams <- readTVar cOutStreams@@ -234,9 +262,9 @@                 mbReserved = Just reserved              mbch <- atomically (readTVar cChannel) >>= return . \case-                ChannelEstablished ch   -> Just ch-                ChannelOurAccept _ _ ch -> Just ch-                _                       -> Nothing+                ChannelEstablished ch -> Just ch+                ChannelOurAccept _ ch -> Just ch+                _                     -> Nothing              mbs <- case mbch of                 Just ch -> do@@ -367,16 +395,18 @@  data ChannelState = ChannelNone                   | ChannelCookieWait -- sent initiation, waiting for response-                  | ChannelCookieReceived Cookie -- received cookie, but no cookie echo yet-                  | ChannelCookieConfirmed Cookie -- received cookie echo, no need to send from our side-                  | ChannelOurRequest (Maybe Cookie) (Stored ChannelRequest)-                  | ChannelPeerRequest (Maybe Cookie) WaitingRef-                  | ChannelOurAccept (Maybe Cookie) (Stored ChannelAccept) Channel+                  | ChannelCookieReceived -- received cookie, but no cookie echo yet+                  | ChannelCookieConfirmed -- received cookie echo, no need to send from our side+                  | ChannelOurRequest (Stored ChannelRequest)+                  | ChannelPeerRequest WaitingRef+                  | ChannelOurAccept (Stored ChannelAccept) Channel                   | ChannelEstablished Channel+                  | ChannelClosed  data ReservedToSend = ReservedToSend     { rsAckedBy :: Maybe (TransportHeaderItem -> Bool)     , rsOnAck :: IO ()+    , rsOnFail :: IO ()     }  data SentPacket = SentPacket@@ -384,6 +414,7 @@     , spRetryCount :: Int     , spAckedBy :: Maybe (TransportHeaderItem -> Bool)     , spOnAck :: IO ()+    , spOnFail :: IO ()     , spData :: BC.ByteString     } @@ -450,6 +481,7 @@     let cAddress = addr     (cDataUp, cDataInternal) <- newFlow     cChannel <- newTVar ChannelNone+    cCookie <- newTVar Nothing     cSecureOutQueue <- newTQueue     cMaxInFlightPackets <- newTVar 4     cReservedPackets <- newTVar 0@@ -465,7 +497,7 @@ passUpIncoming :: GlobalState addr -> STM (IO ()) passUpIncoming GlobalState {..} = do     (Connection {..}, up) <- takeTMVar gNextUp-    writeFlow cDataInternal up+    writeFlow cDataInternal (Just up)     return $ return ()  processIncoming :: GlobalState addr -> STM (IO ())@@ -479,9 +511,9 @@     mbch <- case mbconn of         Nothing -> return Nothing         Just conn -> readTVar (cChannel conn) >>= return . \case-            ChannelEstablished ch   -> Just ch-            ChannelOurAccept _ _ ch -> Just ch-            _                       -> Nothing+            ChannelEstablished ch -> Just ch+            ChannelOurAccept _ ch -> Just ch+            _                     -> Nothing      return $ do         let deserialize = liftEither . runExcept . deserializeObjects gStorage . BL.fromStrict@@ -576,11 +608,12 @@         True -> do             atomically $ do                 conn@Connection {..} <- getConnection gs addr-                channel <- readTVar cChannel+                oldCookie <- readTVar cCookie                 let received = listToMaybe $ mapMaybe (\case CookieSet x -> Just x; _ -> Nothing) header-                case received `mplus` channelCurrentCookie channel of+                case received `mplus` oldCookie of                     Just current -> do-                        cookieEchoReceived gs conn mbpid current+                        writeTVar cCookie (Just current)+                        cookieEchoReceived gs conn mbpid                         return $ Just (conn, Just packet)                     Nothing -> do                         gLog $ show addr <> ": missing cookie set, dropping " <> show header@@ -597,7 +630,8 @@     -> do         atomically $ readTVar cChannel >>= \case             ChannelCookieWait -> do-                writeTVar cChannel $ ChannelCookieReceived cookie+                writeTVar cChannel $ ChannelCookieReceived+                writeTVar cCookie $ Just cookie                 writeFlow gControlFlow (NewConnection conn mbpid)                 return $ Just (conn, Nothing)             _ -> return Nothing@@ -636,17 +670,8 @@     mbpid = listToMaybe $ mapMaybe (\case AnnounceSelf dgst -> Just dgst; _ -> Nothing) header     version = listToMaybe $ filter (\v -> ProtocolVersion v `elem` header) protocolVersions -channelCurrentCookie :: ChannelState -> Maybe Cookie-channelCurrentCookie = \case-    ChannelCookieReceived cookie -> Just cookie-    ChannelCookieConfirmed cookie -> Just cookie-    ChannelOurRequest mbcookie _ -> mbcookie-    ChannelPeerRequest mbcookie _ -> mbcookie-    ChannelOurAccept mbcookie _ _ -> mbcookie-    _ -> Nothing--cookieEchoReceived :: GlobalState addr -> Connection addr -> Maybe RefDigest -> Cookie -> STM ()-cookieEchoReceived GlobalState {..} conn@Connection {..} mbpid cookieSet = do+cookieEchoReceived :: GlobalState addr -> Connection addr -> Maybe RefDigest -> STM ()+cookieEchoReceived GlobalState {..} conn@Connection {..} mbpid = do     readTVar cChannel >>= \case         ChannelNone -> newConn         ChannelCookieWait -> newConn@@ -654,18 +679,18 @@         _ -> return ()   where     update = do-        writeTVar cChannel $ ChannelCookieConfirmed cookieSet+        writeTVar cChannel ChannelCookieConfirmed     newConn = do         update         writeFlow gControlFlow (NewConnection conn mbpid) -generateCookieHeaders :: GlobalState addr -> addr -> ChannelState -> IO [TransportHeaderItem]-generateCookieHeaders gs addr ch = catMaybes <$> sequence [ echoHeader, setHeader ]+generateCookieHeaders :: Connection addr -> ChannelState -> IO [TransportHeaderItem]+generateCookieHeaders Connection {..} ch = catMaybes <$> sequence [ echoHeader, setHeader ]   where-    echoHeader = return $ CookieEcho <$> channelCurrentCookie ch+    echoHeader = fmap CookieEcho <$> atomically (readTVar cCookie)     setHeader = case ch of-        ChannelCookieWait {} -> Just . CookieSet <$> createCookie gs addr-        ChannelCookieReceived {} -> Just . CookieSet <$> createCookie gs addr+        ChannelCookieWait {} -> Just . CookieSet <$> createCookie cGlobalState cAddress+        ChannelCookieReceived {} -> Just . CookieSet <$> createCookie cGlobalState cAddress         _ -> return Nothing  createCookie :: GlobalState addr -> addr -> IO Cookie@@ -676,7 +701,7 @@   reservePacket :: Connection addr -> STM ReservedToSend-reservePacket Connection {..} = do+reservePacket conn@Connection {..} = do     maxPackets <- readTVar cMaxInFlightPackets     reserved <- readTVar cReservedPackets     sent <- length <$> readTVar cSentPackets@@ -685,7 +710,7 @@         retry      writeTVar cReservedPackets $ reserved + 1-    return $ ReservedToSend Nothing (return ())+    return $ ReservedToSend Nothing (return ()) (atomically $ connClose conn)  resendBytes :: Connection addr -> Maybe ReservedToSend -> SentPacket -> IO () resendBytes Connection {..} reserved sp = do@@ -709,6 +734,7 @@         , spRetryCount = -1         , spAckedBy = rsAckedBy =<< reserved         , spOnAck = maybe (return ()) rsOnAck reserved+        , spOnFail = maybe (return ()) rsOnFail reserved         , spData = bs         } @@ -734,13 +760,13 @@                     | isJust mbch = do                         acks <- readTVar cToAcknowledge                         if null acks then retry-                                     else return ((True, TransportPacket (TransportHeader []) [], []), Nothing)+                                     else return ((EncryptedOnly, TransportPacket (TransportHeader []) [], []), Nothing)                     | otherwise = retry              ((secure, packet@(TransportPacket (TransportHeader hitems) content), plainAckedBy), mbReserved) <-                 checkOutstanding <|> checkDataInternal <|> checkAcknowledgements -            when (isNothing mbch && secure) $ do+            when (isNothing mbch && secure >= EncryptedOnly) $ do                 writeTQueue cSecureOutQueue (secure, packet, plainAckedBy)              acknowledge <- case mbch of@@ -748,28 +774,34 @@                 Just _ -> swapTVar cToAcknowledge []              return $ do-                cookieHeaders <- generateCookieHeaders gs cAddress channel-                let header = TransportHeader $ map AcknowledgedSingle acknowledge ++ cookieHeaders ++ hitems+                let onAck = sequence_ $ map (streamAccepted conn) $+                        catMaybes (map (\case StreamOpen n -> Just n; _ -> Nothing) hitems) -                let plain = BL.concat $-                        (serializeObject $ transportToObject gStorage header)-                        : map lazyLoadBytes content+                let mkPlain extraHeaders =+                        let header = TransportHeader $ map AcknowledgedSingle acknowledge ++ extraHeaders ++ hitems+                         in BL.concat $+                            (serializeObject $ transportToObject gStorage header)+                            : map lazyLoadBytes content -                let onAck = case catMaybes (map (\case StreamOpen n -> Just n; _ -> Nothing) hitems) of-                                 [] -> return ()-                                 xs -> sequence_ $ map (streamAccepted conn) xs+                let usePlaintext = do+                        plain <- mkPlain <$> generateCookieHeaders conn channel+                        return $ Just (BL.toStrict plain, plainAckedBy) -                mbs <- case mbch of-                    Just ch -> do+                let useEncryption ch = do+                        plain <- mkPlain <$> return []                         runExceptT (channelEncrypt ch $ BL.toStrict $ 0x00 `BL.cons` plain) >>= \case                             Right (ctext, counter) -> do                                 let isAcked = any isHeaderItemAcknowledged hitems                                 return $ Just (0x80 `B.cons` ctext, if isAcked then [ AcknowledgedSingle $ fromIntegral counter ] else [])                             Left err -> do atomically $ gLog $ "Failed to encrypt data: " ++ err                                            return Nothing-                    Nothing | secure    -> return Nothing-                            | otherwise -> return $ Just (BL.toStrict plain, plainAckedBy) +                mbs <- case (secure, mbch) of+                    (PlaintextOnly, _)          -> usePlaintext+                    (PlaintextAllowed, Nothing) -> usePlaintext+                    (_, Just ch)                -> useEncryption ch+                    (EncryptedOnly, Nothing)    -> return Nothing+                 case mbs of                     Just (bs, ackedBy) -> do                         let mbReserved' = (\rs -> rs@@ -786,17 +818,18 @@                 sps@(_:_) -> return (last sps, init sps)                 _         -> retry             let nextTry = spTime sp + fromNanoSecs 1000000000-            if now < nextTry-              then do-                nextTimeout <- readTVar gNextTimeout-                if nextTimeout <= now || nextTry < nextTimeout-                   then do writeTVar gNextTimeout nextTry-                           return $ return ()-                   else retry-              else do-                reserved <- reservePacket conn-                writeTVar cSentPackets rest-                return $ resendBytes conn (Just reserved) sp+            if | now < nextTry -> do+                    nextTimeout <- readTVar gNextTimeout+                    if nextTimeout <= now || nextTry < nextTimeout+                       then do writeTVar gNextTimeout nextTry+                               return $ return ()+                       else retry+               | spRetryCount sp < 2 -> do+                    reserved <- reservePacket conn+                    writeTVar cSentPackets rest+                    return $ resendBytes conn (Just reserved) sp+               | otherwise -> do+                    return $ spOnFail sp      let handleControlRequests = readFlow gControlFlow >>= \case             RequestConnection addr -> do

@@ -112,10 +112,29 @@ type Storage = Storage' Complete type PartialStorage = Storage' Partial +storageVersion :: String+storageVersion = "0.1"+ openStorage :: FilePath -> IO Storage-openStorage path = do-    createDirectoryIfMissing True $ path ++ "/objects"-    createDirectoryIfMissing True $ path ++ "/heads"+openStorage path = modifyIOError annotate $ do+    let versionPath = path </> "erebos-storage"+    doesFileExist versionPath >>= \case+        True -> readFile versionPath >>= \case+            content | (ver:_) <- lines content, ver == storageVersion -> return ()+                    | otherwise -> fail "unsupported storage version"+        False -> do+            doesDirectoryExist path >>= \case+                True -> do+                    listDirectory path >>= \case+                        contents@(_:_) | "objects" `notElem` contents || "heads" `notElem` contents+                            -> fail "directory is neither empty, nor an existing erebos storage"+                        _ -> return ()+                False -> do+                    createDirectoryIfMissing True $ path+            writeFile versionPath $ storageVersion <> "\n"++    createDirectoryIfMissing True $ path </> "objects"+    createDirectoryIfMissing True $ path </> "heads"     watchers <- newMVar ([], WatchList 1 [])     refgen <- newMVar =<< HT.new     refroots <- newMVar =<< HT.new@@ -125,6 +144,8 @@         , stRefGeneration = refgen         , stRefRoots = refroots         }+  where+    annotate e = annotateIOError e "failed to open storage" Nothing (Just path)  memoryStorage' :: IO (Storage' c') memoryStorage' = do