@@ -41,7 +41,7 @@ federationEnabled = not (federationSecret == "")
4141-- Types
4242-- ============================================
4343
44−type PollEvent = {id: Text, eventType: Text, payload: {encryptedBlob: Text, fromKey: Text, timestamp: Timestamp, toKey: Text}}
44+type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, fromKey: PubkeyHex, timestamp: Timestamp, toKey: PubkeyHex}}
4545
4646type GossipEventType = Text where \s -> s == "online" || s == "offline"
4747
@@ -379,6 +379,12 @@ clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresen
379379
380380clampTimestamp = \timestamp t -> if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
381381
382+nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
383+
384+isFederatedOrigin = \originServer -> not (originServer == "") && not (originServer == serverHostN)
385+
386+takeN = \items n -> fold (\acc e -> if count acc >= n then acc else union acc [e]) [] items
387+
382388dedupMessageIds = \messageIds ->
383389 fold (\acc mid ->
384390 let nmid = normalizeHex mid in
@@ -544,14 +550,12 @@ maybePrune = \t -> do
544550 set *lastPrunedAt = t
545551 pruneExpired t
546552
547−sendBatchedGossipToServer : Text -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int}] -> IO {network} {}
553+sendBatchedGossipToServer : Text -> [{eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}] -> IO {network} {}
548554sendBatchedGossipToServer = \server allEvents ->
549555 if count allEvents == 0
550556 then yield {}
551557 else do
552− let batch = fold (\acc e ->
553− if count acc >= maxGossipEvents then acc
554− else union acc [e]) [] allEvents
558+ let batch = takeN allEvents maxGossipEvents
555559 let rest = diff allEvents batch
556560 batchResult <- fetch (federationScheme ++ server) (RecvGossip {
557561 authorization: "Bearer " ++ federationSecret,
@@ -561,7 +565,7 @@ sendBatchedGossipToServer = \server allEvents ->
561565 Err _ -> yield {}
562566 sendBatchedGossipToServer server rest
563567
564−sendBatchedGossip : [Text] -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int}] -> IO {network} {}
568+sendBatchedGossip : [Text] -> [{eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}] -> IO {network} {}
565569sendBatchedGossip = \peerServers allEvents ->
566570 if count allEvents == 0
567571 then yield {}
@@ -582,7 +586,7 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
582586 sendBatchedGossip peerServers gossipEvents
583587 else yield {}
584588
585−retryForwardsForServer : [{encryptedBlob: Text, fromKey: Text, messageId: Text, timestamp: Timestamp, toKey: Text, toServer: Text}] -> Int -> IO {network} {}
589+retryForwardsForServer : [{encryptedBlob: BlobHex, fromKey: PubkeyHex, messageId: MessageIdHex, timestamp: Timestamp, toKey: PubkeyHex, toServer: ServerName}] -> Int -> IO {network} {}
586590retryForwardsForServer = \entries remaining ->
587591 if count entries == 0 || remaining <= 0
588592 then yield {}
@@ -608,21 +612,19 @@ retryForwardsForServer = \entries remaining ->
608612 Err _ -> yield {}
609613 retryForwardsForServer rest (remaining - 1)
610614
611−retryPendingForwards : [{encryptedBlob: Text, fromKey: Text, messageId: Text, timestamp: Timestamp, toKey: Text, toServer: Text}] -> IO {network} {}
615+retryPendingForwards : [{encryptedBlob: BlobHex, fromKey: PubkeyHex, messageId: MessageIdHex, timestamp: Timestamp, toKey: PubkeyHex, toServer: ServerName}] -> IO {network} {}
612616retryPendingForwards = \pending -> do
613617 let servers = collectPeerServers pending (\x -> x.toServer)
614618 sendToEach servers (\server ->
615619 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
616620 yield {}
617621
618−sendBatchedAcksToServer : [{messageId: Text, toKey: Text, originServer: Text, | a}] -> Text -> IO {network} {}
622+sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> Text -> IO {network} {}
619623sendBatchedAcksToServer = \serverAcks server ->
620624 if count serverAcks == 0
621625 then yield {}
622626 else do
623− let batch = fold (\acc a ->
624− if count acc >= maxMessageIds then acc
625− else union acc [a]) [] serverAcks
627+ let batch = takeN serverAcks maxMessageIds
626628 let rest = diff serverAcks batch
627629 result <- fetch (federationScheme ++ server) (FederationAck {
628630 authorization: "Bearer " ++ federationSecret,
@@ -639,7 +641,7 @@ sendBatchedAcksToServer = \serverAcks server ->
639641 Err _ -> yield {}
640642 sendBatchedAcksToServer rest server
641643
642−sendAcksAndClean : [{messageId: Text, toKey: Text, originServer: Text, | a}] -> [Text] -> IO {network} {}
644+sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> [Text] -> IO {network} {}
643645sendAcksAndClean = \pendingAcks servers ->
644646 sendToEach servers (\server ->
645647 sendBatchedAcksToServer (do a <- pendingAcks; where a.originServer == server; yield a) server)
@@ -655,8 +657,7 @@ pruneAndRetry = \t -> do
655657 ackedIds <- *ackedIds
656658 let pendingAcks = do
657659 a <- ackedIds
658− where not (a.originServer == "")
659− where not (a.originServer == serverHostN)
660+ where isFederatedOrigin a.originServer
660661 yield a
661662 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
662663 broadcastOfflineGossip t result.peerServers result.expiredLocal
@@ -832,7 +833,7 @@ serve = \req -> case req of
832833 Just {value: s} -> s.pubkey
833834 let authorized = isJust session
834835 seqCounter <- *seqCounter
835− let seqTs = if t > seqCounter then t else seqCounter + 1
836+ let seqTs = nextSeqTs t seqCounter
836837 let recipientMsgCount = countMessagesFor messages to t
837838 let shouldInsert = authorized && recipientMsgCount < maxMessagesPerRecipient
838839 set *seqCounter = if shouldInsert then seqTs else seqCounter
@@ -858,9 +859,7 @@ serve = \req -> case req of
858859 forwards <- *forwards
859860 let forwardsAtCapacity = count forwards >= maxForwardsEntries
860861 let forwardSlots = maxForwardsEntries - count forwards
861− let forwardServers = fold (\acc s ->
862− if count acc >= forwardSlots then acc
863− else union acc [s]) [] remoteServers
862+ let forwardServers = takeN remoteServers forwardSlots
864863 set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled
865864 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
866865 else forwards
@@ -910,8 +909,7 @@ serve = \req -> case req of
910909 Just {value: _} -> diff messages ackedMessages
911910 let ackedFederated = do
912911 m <- ackedMessages
913− where not (m.originServer == "")
914− where not (m.originServer == serverHostN)
912+ where isFederatedOrigin m.originServer
915913 yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
916914 ackedIds <- *ackedIds
917915 let combinedAckedIds = union ackedIds ackedFederated
@@ -925,8 +923,7 @@ serve = \req -> case req of
925923 yield {ok: ok, ackedMessages: ackedMessages, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
926924 let fedMessages = do
927925 m <- result.ackedMessages
928− where not (m.originServer == "")
929− where not (m.originServer == serverHostN)
926+ where isFederatedOrigin m.originServer
930927 yield m
931928 let originServers = collectPeerServers fedMessages (\m -> m.originServer)
932929 fork (do
@@ -1063,9 +1060,7 @@ serve = \req -> case req of
10631060 where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e)
10641061 yield e
10651062 let reclaimedEntries = if withinCapacity then []
1066− else fold (\acc e ->
1067− if count acc >= reclaimCapacity then acc
1068− else union acc [e]) [] newEntries
1063+ else takeN newEntries reclaimCapacity
10691064 let afterReclaim = if withinCapacity then afterOffline
10701065 else union updatesOnlyAfterOffline reclaimedEntries
10711066 let appliedOnlineUpdates = if withinCapacity then appliedOnline
@@ -1130,7 +1125,7 @@ serve = \req -> case req of
11301125 pr <- maybePrune t
11311126 messages <- *messages
11321127 seqCounter <- *seqCounter
1133− let seqTs = if t > seqCounter then t else seqCounter + 1
1128+ let seqTs = nextSeqTs t seqCounter
11341129 let recipientMsgCount = if not (authorized && validOrigin) then 0
11351130 else countMessagesFor messages toKey t
11361131 let existsInMessages = if not (authorized && validOrigin) then False {}