mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-07-28 20:50:10 +00:00
ntf: agent ntf workers implementation, store methods (#394)
This commit is contained in:
JRoberts
@@ -48,7 +48,6 @@ library
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220404_ntf_subscriptions_draft
|
||||
Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220608_v2
|
||||
Simplex.Messaging.Client
|
||||
Simplex.Messaging.Client.Agent
|
||||
|
||||
@@ -239,14 +239,14 @@ processCommand c (connId, cmd) = case cmd of
|
||||
newConn :: AgentMonad m => AgentClient -> ConnId -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
|
||||
newConn c connId cMode = do
|
||||
srv <- getSMPServer c
|
||||
(rq, qUri) <- newRcvQueueNtf c srv
|
||||
(rq, qUri) <- newRcvQueueNtf c srv connId
|
||||
g <- asks idsDrg
|
||||
agentVersion <- asks $ smpAgentVersion . config
|
||||
let cData = ConnData {connId, connAgentVersion = agentVersion, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent
|
||||
connId' <- withStore c $ \st -> createRcvConn st g cData rq cMode
|
||||
addSubscription c rq connId'
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (rq, NSCCreate)
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
aVRange <- asks $ smpAgentVRange . config
|
||||
let crData = ConnReqUriData simplexChat aVRange [qUri]
|
||||
case cMode of
|
||||
@@ -301,7 +301,7 @@ joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInf
|
||||
createReplyQueue :: AgentMonad m => AgentClient -> ConnId -> m SMPQueueInfo
|
||||
createReplyQueue c connId = do
|
||||
srv <- getSMPServer c
|
||||
(rq, qUri) <- newRcvQueueNtf c srv
|
||||
(rq, qUri) <- newRcvQueueNtf c srv connId
|
||||
-- TODO reply queue version should be the same as send queue, ignoring it in v1
|
||||
let qInfo = toVersionT qUri SMP.smpClientVersion
|
||||
addSubscription c rq connId
|
||||
@@ -361,7 +361,7 @@ subscribeConnection' c connId =
|
||||
subscribe rq = do
|
||||
subscribeQueue c rq connId
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (rq, NSCCreate)
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
|
||||
resubscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
|
||||
resubscribeConnection' c connId =
|
||||
@@ -564,7 +564,7 @@ deleteConnection' c connId =
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ do
|
||||
removeSubscription c connId
|
||||
sendNtfSubCommand ns (rq, NSCDelete)
|
||||
sendNtfSubCommand ns (connId, NSCDelete)
|
||||
withStore c (`deleteConn` connId)
|
||||
|
||||
-- | Change servers to be used for creating new queues, in Reader monad
|
||||
@@ -696,9 +696,7 @@ initializeNtfSubQ c tkn = do
|
||||
connIds <- atomically $ do
|
||||
nsUpdateToken ns tkn
|
||||
getSubscriptions c
|
||||
forM_ connIds $ \connId -> do
|
||||
rq <- withStore c $ \st -> getRcvQueue st connId
|
||||
atomically $ sendNtfSubCommand ns (rq, NSCCreate)
|
||||
forM_ connIds $ \connId -> atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
|
||||
-- TODO
|
||||
-- There should probably be another function to cancel all subscriptions that would flush the queue first,
|
||||
@@ -1010,11 +1008,11 @@ agentRatchetDecrypt st connId encAgentMsg = do
|
||||
updateRatchet st connId rc' skippedDiff
|
||||
liftEither $ first (SEAgentError . cryptoError) agentMsgBody_
|
||||
|
||||
newRcvQueueNtf :: AgentMonad m => AgentClient -> SMPServer -> m (RcvQueue, SMPQueueUri)
|
||||
newRcvQueueNtf c srv = do
|
||||
newRcvQueueNtf :: AgentMonad m => AgentClient -> SMPServer -> ConnId -> m (RcvQueue, SMPQueueUri)
|
||||
newRcvQueueNtf c srv connId = do
|
||||
(rq, qUri) <- newRcvQueue c srv
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (rq, NSCCreate)
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCCreate)
|
||||
pure (rq, qUri)
|
||||
|
||||
newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue
|
||||
|
||||
@@ -26,6 +26,7 @@ module Simplex.Messaging.Agent.Client
|
||||
sendInvitation,
|
||||
RetryInterval (..),
|
||||
secureQueue,
|
||||
enableQueueNotifications,
|
||||
sendAgentMessage,
|
||||
agentNtfRegisterToken,
|
||||
agentNtfVerifyToken,
|
||||
@@ -80,7 +81,7 @@ import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Notifications.Client
|
||||
import Simplex.Messaging.Notifications.Protocol
|
||||
import Simplex.Messaging.Protocol (BrokerMsg, ErrorType, MsgFlags (..), MsgId, NtfPrivateSignKey, ProtocolServer (..), QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
|
||||
import Simplex.Messaging.Protocol (BrokerMsg, ErrorType, MsgFlags (..), MsgId, NotifierId, NtfPrivateSignKey, NtfPublicVerifyKey, ProtocolServer (..), QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
@@ -413,7 +414,10 @@ newRcvQueue_ a c srv = do
|
||||
e2ePrivKey,
|
||||
e2eDhSecret = Nothing,
|
||||
sndId = Just sndId,
|
||||
status = New
|
||||
status = New,
|
||||
ntfPublicKey = Nothing,
|
||||
ntfPrivateKey = Nothing,
|
||||
notifierId = Nothing
|
||||
}
|
||||
pure (rq, SMPQueueUri srv sndId SMP.smpClientVRange e2eDhKey)
|
||||
|
||||
@@ -502,6 +506,11 @@ secureQueue c RcvQueue {server, rcvId, rcvPrivateKey} senderKey =
|
||||
withLogClient c server rcvId "KEY <key>" $ \smp ->
|
||||
secureSMPQueue smp rcvPrivateKey rcvId senderKey
|
||||
|
||||
enableQueueNotifications :: AgentMonad m => AgentClient -> RcvQueue -> NtfPublicVerifyKey -> m NotifierId
|
||||
enableQueueNotifications c RcvQueue {server, rcvId, rcvPrivateKey} notifierKey =
|
||||
withLogClient c server rcvId "NKEY <nkey>" $ \smp ->
|
||||
enableSMPQueueNotifications smp rcvPrivateKey rcvId notifierKey
|
||||
|
||||
sendAck :: AgentMonad m => AgentClient -> RcvQueue -> MsgId -> m ()
|
||||
sendAck c RcvQueue {server, rcvId, rcvPrivateKey} msgId =
|
||||
withLogClient c server rcvId "ACK" $ \smp ->
|
||||
|
||||
@@ -35,7 +35,6 @@ import Network.Socket
|
||||
import Numeric.Natural
|
||||
import Simplex.Messaging.Agent.Protocol
|
||||
import Simplex.Messaging.Agent.RetryInterval
|
||||
import Simplex.Messaging.Agent.Store (RcvQueue)
|
||||
import Simplex.Messaging.Agent.Store.SQLite
|
||||
import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
|
||||
import Simplex.Messaging.Client
|
||||
@@ -147,12 +146,12 @@ newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do
|
||||
|
||||
data NtfSupervisor = NtfSupervisor
|
||||
{ ntfTkn :: TVar (Maybe NtfToken),
|
||||
ntfSubQ :: TBQueue (RcvQueue, NtfSupervisorCommand),
|
||||
ntfSubQ :: TBQueue (ConnId, NtfSupervisorCommand),
|
||||
ntfWorkers :: TMap NtfServer (TMVar (), Async ()),
|
||||
ntfSMPWorkers :: TMap SMPServer (TMVar (), Async ())
|
||||
}
|
||||
|
||||
data NtfSupervisorCommand = NSCCreate | NSCDelete
|
||||
data NtfSupervisorCommand = NSCCreate | NSCDelete | NSCNtfWorker NtfServer
|
||||
|
||||
newNtfSubSupervisor :: Natural -> STM NtfSupervisor
|
||||
newNtfSubSupervisor qSize = do
|
||||
|
||||
@@ -26,11 +26,13 @@ import qualified Data.Map.Strict as M
|
||||
import Data.Time (getCurrentTime)
|
||||
import Simplex.Messaging.Agent.Client
|
||||
import Simplex.Messaging.Agent.Env.SQLite
|
||||
import Simplex.Messaging.Agent.Protocol (ConnId)
|
||||
import Simplex.Messaging.Agent.Store
|
||||
import Simplex.Messaging.Client.Agent ()
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Notifications.Client
|
||||
import Simplex.Messaging.Notifications.Protocol (NtfTknStatus (..))
|
||||
import Simplex.Messaging.Protocol (ProtocolServer)
|
||||
import Simplex.Messaging.Notifications.Protocol (NtfTknStatus (..), SMPQueueNtf (..))
|
||||
import Simplex.Messaging.Protocol
|
||||
import Simplex.Messaging.TMap (TMap)
|
||||
import qualified Simplex.Messaging.TMap as TM
|
||||
import System.Random (randomR)
|
||||
@@ -46,33 +48,46 @@ runNtfSupervisor c = forever $ do
|
||||
Left e -> liftIO $ print e
|
||||
Right _ -> return ()
|
||||
|
||||
processNtfSub :: forall m. AgentMonad m => AgentClient -> (RcvQueue, NtfSupervisorCommand) -> m ()
|
||||
processNtfSub c (rcvQueue@RcvQueue {server = smpServer, rcvId}, cmd) = do
|
||||
processNtfSub :: forall m. AgentMonad m => AgentClient -> (ConnId, NtfSupervisorCommand) -> m ()
|
||||
processNtfSub c (connId, cmd) = do
|
||||
ntfServer_ <- getNtfServer c
|
||||
ns <- asks ntfSupervisor
|
||||
ntfToken_ <- readTVarIO $ ntfTkn ns
|
||||
case cmd of
|
||||
NSCCreate -> do
|
||||
sub_ <- withStore c $ \st -> getNtfSubscription st rcvQueue
|
||||
case (sub_, ntfServer_, ntfToken_) of
|
||||
(Nothing, Just ntfServer, Just tkn) -> do
|
||||
sub_ <- withStore c $ \st -> getNtfSubscription st connId
|
||||
RcvQueue {notifierId, server = smpServer} <- withStore c $ \st -> getRcvQueue st connId
|
||||
case (sub_, ntfServer_) of
|
||||
(Nothing, Just ntfServer) -> do
|
||||
currentTime <- liftIO getCurrentTime
|
||||
let newSub = newNtfSubscription ntfServer tkn smpServer rcvId currentTime
|
||||
withStore c $ \st -> createNtfSubscription st newSub
|
||||
case notifierId of
|
||||
(Just nId) -> do
|
||||
let newSub = newNtfSubscription connId smpServer (Just nId) ntfServer NASKey currentTime
|
||||
withStore c $ \st -> createNtfSubscription st newSub (NtfSubAction NSACreate)
|
||||
_ -> do
|
||||
let newSub = newNtfSubscription connId smpServer Nothing ntfServer NASNew currentTime
|
||||
withStore c $ \st -> createNtfSubscription st newSub (NtfSubSMPAction NSAKey)
|
||||
-- TODO optimize?
|
||||
-- TODO - read action in getNtfSubscription and decide which worker to create
|
||||
-- TODO - SMP worker can create Ntf worker on NKEY completion
|
||||
addNtfSMPWorker smpServer
|
||||
addNtfWorker ntfServer
|
||||
(Just _, Just ntfServer, Just _) -> do
|
||||
(Just _, Just ntfServer) -> do
|
||||
-- TODO subscription may have to be updated depending on current state:
|
||||
-- TODO - e.g., if it was previously marked for deletion action has to be updated
|
||||
-- TODO - should action depend on subscription status or always be NSAKey (NSACreate if notifierId exists)
|
||||
-- TODO in case worker is currently deleting it? When deleting worker should check for updated_by_supervisor
|
||||
-- TODO and if it is set perform update instead of delete. If worker was not deleting it yet it should
|
||||
-- TODO idempotently replay commands.
|
||||
addNtfSMPWorker smpServer
|
||||
addNtfWorker ntfServer
|
||||
_ -> pure ()
|
||||
NSCDelete -> do
|
||||
withStore c $ \st -> markNtfSubscriptionForDeletion st rcvQueue
|
||||
case (ntfServer_, ntfToken_) of
|
||||
(Just ntfServer, Just _) -> addNtfWorker ntfServer
|
||||
-- TODO delete notifier ID and Key from SMP server (SDEL, then NDEL)
|
||||
withStore c $ \st -> markNtfSubscriptionForDeletion st connId
|
||||
case ntfServer_ of
|
||||
(Just ntfServer) -> addNtfWorker ntfServer
|
||||
_ -> pure ()
|
||||
NSCNtfWorker ntfServer ->
|
||||
addNtfWorker ntfServer
|
||||
where
|
||||
addNtfWorker = addWorker ntfWorkers runNtfWorker
|
||||
addNtfSMPWorker = addWorker ntfSMPWorkers runNtfSMPWorker
|
||||
@@ -93,28 +108,68 @@ processNtfSub c (rcvQueue@RcvQueue {server = smpServer, rcvId}, cmd) = do
|
||||
runNtfWorker :: AgentMonad m => AgentClient -> NtfServer -> TMVar () -> m ()
|
||||
runNtfWorker c srv doWork = forever $ do
|
||||
void . atomically $ readTMVar doWork
|
||||
withStore c $ \st ->
|
||||
getNextNtfSubAction st srv >>= \case
|
||||
Nothing -> void . atomically $ tryTakeTMVar doWork
|
||||
Just (_sub, ntfSubAction) ->
|
||||
forM_ ntfSubAction $ \case
|
||||
NSANew _nKey -> pure ()
|
||||
NSACheck -> pure ()
|
||||
NSADelete -> pure ()
|
||||
getNtfToken_ >>= \case
|
||||
Just tkn@NtfToken {ntfTokenId = Just tknId, ntfTknStatus} ->
|
||||
withStore c (`getNextNtfSubAction` srv) >>= \case
|
||||
Just (ntfSub@NtfSubscription {connId, smpServer}, ntfSubAction, RcvQueue {ntfPrivateKey, notifierId}) -> do
|
||||
case ntfSubAction of
|
||||
NSACreate -> case (ntfPrivateKey, notifierId) of
|
||||
(Just ntfPrivKey, Just nId)
|
||||
| ntfTknStatus == NTActive -> do
|
||||
ntfSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer nId) ntfPrivKey
|
||||
ts <- liftIO getCurrentTime
|
||||
withStore c $ \st ->
|
||||
updateNtfSubscription st connId ntfSub {ntfSubId = Just ntfSubId, ntfSubStatus = NASCreated, ntfSubActionTs = ts} (NtfSubAction NSACheck)
|
||||
| otherwise -> pure () -- error
|
||||
_ -> pure () -- error
|
||||
NSACheck -> pure ()
|
||||
NSADelete -> pure ()
|
||||
Nothing -> noWorkToDo
|
||||
_ -> noWorkToDo
|
||||
delay <- asks $ ntfWorkerThrottle . config
|
||||
liftIO $ threadDelay delay
|
||||
where
|
||||
noWorkToDo = void . atomically $ tryTakeTMVar doWork
|
||||
|
||||
runNtfSMPWorker :: AgentMonad m => AgentClient -> NtfServer -> TMVar () -> m ()
|
||||
runNtfSMPWorker :: forall m. AgentMonad m => AgentClient -> SMPServer -> TMVar () -> m ()
|
||||
runNtfSMPWorker c srv doWork = forever $ do
|
||||
void . atomically $ readTMVar doWork
|
||||
withStore c $ \st ->
|
||||
getNextNtfSubSMPAction st srv >>= \case
|
||||
Nothing -> void . atomically $ tryTakeTMVar doWork
|
||||
Just (_sub, ntfSubAction) ->
|
||||
forM_ ntfSubAction $ \case
|
||||
NSAKey -> pure ()
|
||||
getNtfToken_ >>= \case
|
||||
Just NtfToken {ntfTknStatus} -> do
|
||||
withStore c (`getNextNtfSubSMPAction` srv) >>= \case
|
||||
Just (ntfSub@NtfSubscription {connId, ntfServer}, ntfSubAction, rq@RcvQueue {ntfPublicKey}) -> do
|
||||
case ntfSubAction of
|
||||
NSAKey
|
||||
| ntfTknStatus == NTActive ->
|
||||
case ntfPublicKey of
|
||||
Just ntfPubKey ->
|
||||
enableNotificationsWithNKey ntfPubKey
|
||||
_ -> do
|
||||
C.SignAlg a <- asks (cmdSignAlg . config)
|
||||
(ntfPubKey, ntfPrivKey) <- liftIO $ C.generateSignatureKeyPair a
|
||||
withStore c $ \st -> setRcvQueueNotifierKey st connId ntfPubKey ntfPrivKey
|
||||
enableNotificationsWithNKey ntfPubKey
|
||||
| otherwise -> pure () -- error
|
||||
where
|
||||
enableNotificationsWithNKey ntfPubKey = do
|
||||
nId <- enableQueueNotifications c rq ntfPubKey
|
||||
ts <- liftIO getCurrentTime
|
||||
withStore c $ \st -> do
|
||||
setRcvQueueNotifierId st connId nId
|
||||
updateNtfSubscription st connId ntfSub {ntfQueueId = Just nId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate)
|
||||
ns <- asks ntfSupervisor
|
||||
atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
|
||||
Nothing -> noWorkToDo
|
||||
_ -> noWorkToDo
|
||||
delay <- asks $ ntfWorkerThrottle . config
|
||||
liftIO $ threadDelay delay
|
||||
where
|
||||
noWorkToDo = void . atomically $ tryTakeTMVar doWork
|
||||
|
||||
getNtfToken_ :: AgentMonad m => m (Maybe NtfToken)
|
||||
getNtfToken_ = do
|
||||
tkn <- asks $ ntfTkn . ntfSupervisor
|
||||
readTVarIO tkn
|
||||
|
||||
nsUpdateToken :: NtfSupervisor -> NtfToken -> STM ()
|
||||
nsUpdateToken ns tkn = writeTVar (ntfTkn ns) $ Just tkn
|
||||
@@ -122,7 +177,7 @@ nsUpdateToken ns tkn = writeTVar (ntfTkn ns) $ Just tkn
|
||||
nsRemoveNtfToken :: NtfSupervisor -> STM ()
|
||||
nsRemoveNtfToken ns = writeTVar (ntfTkn ns) Nothing
|
||||
|
||||
sendNtfSubCommand :: NtfSupervisor -> (RcvQueue, NtfSupervisorCommand) -> STM ()
|
||||
sendNtfSubCommand :: NtfSupervisor -> (ConnId, NtfSupervisorCommand) -> STM ()
|
||||
sendNtfSubCommand ns cmd =
|
||||
readTVar (ntfTkn ns)
|
||||
>>= mapM_ (\NtfToken {ntfTknStatus} -> when (ntfTknStatus == NTActive) $ writeTBQueue (ntfSubQ ns) cmd)
|
||||
|
||||
@@ -26,6 +26,9 @@ import Simplex.Messaging.Protocol
|
||||
( MsgBody,
|
||||
MsgFlags,
|
||||
MsgId,
|
||||
NotifierId,
|
||||
NtfPrivateSignKey,
|
||||
NtfPublicVerifyKey,
|
||||
RcvDhSecret,
|
||||
RcvPrivateSignKey,
|
||||
SndPrivateSignKey,
|
||||
@@ -50,6 +53,10 @@ class Monad m => MonadAgentStore s m where
|
||||
setSndQueueStatus :: s -> SndQueue -> QueueStatus -> m ()
|
||||
getRcvQueue :: s -> ConnId -> m RcvQueue
|
||||
|
||||
-- RcvQueue notifier key and ID
|
||||
setRcvQueueNotifierKey :: s -> ConnId -> NtfPublicVerifyKey -> NtfPrivateSignKey -> m ()
|
||||
setRcvQueueNotifierId :: s -> ConnId -> NotifierId -> m ()
|
||||
|
||||
-- Confirmations
|
||||
createConfirmation :: s -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
|
||||
acceptConfirmation :: s -> ConfirmationId -> ConnInfo -> m AcceptedConfirmation
|
||||
@@ -84,20 +91,19 @@ class Monad m => MonadAgentStore s m where
|
||||
|
||||
-- Notification device token persistence
|
||||
createNtfToken :: s -> NtfToken -> m ()
|
||||
|
||||
getDeviceNtfToken :: s -> DeviceToken -> m (Maybe NtfToken, [NtfToken])
|
||||
updateNtfTokenRegistration :: s -> NtfToken -> NtfTokenId -> C.DhSecretX25519 -> m ()
|
||||
updateNtfToken :: s -> NtfToken -> NtfTknStatus -> Maybe NtfTknAction -> m ()
|
||||
removeNtfToken :: s -> NtfToken -> m ()
|
||||
|
||||
-- Notification subscription persistence
|
||||
getNtfSubscription :: s -> RcvQueue -> m (Maybe NtfSubscription)
|
||||
createNtfSubscription :: s -> NtfSubscription -> m ()
|
||||
markNtfSubscriptionForDeletion :: s -> RcvQueue -> m ()
|
||||
updateNtfSubscription :: s -> RcvQueue -> NtfSubscription -> m ()
|
||||
deleteNtfSubscription :: s -> RcvQueue -> m ()
|
||||
getNextNtfSubAction :: s -> NtfServer -> m (Maybe (NtfSubscription, Maybe NtfSubAction))
|
||||
getNextNtfSubSMPAction :: s -> SMPServer -> m (Maybe (NtfSubscription, Maybe NtfSubSMPAction))
|
||||
getNtfSubscription :: s -> ConnId -> m (Maybe NtfSubscription)
|
||||
createNtfSubscription :: s -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
markNtfSubscriptionForDeletion :: s -> ConnId -> m ()
|
||||
updateNtfSubscription :: s -> ConnId -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
deleteNtfSubscription :: s -> ConnId -> m ()
|
||||
getNextNtfSubAction :: s -> NtfServer -> m (Maybe (NtfSubscription, NtfSubAction, RcvQueue))
|
||||
getNextNtfSubSMPAction :: s -> SMPServer -> m (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue))
|
||||
|
||||
-- * Queue types
|
||||
|
||||
@@ -117,7 +123,12 @@ data RcvQueue = RcvQueue
|
||||
-- | sender queue ID
|
||||
sndId :: Maybe SMP.SenderId,
|
||||
-- | queue status
|
||||
status :: QueueStatus
|
||||
status :: QueueStatus,
|
||||
-- | key pair to be used by the notification server to sign transmissions
|
||||
ntfPublicKey :: Maybe NtfPublicVerifyKey,
|
||||
ntfPrivateKey :: Maybe NtfPrivateSignKey,
|
||||
-- | queue ID to be used by the notification server for NSUB command
|
||||
notifierId :: Maybe NotifierId
|
||||
}
|
||||
deriving (Eq, Show)
|
||||
|
||||
@@ -354,13 +365,3 @@ data StoreError
|
||||
| -- | Used to wrap agent errors inside store operations to avoid race conditions
|
||||
SEAgentError AgentErrorType
|
||||
deriving (Eq, Show, Exception)
|
||||
|
||||
-- * Ntf Subscriptions
|
||||
|
||||
-- data NewNtfSubscription = NewNtfSubscription
|
||||
-- { ntfServer :: NtfServer,
|
||||
-- ntfSubId :: NtfSubscriptionId,
|
||||
-- ratchetState :: RatchetX448
|
||||
-- }
|
||||
|
||||
data NtfSubscriptionAction = NSubSMPAction SMPServer | NSubNtfAction NtfServer
|
||||
|
||||
@@ -60,10 +60,10 @@ import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Crypto.Ratchet (RatchetX448, SkippedMsgDiff (..), SkippedMsgKeys)
|
||||
import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Encoding.String
|
||||
import Simplex.Messaging.Notifications.Client (NtfServer, NtfSubAction, NtfSubSMPAction, NtfSubscription, NtfTknAction, NtfToken (..))
|
||||
import Simplex.Messaging.Notifications.Client (NtfServer, NtfSubAction, NtfSubOrSMPAction (..), NtfSubSMPAction, NtfSubscription (..), NtfTknAction, NtfToken (..))
|
||||
import Simplex.Messaging.Notifications.Protocol (DeviceToken (..), NtfTknStatus (..), NtfTokenId)
|
||||
import Simplex.Messaging.Parsers (blobFieldParser, fromTextField_)
|
||||
import Simplex.Messaging.Protocol (MsgBody, MsgFlags, ProtocolServer (..))
|
||||
import Simplex.Messaging.Protocol (MsgBody, MsgFlags, NotifierId, NtfPrivateSignKey, NtfPublicVerifyKey, ProtocolServer (..))
|
||||
import qualified Simplex.Messaging.Protocol as SMP
|
||||
import Simplex.Messaging.Util (bshow, eitherToMaybe, liftIOEither)
|
||||
import Simplex.Messaging.Version
|
||||
@@ -302,7 +302,34 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
[":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId]
|
||||
|
||||
getRcvQueue :: SQLiteStore -> ConnId -> m RcvQueue
|
||||
getRcvQueue _st _connId = throwError SENotImplemented
|
||||
getRcvQueue st connId =
|
||||
liftIOEither . withTransaction st $ \db -> do
|
||||
rq_ <- getRcvQueueByConnId_ db connId
|
||||
pure $ maybe (Left SEConnNotFound) Right rq_
|
||||
|
||||
setRcvQueueNotifierKey :: SQLiteStore -> ConnId -> NtfPublicVerifyKey -> NtfPrivateSignKey -> m ()
|
||||
setRcvQueueNotifierKey st connId ntfPublicKey ntfPrivateKey =
|
||||
liftIO . withTransaction st $ \db ->
|
||||
DB.execute
|
||||
db
|
||||
[sql|
|
||||
UPDATE rcv_queues
|
||||
SET ntf_public_key = ?, ntf_private_key = ?
|
||||
WHERE conn_id = ?
|
||||
|]
|
||||
(ntfPublicKey, ntfPrivateKey, connId)
|
||||
|
||||
setRcvQueueNotifierId :: SQLiteStore -> ConnId -> NotifierId -> m ()
|
||||
setRcvQueueNotifierId st connId nId =
|
||||
liftIO . withTransaction st $ \db ->
|
||||
DB.execute
|
||||
db
|
||||
[sql|
|
||||
UPDATE rcv_queues
|
||||
SET ntf_id = ?
|
||||
WHERE conn_id = ?
|
||||
|]
|
||||
(nId, connId)
|
||||
|
||||
createConfirmation :: SQLiteStore -> TVar ChaChaDRG -> NewConfirmation -> m ConfirmationId
|
||||
createConfirmation st gVar NewConfirmation {connId, senderConf = SMPConfirmation {senderKey, e2ePubKey, connInfo, smpReplyQueues}, ratchetState} =
|
||||
@@ -663,26 +690,133 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
|
||||
|]
|
||||
(provider, token, host, port)
|
||||
|
||||
getNtfSubscription :: SQLiteStore -> RcvQueue -> m (Maybe NtfSubscription)
|
||||
getNtfSubscription _st _rcvQueue = throwError SENotImplemented
|
||||
getNtfSubscription :: SQLiteStore -> ConnId -> m (Maybe NtfSubscription)
|
||||
getNtfSubscription st connId =
|
||||
liftIO . withTransaction st $ \db ->
|
||||
maybeFirstRow ntfSubscription $
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT s.host, s.port, s.key_hash, ns.ntf_host, ns.ntf_port, ns.ntf_key_hash,
|
||||
nsb.smp_ntf_id, nsb.ntf_sub_id, nsb.ntf_sub_status, nsb.ntf_sub_action_ts
|
||||
FROM ntf_subscriptions nsb
|
||||
JOIN servers s ON s.host = nsb.smp_host, s.port = nsb.smp_port
|
||||
JOIN ntf_servers ns USING (ntf_host, ntf_port)
|
||||
WHERE nsb.conn_id = ?
|
||||
|]
|
||||
(Only connId)
|
||||
where
|
||||
ntfSubscription (smpHost, smpPort, smpKeyHash, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs) =
|
||||
let smpServer = SMPServer smpHost smpPort smpKeyHash
|
||||
ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash
|
||||
in NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}
|
||||
|
||||
createNtfSubscription :: SQLiteStore -> NtfSubscription -> m ()
|
||||
createNtfSubscription _st _ntfSub = throwError SENotImplemented
|
||||
createNtfSubscription :: SQLiteStore -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
createNtfSubscription st NtfSubscription {smpServer = (SMPServer host port _), ntfQueueId, ntfServer = (SMPServer ntfHost ntfPort _), ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction =
|
||||
liftIO . withTransaction st $ \db ->
|
||||
DB.execute
|
||||
db
|
||||
[sql|
|
||||
INSERT INTO ntf_subscriptions
|
||||
(smp_host, smp_port, smp_ntf_id, ntf_host, ntf_port, ntf_sub_id,
|
||||
ntf_sub_status, ntf_sub_action, ntf_sub_smp_action, ntf_sub_action_ts)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||||
|]
|
||||
( (host, port, ntfQueueId, ntfHost, ntfPort, ntfSubId)
|
||||
:. (ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs)
|
||||
)
|
||||
where
|
||||
(ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction
|
||||
|
||||
markNtfSubscriptionForDeletion :: s -> RcvQueue -> m ()
|
||||
markNtfSubscriptionForDeletion :: SQLiteStore -> ConnId -> m ()
|
||||
markNtfSubscriptionForDeletion _st _rcvQueue = throwError SENotImplemented
|
||||
|
||||
updateNtfSubscription :: s -> RcvQueue -> NtfSubscription -> m ()
|
||||
updateNtfSubscription _st _rcvQueue _ntfSub = throwError SENotImplemented
|
||||
updateNtfSubscription :: SQLiteStore -> ConnId -> NtfSubscription -> NtfSubOrSMPAction -> m ()
|
||||
updateNtfSubscription st connId NtfSubscription {ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs} ntfAction =
|
||||
liftIO . withTransaction st $ \db -> do
|
||||
r <- maybeFirstRow fromOnly $ DB.query db "SELECT updated_by_supervisor FROM ntf_subscriptions WHERE conn_id = ?" (Only connId)
|
||||
forM_ r $ \updatedBySupervisor -> do
|
||||
updatedAt <- getCurrentTime
|
||||
if updatedBySupervisor
|
||||
then
|
||||
DB.execute
|
||||
db
|
||||
[sql|
|
||||
UPDATE ntf_subscriptions
|
||||
SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, updated_by_supervisor = ?, updated_at = ?
|
||||
WHERE conn_id = ?
|
||||
|]
|
||||
(ntfQueueId, ntfSubId, ntfSubStatus, False, updatedAt, connId)
|
||||
else
|
||||
DB.execute
|
||||
db
|
||||
[sql|
|
||||
UPDATE ntf_subscriptions
|
||||
SET smp_ntf_id = ?, ntf_sub_id = ?, ntf_sub_status = ?, ntf_sub_action = ?, ntf_sub_smp_action = ?, ntf_sub_action_ts = ?, updated_by_supervisor = ?, updated_at = ?
|
||||
WHERE conn_id = ?
|
||||
|]
|
||||
(ntfQueueId, ntfSubId, ntfSubStatus, ntfSubAction, ntfSubSMPAction, ntfSubActionTs, False, updatedAt, connId)
|
||||
where
|
||||
(ntfSubAction, ntfSubSMPAction) = ntfSubAndSMPAction ntfAction
|
||||
|
||||
deleteNtfSubscription :: s -> RcvQueue -> m ()
|
||||
deleteNtfSubscription _st _rcvQueue = throwError SENotImplemented
|
||||
deleteNtfSubscription :: SQLiteStore -> ConnId -> m ()
|
||||
deleteNtfSubscription _st _connId = throwError SENotImplemented
|
||||
|
||||
getNextNtfSubAction :: s -> NtfServer -> m (Maybe (NtfSubscription, Maybe NtfSubAction))
|
||||
getNextNtfSubAction _st _srv = throwError SENotImplemented
|
||||
getNextNtfSubAction :: SQLiteStore -> NtfServer -> m (Maybe (NtfSubscription, NtfSubAction, RcvQueue))
|
||||
getNextNtfSubAction st ntfServer@(ProtocolServer ntfHost ntfPort _) =
|
||||
liftIO . withTransaction st $ \db -> do
|
||||
r <-
|
||||
maybeFirstRow ntfSubscription $
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT ns.connId, s.host, s.port, s.key_hash,
|
||||
ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_action
|
||||
FROM ntf_subscriptions ns
|
||||
JOIN servers s ON s.host = ns.smp_host, s.port = ns.smp_port
|
||||
WHERE ns.ntf_host = ? AND ns.ntf_port = ? AND ns.ntf_sub_action IS NOT NULL
|
||||
ORDER BY ns.ntf_sub_action_ts ASC
|
||||
LIMIT 1
|
||||
|]
|
||||
(ntfHost, ntfPort)
|
||||
case r of
|
||||
Just (ntfSub@NtfSubscription {connId}, ntfSubAction) -> do
|
||||
DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId)
|
||||
rq_ <- getRcvQueueByConnId_ db connId
|
||||
pure $ (\rq -> Just (ntfSub, ntfSubAction, rq)) =<< rq_
|
||||
Nothing -> pure Nothing
|
||||
where
|
||||
ntfSubscription (connId, smpHost, smpPort, smpKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) =
|
||||
let smpServer = SMPServer smpHost smpPort smpKeyHash
|
||||
in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction)
|
||||
|
||||
getNextNtfSubSMPAction :: s -> SMPServer -> m (Maybe (NtfSubscription, Maybe NtfSubSMPAction))
|
||||
getNextNtfSubSMPAction _st _srv = throwError SENotImplemented
|
||||
getNextNtfSubSMPAction :: SQLiteStore -> SMPServer -> m (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue))
|
||||
getNextNtfSubSMPAction st smpServer@(SMPServer smpHost smpPort _) =
|
||||
liftIO . withTransaction st $ \db -> do
|
||||
r <-
|
||||
maybeFirstRow ntfSubscription $
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT ns.connId, s.ntf_host, s.ntf_port, s.ntf_key_hash,
|
||||
ns.smp_ntf_id, ns.ntf_sub_id, ns.ntf_sub_status, ns.ntf_sub_action_ts, ns.ntf_sub_smp_action
|
||||
FROM ntf_subscriptions ns
|
||||
JOIN ntf_servers s USING (ntf_host, ntf_port)
|
||||
WHERE ns.smp_host = ? AND ns.smp_port = ? AND ns.ntf_sub_smp_action IS NOT NULL
|
||||
ORDER BY ns.ntf_sub_action_ts ASC
|
||||
LIMIT 1
|
||||
|]
|
||||
(smpHost, smpPort)
|
||||
case r of
|
||||
Just (ntfSub@NtfSubscription {connId}, ntfSubAction) -> do
|
||||
DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId)
|
||||
rq_ <- getRcvQueueByConnId_ db connId
|
||||
pure $ (\rq -> Just (ntfSub, ntfSubAction, rq)) =<< rq_
|
||||
Nothing -> pure Nothing
|
||||
where
|
||||
ntfSubscription (connId, ntfHost, ntfPort, ntfKeyHash, ntfQueueId, ntfSubId, ntfSubStatus, ntfSubActionTs, ntfSubAction) =
|
||||
let ntfServer = ProtocolServer ntfHost ntfPort ntfKeyHash
|
||||
in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction)
|
||||
|
||||
-- * Auxiliary helpers
|
||||
|
||||
@@ -745,6 +879,9 @@ listToEither e _ = Left e
|
||||
firstRow :: (a -> b) -> e -> IO [a] -> IO (Either e b)
|
||||
firstRow f e a = second f . listToEither e <$> a
|
||||
|
||||
maybeFirstRow :: Functor f => (a -> b) -> f [a] -> f (Maybe b)
|
||||
maybeFirstRow f q = fmap f . listToMaybe <$> q
|
||||
|
||||
-- TODO move from simplex-chat
|
||||
-- firstRow' :: (a -> Either e b) -> e -> IO [a] -> IO (Either e b)
|
||||
-- firstRow' f e a = (f <=< listToEither e) <$> a
|
||||
@@ -855,22 +992,22 @@ getConnData_ dbConn connId' =
|
||||
|
||||
getRcvQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe RcvQueue)
|
||||
getRcvQueueByConnId_ dbConn connId =
|
||||
rcvQueue
|
||||
listToMaybe . map rcvQueue
|
||||
<$> DB.query
|
||||
dbConn
|
||||
[sql|
|
||||
SELECT s.key_hash, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
|
||||
q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status
|
||||
q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status,
|
||||
q.ntf_public_key, q.ntf_private_key, q.ntf_id
|
||||
FROM rcv_queues q
|
||||
INNER JOIN servers s ON q.host = s.host AND q.port = s.port
|
||||
WHERE q.conn_id = ?;
|
||||
|]
|
||||
(Only connId)
|
||||
where
|
||||
rcvQueue [(keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)] =
|
||||
rcvQueue ((keyHash, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (ntfPublicKey, ntfPrivateKey, notifierId)) =
|
||||
let server = SMPServer host port keyHash
|
||||
in Just RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status}
|
||||
rcvQueue _ = Nothing
|
||||
in RcvQueue {server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, ntfPublicKey, ntfPrivateKey, notifierId}
|
||||
|
||||
getSndQueueByConnId_ :: DB.Connection -> ConnId -> IO (Maybe SndQueue)
|
||||
getSndQueueByConnId_ dbConn connId =
|
||||
@@ -1086,3 +1223,7 @@ createWithRandomId gVar create = tryCreate 3
|
||||
|
||||
randomId :: TVar ChaChaDRG -> Int -> IO ByteString
|
||||
randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)
|
||||
|
||||
ntfSubAndSMPAction :: NtfSubOrSMPAction -> (Maybe NtfSubAction, Maybe NtfSubSMPAction)
|
||||
ntfSubAndSMPAction (NtfSubAction nsa) = (Just nsa, Nothing)
|
||||
ntfSubAndSMPAction (NtfSubSMPAction nsa) = (Nothing, Just nsa)
|
||||
|
||||
-41
@@ -1,41 +0,0 @@
|
||||
{-# LANGUAGE QuasiQuotes #-}
|
||||
|
||||
module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220404_ntf_subscriptions_draft where
|
||||
|
||||
import Database.SQLite.Simple (Query)
|
||||
import Database.SQLite.Simple.QQ (sql)
|
||||
|
||||
m20220404_ntf_subscriptions_draft :: Query
|
||||
m20220404_ntf_subscriptions_draft =
|
||||
[sql|
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_id BLOB;
|
||||
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_public_key BLOB;
|
||||
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_private_key BLOB;
|
||||
|
||||
CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues (host, port, ntf_id);
|
||||
|
||||
CREATE TABLE ntf_subscriptions (
|
||||
ntf_host TEXT NOT NULL,
|
||||
ntf_port TEXT NOT NULL,
|
||||
ntf_sub_id BLOB,
|
||||
ntf_sub_status TEXT NOT NULL, -- new, created, active, pending, error_auth
|
||||
ntf_sub_action TEXT, -- if there is an action required on this subscription: create / check / delete
|
||||
ntf_sub_smp_action TEXT, -- action with SMP server: nkey; only one of this and ntf_sub_action can (should) be not null in same record
|
||||
ntf_sub_action_ts TEXT, -- the earliest time for the action, e.g. checks can be scheduled every X hours
|
||||
ntf_token TEXT NOT NULL, -- or BLOB?
|
||||
smp_host TEXT NULL,
|
||||
smp_port TEXT NULL,
|
||||
smp_rcv_id BLOB NULL,
|
||||
smp_ntf_id BLOB,
|
||||
marked_for_deletion INTEGER NOT NULL DEFAULT 0, -- to be checked on updates by workers to not overwrite delete command
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL, -- this is to check subscription status periodically to know when it was last checked
|
||||
PRIMARY KEY (smp_host, smp_port, smp_rcv_id),
|
||||
FOREIGN KEY (ntf_host, ntf_port) REFERENCES ntf_servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE,
|
||||
FOREIGN KEY (smp_host, smp_port, smp_rcv_id) REFERENCES rcv_queues (host, port, rcv_id)
|
||||
ON DELETE SET NULL ON UPDATE CASCADE
|
||||
) WITHOUT ROWID;
|
||||
|]
|
||||
@@ -15,4 +15,36 @@ ALTER TABLE conn_confirmations ADD COLUMN smp_reply_queues BLOB NULL;
|
||||
ALTER TABLE connections ADD COLUMN duplex_handshake INTEGER NULL DEFAULT 0;
|
||||
|
||||
ALTER TABLE rcv_messages ADD COLUMN user_ack INTEGER NULL DEFAULT 0;
|
||||
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_public_key BLOB;
|
||||
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_private_key BLOB;
|
||||
|
||||
ALTER TABLE rcv_queues ADD COLUMN ntf_id BLOB;
|
||||
|
||||
CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues (host, port, ntf_id);
|
||||
|
||||
CREATE TABLE ntf_subscriptions (
|
||||
conn_id BLOB NOT NULL,
|
||||
smp_host TEXT NULL,
|
||||
smp_port TEXT NULL,
|
||||
smp_ntf_id BLOB,
|
||||
ntf_host TEXT NOT NULL,
|
||||
ntf_port TEXT NOT NULL,
|
||||
ntf_sub_id BLOB,
|
||||
ntf_sub_status TEXT NOT NULL, -- see NtfAgentSubStatus
|
||||
ntf_sub_action TEXT, -- if there is an action required on this subscription: NtfSubAction
|
||||
ntf_sub_smp_action TEXT, -- action with SMP server: NtfSubSMPAction; only one of this and ntf_sub_action can (should) be not null in same record
|
||||
ntf_sub_action_ts TEXT, -- the earliest time for the action, e.g. checks can be scheduled every X hours
|
||||
updated_by_supervisor INTEGER NOT NULL DEFAULT 0, -- to be checked on updates by workers to not overwrite supervisor command (state still should be updated)
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
updated_at TEXT NOT NULL DEFAULT (datetime('now')),
|
||||
PRIMARY KEY (conn_id),
|
||||
FOREIGN KEY (conn_id) REFERENCES connections
|
||||
ON DELETE SET NULL ON UPDATE CASCADE,
|
||||
FOREIGN KEY (smp_host, smp_port) REFERENCES servers (host, port)
|
||||
ON DELETE SET NULL ON UPDATE CASCADE,
|
||||
FOREIGN KEY (ntf_host, ntf_port) REFERENCES ntf_servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
) WITHOUT ROWID;
|
||||
|]
|
||||
|
||||
@@ -36,6 +36,9 @@ CREATE TABLE rcv_queues(
|
||||
status TEXT NOT NULL,
|
||||
smp_server_version INTEGER NOT NULL DEFAULT 1,
|
||||
smp_client_version INTEGER,
|
||||
ntf_public_key BLOB,
|
||||
ntf_private_key BLOB,
|
||||
ntf_id BLOB,
|
||||
PRIMARY KEY(host, port, rcv_id),
|
||||
FOREIGN KEY(host, port) REFERENCES servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE,
|
||||
@@ -165,3 +168,27 @@ tkn_dh_secret BLOB, -- DH secret for e2e encryption of notifications
|
||||
FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
) WITHOUT ROWID;
|
||||
CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
|
||||
CREATE TABLE ntf_subscriptions(
|
||||
conn_id BLOB NOT NULL,
|
||||
smp_host TEXT NULL,
|
||||
smp_port TEXT NULL,
|
||||
smp_ntf_id BLOB,
|
||||
ntf_host TEXT NOT NULL,
|
||||
ntf_port TEXT NOT NULL,
|
||||
ntf_sub_id BLOB,
|
||||
ntf_sub_status TEXT NOT NULL, -- see NtfAgentSubStatus
|
||||
ntf_sub_action TEXT, -- if there is an action required on this subscription: NtfSubAction
|
||||
ntf_sub_smp_action TEXT, -- action with SMP server: NtfSubSMPAction; only one of this and ntf_sub_action can(should) be not null in same record
|
||||
ntf_sub_action_ts TEXT, -- the earliest time for the action, e.g. checks can be scheduled every X hours
|
||||
updated_by_supervisor INTEGER NOT NULL DEFAULT 0, -- to be checked on updates by workers to not overwrite supervisor command(state still should be updated)
|
||||
created_at TEXT NOT NULL DEFAULT(datetime('now')),
|
||||
updated_at TEXT NOT NULL DEFAULT(datetime('now')),
|
||||
PRIMARY KEY(conn_id),
|
||||
FOREIGN KEY(conn_id) REFERENCES connections
|
||||
ON DELETE SET NULL ON UPDATE CASCADE,
|
||||
FOREIGN KEY(smp_host, smp_port) REFERENCES servers(host, port)
|
||||
ON DELETE SET NULL ON UPDATE CASCADE,
|
||||
FOREIGN KEY(ntf_host, ntf_port) REFERENCES ntf_servers
|
||||
ON DELETE RESTRICT ON UPDATE CASCADE
|
||||
) WITHOUT ROWID;
|
||||
|
||||
@@ -10,6 +10,7 @@ module Simplex.Messaging.Notifications.Client where
|
||||
import Control.Monad.Except
|
||||
import Control.Monad.Trans.Except
|
||||
import qualified Data.Attoparsec.ByteString.Char8 as A
|
||||
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
|
||||
import Data.Time (UTCTime)
|
||||
import Data.Word (Word16)
|
||||
import Database.SQLite.Simple.FromField (FromField (..))
|
||||
@@ -18,8 +19,9 @@ import Simplex.Messaging.Client
|
||||
import qualified Simplex.Messaging.Crypto as C
|
||||
import Simplex.Messaging.Encoding
|
||||
import Simplex.Messaging.Notifications.Protocol
|
||||
import Simplex.Messaging.Parsers (blobFieldDecoder)
|
||||
import Simplex.Messaging.Protocol (NotifierId, NtfPrivateSignKey, ProtocolServer, RecipientId, SMPServer)
|
||||
import Simplex.Messaging.Parsers (blobFieldDecoder, fromTextField_)
|
||||
import Simplex.Messaging.Protocol (NotifierId, ProtocolServer, SMPServer)
|
||||
import Simplex.Messaging.Agent.Protocol (ConnId)
|
||||
|
||||
type NtfServer = ProtocolServer
|
||||
|
||||
@@ -132,20 +134,22 @@ newNtfToken deviceToken ntfServer (ntfPubKey, ntfPrivKey) ntfDhKeys =
|
||||
ntfTknAction = Just NTARegister
|
||||
}
|
||||
|
||||
data NtfSubOrSMPAction = NtfSubAction NtfSubAction | NtfSubSMPAction NtfSubSMPAction
|
||||
|
||||
data NtfSubAction
|
||||
= NSANew NtfPrivateSignKey
|
||||
= NSACreate
|
||||
| NSACheck
|
||||
| NSADelete
|
||||
deriving (Show)
|
||||
|
||||
instance Encoding NtfSubAction where
|
||||
smpEncode = \case
|
||||
NSANew nKey -> smpEncode ('N', nKey)
|
||||
NSACreate -> "N"
|
||||
NSACheck -> "C"
|
||||
NSADelete -> "D"
|
||||
smpP =
|
||||
A.anyChar >>= \case
|
||||
'N' -> NSANew <$> smpP
|
||||
'N' -> pure NSACreate
|
||||
'C' -> pure NSACheck
|
||||
'D' -> pure NSADelete
|
||||
_ -> fail "bad NtfSubAction"
|
||||
@@ -170,27 +174,66 @@ instance FromField NtfSubSMPAction where fromField = blobFieldDecoder smpDecode
|
||||
|
||||
instance ToField NtfSubSMPAction where toField = toField . smpEncode
|
||||
|
||||
data NtfAgentSubStatus
|
||||
= -- | subscription started
|
||||
NASNew
|
||||
| -- | state after NKEY - notifier ID is assigned to queue on SMP server
|
||||
NASKey
|
||||
| -- | state after SNEW - subscription created on notification server
|
||||
NASCreated
|
||||
| -- | connected and subscribed to SMP server
|
||||
NASActive
|
||||
| -- | communicated by notification server that NEND received (we currently do not support it)
|
||||
NASEnded
|
||||
| -- | communicated by notification server that SMP AUTH error occured
|
||||
NASSMPAuth
|
||||
| -- | state after SDEL (subscription is deleted on notification server)
|
||||
NASDeleted
|
||||
deriving (Eq, Show)
|
||||
|
||||
instance Encoding NtfAgentSubStatus where
|
||||
smpEncode = \case
|
||||
NASNew -> "NEW"
|
||||
NASKey -> "NKEY"
|
||||
NASCreated -> "CREATED"
|
||||
NASActive -> "ACTIVE"
|
||||
NASEnded -> "ENDED"
|
||||
NASSMPAuth -> "SMP_AUTH"
|
||||
NASDeleted -> "DELETED"
|
||||
smpP =
|
||||
A.takeTill (== ' ') >>= \case
|
||||
"NEW" -> pure NASNew
|
||||
"NKEY" -> pure NASKey
|
||||
"CREATED" -> pure NASCreated
|
||||
"ACTIVE" -> pure NASActive
|
||||
"ENDED" -> pure NASEnded
|
||||
"SMP_AUTH" -> pure NASSMPAuth
|
||||
"DELETED" -> pure NASDeleted
|
||||
_ -> fail "bad NtfAgentSubStatus"
|
||||
|
||||
instance FromField NtfAgentSubStatus where fromField = fromTextField_ $ either (const Nothing) Just . smpDecode . encodeUtf8
|
||||
|
||||
instance ToField NtfAgentSubStatus where toField = toField . decodeLatin1 . smpEncode
|
||||
|
||||
data NtfSubscription = NtfSubscription
|
||||
{ ntfServer :: NtfServer,
|
||||
{ connId :: ConnId,
|
||||
smpServer :: SMPServer,
|
||||
ntfQueueId :: Maybe NotifierId,
|
||||
ntfServer :: NtfServer,
|
||||
ntfSubId :: Maybe NtfSubscriptionId,
|
||||
ntfSubStatus :: NtfSubStatus,
|
||||
ntfSubActionTs :: UTCTime,
|
||||
ntfToken :: NtfToken, -- ?
|
||||
smpServer :: SMPServer, -- use SMPQueueNtf?
|
||||
rcvQueueId :: RecipientId,
|
||||
ntfQueueId :: Maybe NotifierId
|
||||
ntfSubStatus :: NtfAgentSubStatus,
|
||||
ntfSubActionTs :: UTCTime
|
||||
}
|
||||
deriving (Show)
|
||||
|
||||
newNtfSubscription :: NtfServer -> NtfToken -> SMPServer -> RecipientId -> UTCTime -> NtfSubscription
|
||||
newNtfSubscription ntfServer ntfToken smpServer rcvQueueId ntfSubActionTs =
|
||||
newNtfSubscription :: ConnId -> SMPServer -> Maybe NotifierId -> NtfServer -> NtfAgentSubStatus -> UTCTime -> NtfSubscription
|
||||
newNtfSubscription connId smpServer ntfQueueId ntfServer ntfSubStatus ntfSubActionTs =
|
||||
NtfSubscription
|
||||
{ ntfServer,
|
||||
ntfSubId = Nothing,
|
||||
ntfSubStatus = NSKey,
|
||||
ntfSubActionTs,
|
||||
ntfToken,
|
||||
{ connId,
|
||||
smpServer,
|
||||
rcvQueueId,
|
||||
ntfQueueId = Nothing
|
||||
ntfQueueId,
|
||||
ntfServer,
|
||||
ntfSubId = Nothing,
|
||||
ntfSubStatus,
|
||||
ntfSubActionTs
|
||||
}
|
||||
|
||||
@@ -363,9 +363,7 @@ type NtfSubscriptionId = NtfEntityId
|
||||
type NtfTokenId = NtfEntityId
|
||||
|
||||
data NtfSubStatus
|
||||
= -- | state after NKEY
|
||||
NSKey
|
||||
| -- | state after SNEW
|
||||
= -- | state after SNEW
|
||||
NSNew
|
||||
| -- | pending connection/subscription to SMP server
|
||||
NSPending
|
||||
@@ -379,7 +377,6 @@ data NtfSubStatus
|
||||
|
||||
instance Encoding NtfSubStatus where
|
||||
smpEncode = \case
|
||||
NSKey -> "KEY"
|
||||
NSNew -> "NEW"
|
||||
NSPending -> "PENDING" -- e.g. after SMP server disconnect/timeout while ntf server is retrying to connect
|
||||
NSActive -> "ACTIVE"
|
||||
@@ -387,7 +384,6 @@ instance Encoding NtfSubStatus where
|
||||
NSSMPAuth -> "SMP_AUTH"
|
||||
smpP =
|
||||
A.takeTill (== ' ') >>= \case
|
||||
"KEY" -> pure NSKey
|
||||
"NEW" -> pure NSNew
|
||||
"PENDING" -> pure NSPending
|
||||
"ACTIVE" -> pure NSActive
|
||||
|
||||
@@ -506,10 +506,10 @@ type SndPrivateSignKey = C.APrivateSignKey
|
||||
-- | Sender's public key used by SMP server to verify authorization of SMP commands.
|
||||
type SndPublicVerifyKey = C.APublicVerifyKey
|
||||
|
||||
-- | Private key used by push notifications server to authorize (sign) LSTN command.
|
||||
-- | Private key used by push notifications server to authorize (sign) NSUB command.
|
||||
type NtfPrivateSignKey = C.APrivateSignKey
|
||||
|
||||
-- | Public key used by SMP server to verify authorization of LSTN command sent by push notifications server.
|
||||
-- | Public key used by SMP server to verify authorization of NSUB command sent by push notifications server.
|
||||
type NtfPublicVerifyKey = C.APublicVerifyKey
|
||||
|
||||
-- | SMP message server ID.
|
||||
|
||||
@@ -166,7 +166,10 @@ rcvQueue1 =
|
||||
e2ePrivKey = testPrivDhKey,
|
||||
e2eDhSecret = Nothing,
|
||||
sndId = Just "2345",
|
||||
status = New
|
||||
status = New,
|
||||
ntfPublicKey = Nothing,
|
||||
ntfPrivateKey = Nothing,
|
||||
notifierId = Nothing
|
||||
}
|
||||
|
||||
sndQueue1 :: SndQueue
|
||||
@@ -332,7 +335,10 @@ testUpgradeSndConnToDuplex =
|
||||
e2ePrivKey = testPrivDhKey,
|
||||
e2eDhSecret = Nothing,
|
||||
sndId = Just "4567",
|
||||
status = New
|
||||
status = New,
|
||||
ntfPublicKey = Nothing,
|
||||
ntfPrivateKey = Nothing,
|
||||
notifierId = Nothing
|
||||
}
|
||||
upgradeSndConnToDuplex store "conn1" anotherRcvQueue
|
||||
`throwsError` SEBadConnType CRcv
|
||||
|
||||
Reference in New Issue
Block a user