mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-07-16 21:01:51 +00:00
delivery to non-current queues
This commit is contained in:
Evgeny Poberezkin
@@ -92,6 +92,7 @@ import qualified Data.List.NonEmpty as L
|
||||
import Data.Map.Strict (Map)
|
||||
import qualified Data.Map.Strict as M
|
||||
import Data.Maybe (isJust)
|
||||
import qualified Data.Set as S
|
||||
import qualified Data.Text as T
|
||||
import Data.Time.Clock
|
||||
import Data.Time.Clock.System (systemToUTCTime)
|
||||
@@ -513,8 +514,9 @@ subscribeConnection' c connId =
|
||||
withStore c (`getConn` connId) >>= \conn -> do
|
||||
resumeConnCmds c connId
|
||||
case conn of
|
||||
SomeConn _ (DuplexConnection cData rq sq rq' _) -> do
|
||||
SomeConn _ (DuplexConnection cData rq sq rq' sq') -> do
|
||||
resumeMsgDelivery c cData sq
|
||||
mapM_ (resumeMsgDelivery c cData) sq'
|
||||
void . forkIO $ doRcvQueueAction c cData rq sq
|
||||
mapM_ subscribe rq' `catchError` \_ -> pure ()
|
||||
subscribe rq
|
||||
@@ -562,6 +564,7 @@ createNextRcvQueue c cData@ConnData {connId} rq@RcvQueue {server, sndId} sq = do
|
||||
pure SMPQueueUri {clientVRange, queueAddress}
|
||||
_ -> do
|
||||
srv <- getNextSMPServer c [server]
|
||||
liftIO $ putStrLn $ "createNextRcvQueue " <> show server <> " -> " <> show srv
|
||||
(rq', qUri) <- newRcvQueue c srv clientVRange False
|
||||
withStore' c $ \db -> dbCreateNextRcvQueue db connId rq rq'
|
||||
pure qUri
|
||||
@@ -572,10 +575,10 @@ secureNextRcvQueue :: AgentMonad m => AgentClient -> ConnData -> RcvQueue -> Snd
|
||||
secureNextRcvQueue c cData rq sq rq'@RcvQueue {server, sndId, status, sndPublicKey} = do
|
||||
when (status == Confirmed) $ case sndPublicKey of
|
||||
Just sKey -> do
|
||||
secureQueue c rq sKey
|
||||
secureQueue c rq' sKey
|
||||
withStore' c $ \db -> setRcvQueueStatus db rq' Secured
|
||||
_ -> do
|
||||
-- notify user: no sender key
|
||||
-- TODO notify user: no sender key
|
||||
pure ()
|
||||
void . enqueueMessage c cData sq SMP.noMsgFlags $ QREADY (server, sndId)
|
||||
withStore' c $ \db -> setRcvQueueAction db rq Nothing
|
||||
@@ -615,7 +618,7 @@ subscribeConnections' c connIds = do
|
||||
-- prepare map for batch subscriptions
|
||||
srvRcvQs :: Map SMPServer (Map (ConnId, Bool) RcvQueue) = M.foldlWithKey' addRcvQueues M.empty rcvQs
|
||||
-- TODO start message delivery for non-current queues
|
||||
mapM_ (mapM_ (uncurry $ resumeMsgDelivery c) . sndQueue) cs
|
||||
mapM_ (mapM_ resumeDelivery . sndQueue) cs
|
||||
mapM_ (resumeConnCmds c) $ M.keys cs
|
||||
-- send batch subscriptions concurrently to all servers
|
||||
rcvRs <- mapConcurrently subscribe (M.assocs srvRcvQs)
|
||||
@@ -661,10 +664,14 @@ subscribeConnections' c connIds = do
|
||||
forM_ (concatMap M.assocs rcvRs) $ \case
|
||||
(connId, Right _) -> atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCCreate)
|
||||
_ -> pure ()
|
||||
sndQueue :: SomeConn -> Maybe (ConnData, SndQueue)
|
||||
resumeDelivery :: (ConnData, SndQueue, Maybe SndQueue) -> m ()
|
||||
resumeDelivery (cData, sq, sq') = do
|
||||
resumeMsgDelivery c cData sq
|
||||
mapM_ (resumeMsgDelivery c cData) sq'
|
||||
sndQueue :: SomeConn -> Maybe (ConnData, SndQueue, Maybe SndQueue)
|
||||
sndQueue = \case
|
||||
SomeConn _ (DuplexConnection cData _ sq _ _) -> Just (cData, sq)
|
||||
SomeConn _ (SndConnection cData sq) -> Just (cData, sq)
|
||||
SomeConn _ (DuplexConnection cData _ sq _ sq') -> Just (cData, sq, sq')
|
||||
SomeConn _ (SndConnection cData sq) -> Just (cData, sq, Nothing)
|
||||
_ -> Nothing
|
||||
notifyResultError :: Map ConnId (Either AgentErrorType ()) -> m ()
|
||||
notifyResultError rs = do
|
||||
@@ -829,7 +836,7 @@ runCommandProcessing c@AgentClient {subQ} server = do
|
||||
-- ^ ^ ^ async command processing /
|
||||
|
||||
enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId
|
||||
enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage = do
|
||||
enqueueMessage c cData@ConnData {connId, connAgentVersion} sq@SndQueue {currSndQueue} msgFlags aMessage = do
|
||||
resumeMsgDelivery c cData sq
|
||||
msgId <- storeSentMsg
|
||||
queuePendingMsgs c sq [msgId]
|
||||
@@ -846,23 +853,24 @@ enqueueMessage c cData@ConnData {connId, connAgentVersion} sq msgFlags aMessage
|
||||
encAgentMessage <- agentRatchetEncrypt db connId agentMsgStr e2eEncUserMsgLength
|
||||
let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = connAgentVersion, encAgentMessage}
|
||||
msgType = agentMessageType agentMsg
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, internalHash, prevMsgHash}
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgFlags, msgBody, internalHash, prevMsgHash, currSndQueue}
|
||||
liftIO $ createSndMsg db connId msgData
|
||||
pure internalId
|
||||
|
||||
resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId} = do
|
||||
resumeMsgDelivery c cData@ConnData {connId} sq@SndQueue {server, sndId, currSndQueue = current} = do
|
||||
let qKey = (server, sndId)
|
||||
unlessM (queueDelivering qKey) $ do
|
||||
mq <- atomically $ getPendingMsgQ c sq
|
||||
async (runSmpQueueMsgDelivery c cData mq)
|
||||
unlessM (queueDelivering qKey) $
|
||||
async (runSmpQueueMsgDelivery c cData sq)
|
||||
>>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
|
||||
unlessM connQueued $
|
||||
withStore' c (`getPendingMsgs` connId)
|
||||
withStore' c (\db -> getPendingMsgs db connId current)
|
||||
>>= queuePendingMsgs c sq
|
||||
where
|
||||
queueDelivering qKey = atomically $ TM.member qKey (smpQueueMsgDeliveries c)
|
||||
connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connMsgsQueued c)
|
||||
connQueued = atomically $
|
||||
stateTVar (connMsgsQueued c) $ \s ->
|
||||
let k = (connId, current) in (S.member k s, S.insert k s)
|
||||
|
||||
queuePendingMsgs :: AgentMonad m => AgentClient -> SndQueue -> [InternalId] -> m ()
|
||||
queuePendingMsgs c sq msgIds = atomically $ do
|
||||
@@ -882,8 +890,9 @@ getPendingMsgQ c SndQueue {server, sndId} = do
|
||||
TM.insert qKey mq $ smpQueueMsgQueues c
|
||||
pure mq
|
||||
|
||||
runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> TQueue InternalId -> m ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandshake} mq = do
|
||||
runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandshake} sq = do
|
||||
mq <- atomically $ getPendingMsgQ c sq
|
||||
ri <- asks $ messageRetryInterval . config
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
@@ -895,7 +904,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
|
||||
E.try (withStore c $ \db -> getPendingMsgData db connId msgId) >>= \case
|
||||
Left (e :: E.SomeException) ->
|
||||
notify $ MERR mId (INTERNAL $ show e)
|
||||
Right (rq_, sq, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) ->
|
||||
Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) -> do
|
||||
withRetryInterval ri $ \loop -> do
|
||||
resp <- tryError $ case msgType of
|
||||
AM_CONN_INFO -> sendConfirmation c sq msgBody
|
||||
@@ -1322,11 +1331,11 @@ pickServer = \case
|
||||
getNextSMPServer :: AgentMonad m => AgentClient -> [SMPServer] -> m SMPServer
|
||||
getNextSMPServer c usedSrvs = do
|
||||
srvs <- readTVarIO $ smpServers c
|
||||
case L.nonEmpty $ deleteFirstsBy different (L.toList srvs) usedSrvs of
|
||||
case L.nonEmpty $ deleteFirstsBy sameAddress (L.toList srvs) usedSrvs of
|
||||
Just srvs' -> pickServer srvs'
|
||||
_ -> pickServer srvs
|
||||
where
|
||||
different (SMPServer host port _) (SMPServer host' port' _) = host /= host' || port /= port'
|
||||
sameAddress (SMPServer host port _) (SMPServer host' port' _) = host == host' && port == port'
|
||||
|
||||
subscriber :: (MonadUnliftIO m, MonadReader Env m) => AgentClient -> m ()
|
||||
subscriber c@AgentClient {msgQ} = forever $ do
|
||||
@@ -1559,20 +1568,20 @@ processSMPTransmission c@AgentClient {smpClients, subQ} transmission@(srv, v, se
|
||||
-- processed by queue sender
|
||||
rqNewMsg :: (SMPServer, SMP.SenderId) -> SMPQueueUri -> m ()
|
||||
rqNewMsg (smpServer, senderId) nextQUri = case conn of
|
||||
DuplexConnection _ _ sq@SndQueue {server, sndId} nextRq_ _ -> do
|
||||
liftIO $ print $ "rqNewMsg " <> show (SMP.port server) <> " " <> show (currSndQueue sq)
|
||||
DuplexConnection _ _ sq@SndQueue {server, sndId, currSndQueue = curr} nextRq_ _ -> do
|
||||
liftIO $ print $ "rqNewMsg " <> show (SMP.port server) <> " " <> show curr
|
||||
unless (smpServer == server && senderId == sndId) . throwError $ INTERNAL "incorrect queue address"
|
||||
clientVRange <- asks $ smpClientVRange . config
|
||||
case (nextQUri `compatibleVersion` clientVRange) of
|
||||
Just qInfo@(Compatible nextQInfo) -> do
|
||||
sq'@SndQueue {sndPublicKey, e2ePubKey, server = srv'} <- newSndQueue qInfo False
|
||||
sq'@SndQueue {sndPublicKey, e2ePubKey, server = srv', currSndQueue = curr'} <- newSndQueue qInfo False
|
||||
withStore' c $ \db -> dbCreateNextSndQueue db connId sq sq'
|
||||
liftIO $ print $ "rqNewMsg: next SndQueue " <> show (SMP.port srv') <> " " <> show (currSndQueue sq')
|
||||
liftIO $ print $ "rqNewMsg: next SndQueue " <> show (SMP.port srv') <> " " <> show curr'
|
||||
case (sndPublicKey, e2ePubKey) of
|
||||
(Just nextSenderKey, Just dhPublicKey) -> do
|
||||
let qAddr = (queueAddress (nextQInfo :: SMPQueueInfo)) {dhPublicKey}
|
||||
nextQueueInfo = (nextQInfo :: SMPQueueInfo) {queueAddress = qAddr}
|
||||
void . enqueueMessage c cData sq SMP.noMsgFlags $ QKEYS {nextSenderKey, nextQueueInfo}
|
||||
void $ enqueueMessage c cData sq SMP.noMsgFlags QKEYS {nextSenderKey, nextQueueInfo}
|
||||
let conn' = DuplexConnection cData rq sq nextRq_ (Just sq')
|
||||
notify . SWITCH SPStarted $ connectionStats conn'
|
||||
_ -> throwError $ INTERNAL "absent sender keys"
|
||||
@@ -1604,8 +1613,8 @@ processSMPTransmission c@AgentClient {smpClients, subQ} transmission@(srv, v, se
|
||||
-- processed by queue sender
|
||||
rqReady :: (SMPServer, SMP.SenderId) -> m ()
|
||||
rqReady (smpServer, senderId) = case conn of
|
||||
DuplexConnection _ _ sq@SndQueue {server = srv1} _ nextSq_ -> do
|
||||
liftIO $ print $ "rqReady " <> show (SMP.port srv1) <> " " <> show (currSndQueue sq)
|
||||
DuplexConnection _ _ SndQueue {server = srv1, currSndQueue = curr} _ nextSq_ -> do
|
||||
liftIO $ print $ "rqReady " <> show (SMP.port srv1) <> " " <> show curr
|
||||
case nextSq_ of
|
||||
Just sq'@SndQueue {server, sndId, currSndQueue = curr'} -> do
|
||||
liftIO $ print $ "rqReady next SndQueue " <> show (SMP.port server) <> " " <> show curr'
|
||||
@@ -1645,7 +1654,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} transmission@(srv, v, se
|
||||
switchQueues :: MsgDeliveryKey -> MsgDeliveryKey -> m Bool
|
||||
switchQueues k k' = withStore' c $ \db -> do
|
||||
ok <- atomically $ (switchDeliveries k k' $> True) `orElse` pure False
|
||||
when ok $ switchCurrSndQueue db sq sq'
|
||||
when ok $ switchCurrSndQueue db sq
|
||||
pure ok
|
||||
switchDeliveries :: MsgDeliveryKey -> MsgDeliveryKey -> STM ()
|
||||
switchDeliveries k k' = do
|
||||
@@ -1739,7 +1748,7 @@ enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eE
|
||||
encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength
|
||||
let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption, encConnInfo}
|
||||
msgType = agentMessageType agentMsg
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
|
||||
msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash, currSndQueue = True}
|
||||
liftIO $ createSndMsg db connId msgData
|
||||
pure internalId
|
||||
|
||||
|
||||
@@ -164,7 +164,7 @@ data AgentClient = AgentClient
|
||||
-- Bool in tuple keys shows whether the queue is the current one (True) or the one the connection switches to (False)
|
||||
activeSubs :: TMap2 SMPServer (ConnId, Bool) RcvQueue,
|
||||
pendingSubs :: TMap2 SMPServer (ConnId, Bool) RcvQueue,
|
||||
connMsgsQueued :: TMap ConnId Bool,
|
||||
connMsgsQueued :: TVar (Set (ConnId, Bool)),
|
||||
smpQueueMsgQueues :: TMap MsgDeliveryKey (TQueue InternalId),
|
||||
smpQueueMsgDeliveries :: TMap MsgDeliveryKey (Async ()),
|
||||
nextRcvQueueMsgs :: TMap (SMPServer, SMP.RecipientId) [ServerTransmission BrokerMsg],
|
||||
@@ -219,7 +219,7 @@ newAgentClient InitialAgentServers {smp, ntf, netCfg} agentEnv = do
|
||||
subscrConns <- newTVar S.empty
|
||||
activeSubs <- TM2.empty
|
||||
pendingSubs <- TM2.empty
|
||||
connMsgsQueued <- TM.empty
|
||||
connMsgsQueued <- newTVar S.empty
|
||||
smpQueueMsgQueues <- TM.empty
|
||||
smpQueueMsgDeliveries <- TM.empty
|
||||
nextRcvQueueMsgs <- TM.empty
|
||||
|
||||
@@ -293,7 +293,8 @@ data SndMsgData = SndMsgData
|
||||
msgFlags :: MsgFlags,
|
||||
msgBody :: MsgBody,
|
||||
internalHash :: MsgHash,
|
||||
prevMsgHash :: MsgHash
|
||||
prevMsgHash :: MsgHash,
|
||||
currSndQueue :: Bool
|
||||
}
|
||||
|
||||
data PendingMsgData = PendingMsgData
|
||||
@@ -301,7 +302,8 @@ data PendingMsgData = PendingMsgData
|
||||
msgType :: AgentMessageType,
|
||||
msgFlags :: MsgFlags,
|
||||
msgBody :: MsgBody,
|
||||
internalTs :: InternalTs
|
||||
internalTs :: InternalTs,
|
||||
currSndQueue :: Bool
|
||||
}
|
||||
deriving (Show)
|
||||
|
||||
|
||||
@@ -511,8 +511,8 @@ switchCurrRcvQueue db RcvQueue {server = (SMPServer host port _), rcvId} RcvQueu
|
||||
DB.execute db "DELETE FROM rcv_queues WHERE host = ? AND port = ? AND rcv_id = ? AND curr_rcv_queue = ?" (host, port, rcvId, True)
|
||||
DB.execute db "UPDATE rcv_queues SET curr_rcv_queue = ? WHERE rcv_queue_id = ? AND curr_rcv_queue = ?" (True, dbNextRcvQueueId, False)
|
||||
|
||||
switchCurrSndQueue :: DB.Connection -> SndQueue -> SndQueue -> IO ()
|
||||
switchCurrSndQueue db SndQueue {server = (SMPServer host port _), sndId} SndQueue {dbNextSndQueueId} = do
|
||||
switchCurrSndQueue :: DB.Connection -> SndQueue -> IO ()
|
||||
switchCurrSndQueue db SndQueue {server = (SMPServer host port _), sndId, dbNextSndQueueId} = do
|
||||
DB.execute db "DELETE FROM snd_queues WHERE host = ? AND port = ? AND snd_id = ? AND curr_snd_queue = ?" (host, port, sndId, True)
|
||||
DB.execute db "UPDATE snd_queues SET curr_snd_queue = ? WHERE snd_queue_id = ? AND curr_snd_queue = ?" (True, dbNextSndQueueId, False)
|
||||
|
||||
@@ -686,31 +686,30 @@ createSndMsg db connId sndMsgData = do
|
||||
insertSndMsgDetails_ db connId sndMsgData
|
||||
updateHashSnd_ db connId sndMsgData
|
||||
|
||||
getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, SndQueue, PendingMsgData))
|
||||
getPendingMsgData db connId msgId = runExceptT $ do
|
||||
rq_ <- liftIO $ getRcvQueueByConnId_ db connId
|
||||
sq <- ExceptT $ maybe (Left SEConnNotFound) Right <$> getSndQueueByConnId_ db connId
|
||||
ExceptT $ (rq_,sq,) <$$> firstRow pendingMsgData SEMsgNotFound getMsgData_
|
||||
getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, PendingMsgData))
|
||||
getPendingMsgData db connId msgId = do
|
||||
rq_ <- getRcvQueueByConnId_ db connId
|
||||
(rq_,) <$$> firstRow pendingMsgData SEMsgNotFound getMsgData_
|
||||
where
|
||||
getMsgData_ =
|
||||
DB.query
|
||||
db
|
||||
[sql|
|
||||
SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts
|
||||
SELECT m.msg_type, m.msg_flags, m.msg_body, m.internal_ts, s.curr_snd_queue
|
||||
FROM messages m
|
||||
JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
|
||||
WHERE m.conn_id = ? AND m.internal_id = ?
|
||||
|]
|
||||
(connId, msgId)
|
||||
pendingMsgData :: (AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs) -> PendingMsgData
|
||||
pendingMsgData (msgType, msgFlags_, msgBody, internalTs) =
|
||||
pendingMsgData :: (AgentMessageType, Maybe MsgFlags, MsgBody, InternalTs, Bool) -> PendingMsgData
|
||||
pendingMsgData (msgType, msgFlags_, msgBody, internalTs, currSndQueue) =
|
||||
let msgFlags = fromMaybe SMP.noMsgFlags msgFlags_
|
||||
in PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs}
|
||||
in PendingMsgData {msgId, msgType, msgFlags, msgBody, internalTs, currSndQueue}
|
||||
|
||||
getPendingMsgs :: DB.Connection -> ConnId -> IO [InternalId]
|
||||
getPendingMsgs db connId =
|
||||
getPendingMsgs :: DB.Connection -> ConnId -> Bool -> IO [InternalId]
|
||||
getPendingMsgs db connId current =
|
||||
map fromOnly
|
||||
<$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ?" (Only connId)
|
||||
<$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_id = ? AND curr_snd_queue = ?" (connId, current)
|
||||
|
||||
setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError SMP.MsgId)
|
||||
setMsgUserAck db connId agentMsgId = do
|
||||
@@ -1570,15 +1569,16 @@ insertSndMsgDetails_ dbConn connId SndMsgData {..} =
|
||||
dbConn
|
||||
[sql|
|
||||
INSERT INTO snd_messages
|
||||
( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash)
|
||||
( conn_id, internal_snd_id, internal_id, internal_hash, previous_msg_hash, curr_snd_queue)
|
||||
VALUES
|
||||
(:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash);
|
||||
(:conn_id,:internal_snd_id,:internal_id,:internal_hash,:previous_msg_hash,:curr_snd_queue);
|
||||
|]
|
||||
[ ":conn_id" := connId,
|
||||
":internal_snd_id" := internalSndId,
|
||||
":internal_id" := internalId,
|
||||
":internal_hash" := internalHash,
|
||||
":previous_msg_hash" := prevMsgHash
|
||||
":previous_msg_hash" := prevMsgHash,
|
||||
":curr_snd_queue" := currSndQueue
|
||||
]
|
||||
|
||||
updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
|
||||
|
||||
@@ -53,4 +53,9 @@ UPDATE snd_queues SET updated_at = '1970-01-01 00:00:00';
|
||||
|
||||
CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(snd_queue_id);
|
||||
CREATE UNIQUE INDEX idx_next_snd_queue_id ON snd_queues(next_snd_queue_id);
|
||||
|
||||
-- *
|
||||
|
||||
ALTER TABLE snd_messages ADD COLUMN curr_snd_queue INTEGER CHECK (curr_snd_queue NOT NULL);
|
||||
UPDATE snd_messages SET curr_snd_queue = 1;
|
||||
|]
|
||||
|
||||
@@ -113,6 +113,7 @@ CREATE TABLE snd_messages(
|
||||
internal_id INTEGER NOT NULL,
|
||||
internal_hash BLOB NOT NULL,
|
||||
previous_msg_hash BLOB NOT NULL DEFAULT x'',
|
||||
curr_snd_queue INTEGER CHECK(curr_snd_queue NOT NULL),
|
||||
PRIMARY KEY(conn_id, internal_snd_id),
|
||||
FOREIGN KEY(conn_id, internal_id) REFERENCES messages
|
||||
ON DELETE CASCADE
|
||||
|
||||
@@ -460,7 +460,8 @@ mkSndMsgData internalId internalSndId internalHash =
|
||||
msgFlags = SMP.noMsgFlags,
|
||||
msgBody = hw,
|
||||
internalHash,
|
||||
prevMsgHash = internalHash
|
||||
prevMsgHash = internalHash,
|
||||
currSndQueue = True
|
||||
}
|
||||
|
||||
testCreateSndMsg_ :: DB.Connection -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation
|
||||
|
||||
Reference in New Issue
Block a user