@@ -590,10 +590,7 @@ pruneExpired = \t -> do
590 a <- *ackedIds
590 a <- *ackedIds
591 where a.ackedAt > messageCutoff
591 where a.ackedAt > messageCutoff
593− yield {expiredLocal: (do
595− where p.expiresAt <= t && isLocalServer p.server && none presenceAfter (\q -> q.pubkey == p.pubkey && isLocalServer q.server)
596− yield p), peerServers: peerServersBefore}
593+ yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
598maybePrune = \t -> do
595maybePrune = \t -> do
599 lastPruned <- *lastPrunedAt
596 lastPruned <- *lastPrunedAt
@@ -678,11 +675,7 @@ withSessionAuth = \authorization t onUnauth onAuth -> do
679removeForward = \mid srv -> atomic do
676removeForward = \mid srv -> atomic do
680 forwards <- *forwards
677 forwards <- *forwards
681− let updatedForwards = do
678+ set *forwards = forwardsExcluding forwards srv [mid]
683− where not (x.messageId == mid && x.toServer == srv)
685− set *forwards = updatedForwards
687retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
680retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
688retryForwardsForServer = \entries remaining ->
681retryForwardsForServer = \entries remaining ->
@@ -711,15 +704,9 @@ forwardQueuedMessages = \targets t -> do
711 messages <- *messages
704 messages <- *messages
712 forwards <- *forwards
705 forwards <- *forwards
713 presence <- *presence
706 presence <- *presence
714− let localOnline = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
707+ let localOnline = localOnlineKeys presence t
715 let remoteTargets = do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
708 let remoteTargets = do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
716− let candidates = do
709+ let candidates = forwardCandidates remoteTargets messages forwards t
717− target <- remoteTargets
719− where m.toKey == target.pubkey
720− where none forwards (\f -> f.messageId == m.messageId && f.toServer == target.server)
721− yield {messageId: m.messageId, toServer: target.server, createdAt: t,
722− toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
723 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
710 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
724 let updatedForwards = union forwards capped
711 let updatedForwards = union forwards capped
725 set *forwards = updatedForwards
712 set *forwards = updatedForwards
@@ -801,10 +788,7 @@ pollOnce = \token cursor startWakeup -> do
801 atomic do
788 atomic do
802 wakeup <- *pollWakeup
789 wakeup <- *pollWakeup
803 allMessages <- *messages
790 allMessages <- *messages
804− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (do
791+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t))
806− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
808 if count limitedMsgs > 0
792 if count limitedMsgs > 0
810 let evts = do
794 let evts = do
@@ -819,6 +803,49 @@ pollOnce = \token cursor startWakeup -> do
819 then yield (PollOk {events: [], cursor: clampedCursor})
803 then yield (PollOk {events: [], cursor: clampedCursor})
820 else retry
804 else retry
806+-- ============================================
808+-- ============================================
810+presenceForKey = \presence pubkey t ->
811+ do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
813+localOnlineKeys = \presence t ->
814+ do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
816+presenceWithoutLocalKey = \presence pubkey ->
817+ do p <- presence; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
819+expiredLocalPresence = \before after t ->
820+ do p <- before; where p.expiresAt <= t && isLocalServer p.server && none after (\q -> q.pubkey == p.pubkey && isLocalServer q.server); yield p
822+bestServerForKey = \entries ->
823+ if any entries (\p -> isLocalServer p.server) then serverHostN
824+ else case maxBy (do p <- entries; where not (isLocalServer p.server); yield p) (\p -> p.expiresAt) of
825+ Just {value: p} -> p.server
828+pollableMessages = \messages pubkey cursor t ->
829+ do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
831+matchingAckMessages = \messages pubkey messageIds ->
832+ do m <- messages; where m.toKey == pubkey && any messageIds (\mid -> m.messageId == mid); yield m
834+federatedAckEntries = \ackedMessages t ->
835+ do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
837+forwardsExcluding = \forwards server messageIds ->
838+ do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
840+forwardCandidates = \targets messages forwards t ->
844+ where m.toKey == target.pubkey
845+ where none forwards (\f -> f.messageId == m.messageId && f.toServer == target.server)
846+ yield {messageId: m.messageId, toServer: target.server, createdAt: t,
847+ toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp}
822-- ============================================
849-- ============================================
823-- Request Handler
850-- Request Handler
824-- ============================================
851-- ============================================
@@ -862,10 +889,7 @@ serve = \req -> case req of
862 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
889 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
863 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
890 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
864 set *sessions = updatedSessions
891 set *sessions = updatedSessions
865− let presenceWithout = do
892+ let presenceWithout = presenceWithoutLocalKey presence pubkey
867− where p.pubkey != pubkey || not (isLocalServer p.server)
869 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
893 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
870 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
894 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
871 set *presence = updatedPresence
895 set *presence = updatedPresence
@@ -916,7 +940,7 @@ serve = \req -> case req of
916 set *messages = union messages
940 set *messages = union messages
917 [{messageId: msgId, toKey: to,
941 [{messageId: msgId, toKey: to,
918 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
942 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
919− let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
943+ let toPresence = presenceForKey presence to t
920 let localOnline = any toPresence (\p -> isLocalServer p.server)
944 let localOnline = any toPresence (\p -> isLocalServer p.server)
921 let remoteServers = collectPeerServers toPresence (\p -> p.server)
945 let remoteServers = collectPeerServers toPresence (\p -> p.server)
922 forwards <- *forwards
946 forwards <- *forwards
@@ -958,16 +982,10 @@ serve = \req -> case req of
958 (yield {pr: pr, ack: AckUnauthorized {}})
982 (yield {pr: pr, ack: AckUnauthorized {}})
959 (\pubkey -> do
983 (\pubkey -> do
960 messages <- *messages
984 messages <- *messages
961− let ackedMessages = do
985+ let ackedMessages = matchingAckMessages messages pubkey dedupedIds
963− where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
965 let remainingMessages = diff messages ackedMessages
986 let remainingMessages = diff messages ackedMessages
966 set *messages = remainingMessages
987 set *messages = remainingMessages
967− let ackedFederated = do
988+ let ackedFederated = federatedAckEntries ackedMessages t
969− where m.originServer != ""
970− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
971 ackedIds <- *ackedIds
989 ackedIds <- *ackedIds
972 let combinedAckedIds = union ackedIds ackedFederated
990 let combinedAckedIds = union ackedIds ackedFederated
973 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
991 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
@@ -993,15 +1011,7 @@ serve = \req -> case req of
993 (yield {pr: pr, server: "", authorized: False {}})
1011 (yield {pr: pr, server: "", authorized: False {}})
994 (\_ -> do
1012 (\_ -> do
995 presence <- *presence
1013 presence <- *presence
996− let valid = do
1014+ let foundServer = bestServerForKey (presenceForKey presence pubkey t)
998− where p.pubkey == pubkey && p.expiresAt > t
1000− let localOnline = any valid (\p -> isLocalServer p.server)
1001− let remoteValid = do p <- valid; where not (isLocalServer p.server); yield p
1002− let bestRemote = maxBy remoteValid (\p -> p.expiresAt)
1003− let foundServer = if localOnline then serverHostN
1004− else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
1005 yield {pr: pr, server: foundServer, authorized: True {}})
1015 yield {pr: pr, server: foundServer, authorized: True {}})
1006 forkBroadcastOffline t result.pr
1016 forkBroadcastOffline t result.pr
1007 yield (respond {server: result.server, authorized: result.authorized})
1017 yield (respond {server: result.server, authorized: result.authorized})
@@ -1075,11 +1085,7 @@ serve = \req -> case req of
1075 pruneResult <- atomic do
1085 pruneResult <- atomic do
1076 pr <- maybePrune t
1086 pr <- maybePrune t
1077 forwards <- *forwards
1087 forwards <- *forwards
1078− let remainingForwards = do
1088+ set *forwards = forwardsExcluding forwards fromServer dedupedIds
1080− where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1082− set *forwards = remainingForwards
1083 yield pr
1089 yield pr
1084 forkBroadcastOffline t pruneResult
1090 forkBroadcastOffline t pruneResult
1085 yield (respond {ok: True {}}))
1091 yield (respond {ok: True {}}))