@@ -590,10 +590,7 @@ pruneExpired = \t -> do
590590 a <- *ackedIds
591591 where a.ackedAt > messageCutoff
592592 yield a
593− yield {expiredLocal: (do
594− p <- presenceBefore
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}
597594
598595maybePrune = \t -> do
599596 lastPruned <- *lastPrunedAt
@@ -678,11 +675,7 @@ withSessionAuth = \authorization t onUnauth onAuth -> do
678675
679676removeForward = \mid srv -> atomic do
680677 forwards <- *forwards
681− let updatedForwards = do
682− x <- forwards
683− where not (x.messageId == mid && x.toServer == srv)
684− yield x
685− set *forwards = updatedForwards
678+ set *forwards = forwardsExcluding forwards srv [mid]
686679
687680retryForwardsForServer : [ForwardEntry] -> Int -> IO {network} {}
688681retryForwardsForServer = \entries remaining ->
@@ -711,15 +704,9 @@ forwardQueuedMessages = \targets t -> do
711704 messages <- *messages
712705 forwards <- *forwards
713706 presence <- *presence
714− let localOnline = do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
707+ let localOnline = localOnlineKeys presence t
715708 let remoteTargets = do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
716− let candidates = do
717− target <- remoteTargets
718− m <- messages
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}
709+ let candidates = forwardCandidates remoteTargets messages forwards t
723710 let capped = takeRelation (maxForwardsEntries - count forwards) candidates
724711 let updatedForwards = union forwards capped
725712 set *forwards = updatedForwards
@@ -801,10 +788,7 @@ pollOnce = \token cursor startWakeup -> do
801788 atomic do
802789 wakeup <- *pollWakeup
803790 allMessages <- *messages
804− let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (do
805− m <- allMessages
806− where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
807− yield m))
791+ let limitedMsgs = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (pollableMessages allMessages pubkey clampedCursor t))
808792 if count limitedMsgs > 0
809793 then
810794 let evts = do
@@ -819,6 +803,49 @@ pollOnce = \token cursor startWakeup -> do
819803 then yield (PollOk {events: [], cursor: clampedCursor})
820804 else retry
821805
806+-- ============================================
807+-- Derived Relations
808+-- ============================================
809+
810+presenceForKey = \presence pubkey t ->
811+ do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
812+
813+localOnlineKeys = \presence t ->
814+ do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
815+
816+presenceWithoutLocalKey = \presence pubkey ->
817+ do p <- presence; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
818+
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
821+
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
826+ Nothing {} -> ""
827+
828+pollableMessages = \messages pubkey cursor t ->
829+ do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
830+
831+matchingAckMessages = \messages pubkey messageIds ->
832+ do m <- messages; where m.toKey == pubkey && any messageIds (\mid -> m.messageId == mid); yield m
833+
834+federatedAckEntries = \ackedMessages t ->
835+ do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
836+
837+forwardsExcluding = \forwards server messageIds ->
838+ do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
839+
840+forwardCandidates = \targets messages forwards t ->
841+ do
842+ target <- targets
843+ m <- messages
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}
848+
822849-- ============================================
823850-- Request Handler
824851-- ============================================
@@ -862,10 +889,7 @@ serve = \req -> case req of
862889 let capped = capPerKeyAndGlobal sessions (\s -> s.pubkey == pubkey) maxSessionsPerKey (\s -> s.expiresAt) t maxSessions
863890 let updatedSessions = union capped [{token: token, pubkey: pubkey, expiresAt: expiry}]
864891 set *sessions = updatedSessions
865− let presenceWithout = do
866− p <- presence
867− where p.pubkey != pubkey || not (isLocalServer p.server)
868− yield p
892+ let presenceWithout = presenceWithoutLocalKey presence pubkey
869893 let presenceEvicted = capGlobal presenceWithout maxPresenceEntries (\p -> p.expiresAt)
870894 let updatedPresence = union presenceEvicted [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]
871895 set *presence = updatedPresence
@@ -916,7 +940,7 @@ serve = \req -> case req of
916940 set *messages = union messages
917941 [{messageId: msgId, toKey: to,
918942 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
920944 let localOnline = any toPresence (\p -> isLocalServer p.server)
921945 let remoteServers = collectPeerServers toPresence (\p -> p.server)
922946 forwards <- *forwards
@@ -958,16 +982,10 @@ serve = \req -> case req of
958982 (yield {pr: pr, ack: AckUnauthorized {}})
959983 (\pubkey -> do
960984 messages <- *messages
961− let ackedMessages = do
962− m <- messages
963− where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)
964− yield m
985+ let ackedMessages = matchingAckMessages messages pubkey dedupedIds
965986 let remainingMessages = diff messages ackedMessages
966987 set *messages = remainingMessages
967− let ackedFederated = do
968− m <- ackedMessages
969− where m.originServer != ""
970− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
988+ let ackedFederated = federatedAckEntries ackedMessages t
971989 ackedIds <- *ackedIds
972990 let combinedAckedIds = union ackedIds ackedFederated
973991 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
@@ -993,15 +1011,7 @@ serve = \req -> case req of
9931011 (yield {pr: pr, server: "", authorized: False {}})
9941012 (\_ -> do
9951013 presence <- *presence
996− let valid = do
997− p <- presence
998− where p.pubkey == pubkey && p.expiresAt > t
999− yield p
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 {} -> ""
1014+ let foundServer = bestServerForKey (presenceForKey presence pubkey t)
10051015 yield {pr: pr, server: foundServer, authorized: True {}})
10061016 forkBroadcastOffline t result.pr
10071017 yield (respond {server: result.server, authorized: result.authorized})
@@ -1075,11 +1085,7 @@ serve = \req -> case req of
10751085 pruneResult <- atomic do
10761086 pr <- maybePrune t
10771087 forwards <- *forwards
1078− let remainingForwards = do
1079− f <- forwards
1080− where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1081− yield f
1082− set *forwards = remainingForwards
1088+ set *forwards = forwardsExcluding forwards fromServer dedupedIds
10831089 yield pr
10841090 forkBroadcastOffline t pruneResult
10851091 yield (respond {ok: True {}}))