diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 449f710bc..f76501074 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -622,11 +622,10 @@ deleteConnection' c connId = delete :: RcvQueue -> m () delete rq = do deleteQueue c rq - ns <- asks ntfSupervisor - atomically $ do - removeSubscription c connId - sendNtfSubCommand ns (connId, NSCDelete) + atomically $ removeSubscription c connId withStore' c (`deleteConn` connId) + ns <- asks ntfSupervisor + atomically $ sendNtfSubCommand ns (connId, NSCDelete) -- | Change servers to be used for creating new queues, in Reader monad setSMPServers' :: AgentMonad m => AgentClient -> NonEmpty SMPServer -> m () diff --git a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs index d2628325b..cc6ea7ce7 100644 --- a/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs +++ b/src/Simplex/Messaging/Agent/NtfSubSupervisor.hs @@ -124,18 +124,18 @@ runNtfWorker c srv doWork = forever $ do ts <- liftIO getCurrentTime case nextSub_ of Nothing -> noWorkToDo - Just (ntfSub@NtfSubscription {connId, smpServer, ntfSubId}, ntfSubAction, RcvQueue {clientNtfCreds}) -> + Just (ntfSub@NtfSubscription {connId, smpServer, ntfSubId}, ntfSubAction) -> unlessM (rescheduleAction doWork ts ntfSub) $ case ntfSubAction of - NSACreate -> case clientNtfCreds of - Just ClientNtfCreds {ntfPrivateKey, notifierId} - | ntfTknStatus == NTActive -> do + NSACreate -> do + rq_ <- withStore' c (`getRcvQueue` connId) + case (rq_, ntfTknStatus) of + (Right RcvQueue {clientNtfCreds = Just ClientNtfCreds {ntfPrivateKey, notifierId}}, NTActive) -> do nSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer notifierId) ntfPrivateKey let actionTs = addUTCTime 30 ts withStore' c $ \db -> updateNtfSubscription db connId ntfSub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew, ntfSubActionTs = actionTs} (NtfSubAction NSACheck) - | otherwise -> ntfInternalError c connId "NSACreate - token not active" - _ -> ntfInternalError c connId "NSACreate - no notifier key or ID" + _ -> ntfInternalError c connId "NSACreate - no notifier queue credentials or token not active" NSACheck -> case ntfSubId of Just nSubId -> agentNtfCheckSubscription c nSubId tkn >>= \case @@ -177,24 +177,27 @@ runNtfSMPWorker c srv doWork = forever $ do ts <- liftIO getCurrentTime case nextSub_ of Nothing -> noWorkToDo - Just (ntfSub@NtfSubscription {connId, ntfServer}, ntfSubAction, rq) -> + Just (ntfSub@NtfSubscription {connId, ntfServer}, ntfSubAction) -> unlessM (rescheduleAction doWork ts ntfSub) $ case ntfSubAction of - NSASmpKey - | ntfTknStatus == NTActive -> do - C.SignAlg a <- asks (cmdSignAlg . config) - (ntfPublicKey, ntfPrivateKey) <- liftIO $ C.generateSignatureKeyPair a - (rcvNtfPubDhKey, rcvNtfPrivDhKey) <- liftIO C.generateKeyPair' - (notifierId, rcvNtfSrvPubDhKey) <- enableQueueNotifications c rq ntfPublicKey rcvNtfPubDhKey - let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey - withStore' c $ \st -> do - setRcvQueueNtfCreds st connId ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} - updateNtfSubscription st connId ntfSub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate) - ns <- asks ntfSupervisor - atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer) - | otherwise -> ntfInternalError c connId "NSASmpKey - token not active" + NSASmpKey -> do + rq_ <- withStore' c (`getRcvQueue` connId) + case (rq_, ntfTknStatus) of + (Right rq, NTActive) -> do + C.SignAlg a <- asks (cmdSignAlg . config) + (ntfPublicKey, ntfPrivateKey) <- liftIO $ C.generateSignatureKeyPair a + (rcvNtfPubDhKey, rcvNtfPrivDhKey) <- liftIO C.generateKeyPair' + (notifierId, rcvNtfSrvPubDhKey) <- enableQueueNotifications c rq ntfPublicKey rcvNtfPubDhKey + let rcvNtfDhSecret = C.dh' rcvNtfSrvPubDhKey rcvNtfPrivDhKey + withStore' c $ \st -> do + setRcvQueueNtfCreds st connId ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} + updateNtfSubscription st connId ntfSub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey, ntfSubActionTs = ts} (NtfSubAction NSACreate) + ns <- asks ntfSupervisor + atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer) + _ -> ntfInternalError c connId "NSASmpKey - no rcv queue or token not active" NSASmpDelete -> do - disableQueueNotifications c rq + rq_ <- withStore' c (`getRcvQueue` connId) + forM_ rq_ $ \rq -> disableQueueNotifications c rq withStore' c $ \db -> deleteNtfSubscription db connId _ -> noWorkToDo delay <- asks $ ntfWorkerThrottle . config diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 3f4b96529..cabb883c3 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -793,11 +793,11 @@ deleteNtfSubscription db connId = do (Nothing :: Maybe SMP.NotifierId, Nothing :: Maybe NtfSubscriptionId, NASDeleted, False, updatedAt, connId) else DB.execute db "DELETE FROM ntf_subscriptions WHERE conn_id = ?" (Only connId) -getNextNtfSubAction :: DB.Connection -> NtfServer -> IO (Maybe (NtfSubscription, NtfSubAction, RcvQueue)) +getNextNtfSubAction :: DB.Connection -> NtfServer -> IO (Maybe (NtfSubscription, NtfSubAction)) getNextNtfSubAction db ntfServer@(ProtocolServer ntfHost ntfPort _) = do - maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do + maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \ntfSub@(NtfSubscription {connId}, _) -> do DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) - (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId + pure $ Just ntfSub where getNtfSubAction_ = DB.query @@ -816,11 +816,11 @@ getNextNtfSubAction db ntfServer@(ProtocolServer ntfHost ntfPort _) = do let smpServer = SMPServer smpHost smpPort smpKeyHash in (NtfSubscription {connId, smpServer, ntfQueueId, ntfServer, ntfSubId, ntfSubStatus, ntfSubActionTs}, ntfSubAction) -getNextNtfSubSMPAction :: DB.Connection -> SMPServer -> IO (Maybe (NtfSubscription, NtfSubSMPAction, RcvQueue)) +getNextNtfSubSMPAction :: DB.Connection -> SMPServer -> IO (Maybe (NtfSubscription, NtfSubSMPAction)) getNextNtfSubSMPAction db smpServer@(SMPServer smpHost smpPort _) = do - maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \(ntfSub@NtfSubscription {connId}, ntfSubAction) -> do + maybeFirstRow ntfSubscription getNtfSubAction_ $>>= \ntfSub@(NtfSubscription {connId}, _) -> do DB.execute db "UPDATE ntf_subscriptions SET updated_by_supervisor = ? WHERE conn_id = ?" (False, connId) - (ntfSub,ntfSubAction,) <$$> getRcvQueueByConnId_ db connId + pure $ Just ntfSub where getNtfSubAction_ = DB.query diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220608_v2.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220608_v2.hs index 753b3c164..544ccf60f 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220608_v2.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220608_v2.hs @@ -27,7 +27,7 @@ ALTER TABLE rcv_queues ADD COLUMN rcv_ntf_dh_secret BLOB; CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues (host, port, ntf_id); CREATE TABLE ntf_subscriptions ( - conn_id BLOB NOT NULL, -- ? make nullable for when connection is deleted but we still need to delete subscription + conn_id BLOB NOT NULL, smp_host TEXT NULL, smp_port TEXT NULL, smp_ntf_id BLOB, @@ -42,8 +42,6 @@ CREATE TABLE ntf_subscriptions ( 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 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 2a64a78f4..5a4436e8b 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -171,7 +171,7 @@ tkn_dh_secret BLOB, -- DH secret for e2e encryption of notifications ) 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, -- ? make nullable for when connection is deleted but we still need to delete subscription + conn_id BLOB NOT NULL, smp_host TEXT NULL, smp_port TEXT NULL, smp_ntf_id BLOB, @@ -186,8 +186,6 @@ CREATE TABLE ntf_subscriptions( 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