smp server: short links and owners for channels (#1506)

* smp server: short links and owners for channels

* types

* support mutliple rcv keys

* fix down migration, test/create server schema dump

* reduce schema dump

* parameterize type for link data by connection type

* return full connection link data

* test version

* change short link encoding

* test: print pg_dump output

* server pages, link encoding

* fix connection request when queue data and sender ID are created for old servers

* test, change pattern

* ci: install postgresql tools in runner (#1507)

* ci: install postgresql tools in runner

* ci: docker shell abort on error

* fix pattern for ghc 8.10.7

* patch ConnReqUriData SMP encoding to preserve queue mode after decoding

* test for RKEY

* fix/test store log with RKEY

---------

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>
This commit is contained in:
Evgeny
2025-04-10 19:09:47 +01:00
committed by GitHub
co-authored by sh
parent 3a3f9fd51e
commit b7a9542213
32 changed files with 870 additions and 269 deletions
@@ -41,10 +41,10 @@ import Data.Either (fromRight)
import Data.Functor (($>))
import Data.Int (Int64)
import Data.List (intersperse)
import Data.List.NonEmpty (NonEmpty)
import qualified Data.Map.Strict as M
import Data.Maybe (catMaybes, fromMaybe)
import qualified Data.Text as T
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import Data.Time.Clock.System (SystemTime (..), getSystemTime)
import Database.PostgreSQL.Simple (Binary (..), Only (..), Query, SqlError, (:.) (..))
import qualified Database.PostgreSQL.Simple as DB
@@ -58,6 +58,8 @@ import Simplex.Messaging.Agent.Client (withLockMap)
import Simplex.Messaging.Agent.Lock (Lock)
import Simplex.Messaging.Agent.Store.Postgres (createDBStore, closeDBStore)
import Simplex.Messaging.Agent.Store.Postgres.Common
import Simplex.Messaging.Agent.Store.Postgres.DB (blobFieldDecoder)
import qualified Simplex.Messaging.Crypto as C
import Simplex.Messaging.Encoding
import Simplex.Messaging.Protocol
import Simplex.Messaging.Server.QueueStore
@@ -74,8 +76,8 @@ import System.IO (IOMode (..), hFlush, stdout)
import UnliftIO.STM
#if !defined(dbPostgres)
import Simplex.Messaging.Agent.Store.Postgres.DB (blobFieldDecoder, fromTextField_)
import qualified Simplex.Messaging.Crypto as C
import Data.Text.Encoding (decodeLatin1, encodeUtf8)
import Simplex.Messaging.Agent.Store.Postgres.DB (fromTextField_)
import Simplex.Messaging.Encoding.String
#endif
@@ -257,6 +259,16 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
Just k | sKey /= k -> throwE AUTH
_ -> pure ()
updateKeys :: PostgresQueueStore q -> q -> NonEmpty RcvPublicAuthKey -> IO (Either ErrorType ())
updateKeys st sq rKeys =
withQueueRec sq "updateKeys" $ \q -> do
assertUpdated $ withDB' "updateKeys" st $ \db ->
DB.execute db "UPDATE msg_queues SET recipient_keys = ? WHERE recipient_id = ? AND deleted_at IS NULL" (rKeys, rId)
atomically $ writeTVar (queueRec sq) $ Just q {recipientKeys = rKeys}
withLog "updateKeys" st $ \s -> logUpdateKeys s rId rKeys
where
rId = recipientId sq
addQueueNotifier :: PostgresQueueStore q -> q -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId, notifierKey, rcvNtfDhSecret} =
withQueueRec sq "addQueueNotifier" $ \q ->
@@ -362,7 +374,7 @@ batchInsertQueues tty queues toStore = do
let st = dbStore toStore
count <-
withConnection st $ \db -> do
DB.copy_ db "COPY msg_queues (recipient_id, recipient_key, rcv_dh_secret, sender_id, sender_key, snd_secure, notifier_id, notifier_key, rcv_ntf_dh_secret, status, updated_at) FROM STDIN WITH (FORMAT CSV)"
DB.copy_ db "COPY msg_queues (recipient_id, recipient_keys, rcv_dh_secret, sender_id, sender_key, snd_secure, notifier_id, notifier_key, rcv_ntf_dh_secret, status, updated_at) FROM STDIN WITH (FORMAT CSV)"
mapM_ (putQueue db) (zip [1..] qs)
DB.putCopyEnd db
Only qCnt : _ <- withConnection st (`DB.query_` "SELECT count(*) FROM msg_queues")
@@ -378,7 +390,7 @@ insertQueueQuery :: Query
insertQueueQuery =
[sql|
INSERT INTO msg_queues
(recipient_id, recipient_key, rcv_dh_secret, sender_id, sender_key, queue_mode, notifier_id, notifier_key, rcv_ntf_dh_secret, status, updated_at, link_id, fixed_data, user_data)
(recipient_id, recipient_keys, rcv_dh_secret, sender_id, sender_key, queue_mode, notifier_id, notifier_key, rcv_ntf_dh_secret, status, updated_at, link_id, fixed_data, user_data)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|]
@@ -407,7 +419,7 @@ foldQueueRecs tty withData st skipOld_ f = do
queueRecQuery :: Query
queueRecQuery =
[sql|
SELECT recipient_id, recipient_key, rcv_dh_secret,
SELECT recipient_id, recipient_keys, rcv_dh_secret,
sender_id, sender_key, queue_mode,
notifier_id, notifier_key, rcv_ntf_dh_secret,
status, updated_at,
@@ -418,7 +430,7 @@ queueRecQuery =
queueRecQueryWithData :: Query
queueRecQueryWithData =
[sql|
SELECT recipient_id, recipient_key, rcv_dh_secret,
SELECT recipient_id, recipient_keys, rcv_dh_secret,
sender_id, sender_key, queue_mode,
notifier_id, notifier_key, rcv_ntf_dh_secret,
status, updated_at,
@@ -426,23 +438,23 @@ queueRecQueryWithData =
FROM msg_queues
|]
type QueueRecRow = (RecipientId, RcvPublicAuthKey, RcvDhSecret, SenderId, Maybe SndPublicAuthKey, Maybe QueueMode, Maybe NotifierId, Maybe NtfPublicAuthKey, Maybe RcvNtfDhSecret, ServerEntityStatus, Maybe RoundedSystemTime, Maybe LinkId)
type QueueRecRow = (RecipientId, NonEmpty RcvPublicAuthKey, RcvDhSecret, SenderId, Maybe SndPublicAuthKey, Maybe QueueMode, Maybe NotifierId, Maybe NtfPublicAuthKey, Maybe RcvNtfDhSecret, ServerEntityStatus, Maybe RoundedSystemTime, Maybe LinkId)
queueRecToRow :: (RecipientId, QueueRec) -> QueueRecRow :. (Maybe EncDataBytes, Maybe EncDataBytes)
queueRecToRow (rId, QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier = n, status, updatedAt}) =
(rId, recipientKey, rcvDhSecret, senderId, senderKey, queueMode, notifierId <$> n, notifierKey <$> n, rcvNtfDhSecret <$> n, status, updatedAt, linkId_)
queueRecToRow (rId, QueueRec {recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier = n, status, updatedAt}) =
(rId, recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, notifierId <$> n, notifierKey <$> n, rcvNtfDhSecret <$> n, status, updatedAt, linkId_)
:. (fst <$> queueData_, snd <$> queueData_)
where
(linkId_, queueData_) = queueDataColumns queueData
queueRecToText :: (RecipientId, QueueRec) -> ByteString
queueRecToText (rId, QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier = n, status, updatedAt}) =
queueRecToText (rId, QueueRec {recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier = n, status, updatedAt}) =
LB.toStrict $ BB.toLazyByteString $ mconcat tabFields <> BB.char7 '\n'
where
tabFields = BB.char7 ',' `intersperse` fields
fields =
[ renderField (toField rId),
renderField (toField recipientKey),
renderField (toField recipientKeys),
renderField (toField rcvDhSecret),
renderField (toField senderId),
nullable senderKey,
@@ -473,17 +485,17 @@ queueDataColumns = \case
Nothing -> (Nothing, Nothing)
rowToQueueRec :: QueueRecRow -> (RecipientId, QueueRec)
rowToQueueRec (rId, recipientKey, rcvDhSecret, senderId, senderKey, queueMode, notifierId_, notifierKey_, rcvNtfDhSecret_, status, updatedAt, linkId_) =
rowToQueueRec (rId, recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, notifierId_, notifierKey_, rcvNtfDhSecret_, status, updatedAt, linkId_) =
let notifier = NtfCreds <$> notifierId_ <*> notifierKey_ <*> rcvNtfDhSecret_
queueData = (,(EncDataBytes "", EncDataBytes "")) <$> linkId_
in (rId, QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier, status, updatedAt})
in (rId, QueueRec {recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier, status, updatedAt})
rowToQueueRecWithData :: QueueRecRow :. (Maybe EncDataBytes, Maybe EncDataBytes) -> (RecipientId, QueueRec)
rowToQueueRecWithData ((rId, recipientKey, rcvDhSecret, senderId, senderKey, queueMode, notifierId_, notifierKey_, rcvNtfDhSecret_, status, updatedAt, linkId_) :. (immutableData_, userData_)) =
rowToQueueRecWithData ((rId, recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, notifierId_, notifierKey_, rcvNtfDhSecret_, status, updatedAt, linkId_) :. (immutableData_, userData_)) =
let notifier = NtfCreds <$> notifierId_ <*> notifierKey_ <*> rcvNtfDhSecret_
encData = fromMaybe (EncDataBytes "")
queueData = (,(encData immutableData_, encData userData_)) <$> linkId_
in (rId, QueueRec {recipientKey, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier, status, updatedAt})
in (rId, QueueRec {recipientKeys, rcvDhSecret, senderId, senderKey, queueMode, queueData, notifier, status, updatedAt})
setStatusDB :: StoreQueueClass q => String -> PostgresQueueStore q -> q -> ServerEntityStatus -> ExceptT ErrorType IO () -> IO (Either ErrorType ())
setStatusDB op st sq status writeLog =
@@ -528,6 +540,10 @@ instance ToField EntityId where toField (EntityId s) = toField $ Binary s
deriving newtype instance FromField EntityId
instance ToField (NonEmpty C.APublicAuthKey) where toField = toField . Binary . smpEncode
instance FromField (NonEmpty C.APublicAuthKey) where fromField = blobFieldDecoder smpDecode
#if !defined(dbPostgres)
instance FromField QueueMode where fromField = fromTextField_ $ eitherToMaybe . smpDecode . encodeUtf8