@@ -439,19 +439,23 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
439439-- Generic helpers
440440-- ============================================
441441
442+filter = \xs pred -> do x <- xs; where pred x; yield x
443+pluck = \xs f -> do x <- xs; yield (f x)
444+countWhere = \xs pred -> count (filter xs pred)
442445any : [a] -> (a -> Bool) -> Bool
443−any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0
446+any = \xs pred -> countWhere xs pred > 0
444447none : [a] -> (a -> Bool) -> Bool
445448none = \xs pred -> not (any xs pred)
446449all : [a] -> (a -> Bool) -> Bool
447450all = \xs pred -> none xs (\x -> not (pred x))
451+elem = \xs x -> any xs (\y -> y == x)
448452when = \cond action -> if cond then action else yield {}
449453clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
450454min = \a b -> if a < b then a else b
451455max = \a b -> if a > b then a else b
452456
453457-- Dedup: keeps first occurrence (O(n^2)).
454−unique = \items -> fold (\acc x -> if any acc (\a -> a == x) then acc else union acc [x]) [] items
458+unique = \items -> fold (\acc x -> if elem acc x then acc else union acc [x]) [] items
455459
456460-- Replace a ref's contents with the entries that satisfy `keep`. Caller controls atomicity.
457461pruneRef = \ref keep -> do
@@ -596,9 +600,25 @@ withMaintAuth = \authorization onUnauth onAuth -> do
596600 authorized <- verifyMaintenanceAuth authorization
597601 if authorized then onAuth else onUnauth
598602
603+-- Common gate for client handlers: drain check, global rate, session auth.
604+-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
605+gateAuth = \auth ip t respond mkErr onAuth -> do
606+ dp <- *drainPhase
607+ if dp >= 2 then yield (respond (mkErr "draining"))
608+ else do
609+ g <- checkGlobalRate t
610+ if g then yield (respond (mkErr "rate_limited"))
611+ else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth
612+
613+-- Variant without the drain check (used by lookup/status handlers).
614+gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do
615+ g <- checkGlobalRate t
616+ if g then yield (respond (mkErr "rate_limited"))
617+ else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth
618+
599619authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
600620authenticateSession = \sessions token t ->
601− case headOf (do s <- sessions; where s.expiresAt > t && s.token == token; yield s) of
621+ case headOf (filter sessions (\s -> s.expiresAt > t && s.token == token)) of
602622 Nothing {} -> Unauthorized {}
603623 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
604624
@@ -612,12 +632,6 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
612632 if matchIp clientIp ip then onAuth pubkey else onUnauth
613633 _ -> onUnauth
614634
615−isBlockedPeer : Text -> [Text] -> Bool
616−isBlockedPeer = \server blocked -> any blocked (\b -> b == server)
617−
618−isAllowedByAllowlist : Text -> [Text] -> Bool
619−isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)
620−
621635splitOnColon : Text -> {before: Text, after: Text}
622636splitOnColon = \s ->
623637 if s == "" then {before: "", after: ""}
@@ -668,8 +682,8 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
668682 blocked <- *blockedPeers
669683 allowed <- *allowedPeers
670684 let preErr =
671− if isBlockedPeer fromServer blocked then "blocked"
672− else if not (isAllowedByAllowlist fromServer allowed) then "not_allowed"
685+ if elem blocked fromServer then "blocked"
686+ else if count allowed > 0 && not (elem allowed fromServer) then "not_allowed"
673687 else if not precondition then "invalid_request"
674688 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance) then "clock_skew"
675689 else ""
@@ -689,14 +703,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
689703-- Returns {limited, entries}. Increments count when the key fits within the window.
690704-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.
691705checkRate = \entries key maxRate windowMs maxEntries t ->
692− let cur = headOf (do r <- entries; where r.key == key && t - r.windowStart < windowMs; yield r) in
706+ let cur = headOf (filter entries (\r -> r.key == key && t - r.windowStart < windowMs)) in
693707 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in
694708 if nextCount > maxRate
695709 then {limited: True {}, entries: entries}
696710 else
697− let others = do x <- entries; where x.key != key; yield x in
698711 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in
699− {limited: False {}, entries: capList (union others [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}
712+ {limited: False {}, entries: capList (union (filter entries (\x -> x.key != key)) [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}
700713
701714checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
702715 ws <- *windowRef
@@ -744,7 +757,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
744757
745758shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
746759shouldRetryServer = \server t failures ->
747− case headOf (do f <- failures; where f.server == server; yield f) of
760+ case headOf (filter failures (\f -> f.server == server)) of
748761 Nothing {} -> True {}
749762 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures
750763
@@ -752,11 +765,10 @@ recordFailure = \server -> do
752765 t <- now
753766 atomic do
754767 failures <- *forwardFailures
755− let others = do f <- failures; where f.server != server; yield f
756− let new = case headOf (do f <- failures; where f.server == server; yield f) of
768+ let new = case headOf (filter failures (\f -> f.server == server)) of
757769 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
758770 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
759− set *forwardFailures = union others [new]
771+ set *forwardFailures = union (filter failures (\f -> f.server != server)) [new]
760772
761773clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
762774
@@ -782,7 +794,7 @@ dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
782794dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)
783795
784796presenceForKey = \presence pubkey t ->
785− do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
797+ filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)
786798
787799-- Pick local if present, else the remote with the latest expiresAt; "" when none.
788800bestServerForKey = \entries ->
@@ -796,7 +808,7 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey
796808
797809collectPeers : [a] -> (a -> Text) -> [Text]
798810collectPeers = \items getServer ->
799− unique (do x <- items; let s = getServer x; where not (isLocalServer s); yield s)
811+ unique (filter (pluck items getServer) (\s -> not (isLocalServer s)))
800812
801813knownPeerServersFromPresence : [PresenceEntry] -> [Text]
802814knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
@@ -805,7 +817,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
805817mergePresence = \existing incoming ->
806818 fold (\acc e ->
807819 if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt) then acc
808− else union (do x <- acc; where x.pubkey != e.pubkey || x.server != e.server; yield x) [e]) existing incoming
820+ else union (filter acc (\x -> x.pubkey != e.pubkey || x.server != e.server)) [e]) existing incoming
809821
810822-- ============================================
811823-- Pruning
@@ -827,9 +839,9 @@ pruneExpired = \t -> atomic do
827839 pruneRef pendingFederationCancels (\x -> t - x.createdAt < messageMaxAge)
828840 -- presence: remember locally-expired keys for offline gossip
829841 pres <- *presence
830− let keptPres = do p <- pres; where p.expiresAt > t; yield p
842+ let keptPres = filter pres (\p -> p.expiresAt > t)
831843 set *presence = keptPres
832− yield (do p <- pres; where p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server); yield p)
844+ yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server)))
833845
834846-- ============================================
835847-- Federation gossip
@@ -852,8 +864,7 @@ broadcastOfflineGossip = \t expiredLocal -> do
852864 presence <- *presence
853865 let peers = knownPeerServersFromPresence presence
854866 when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do
855− let events = do p <- expiredLocal; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}
856− sendBatchedGossip peers events gossipOriginHops)
867+ sendBatchedGossip peers (pluck expiredLocal (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops)
857868
858869-- ============================================
859870-- Forward retries and acks
@@ -864,9 +875,9 @@ retryPendingForwards = do
864875 fwds <- *forwards
865876 failures <- *forwardFailures
866877 blocked <- *blockedPeers
867− let servers = do s <- collectPeers fwds (\f -> f.toServer); where shouldRetryServer s t failures && not (isBlockedPeer s blocked); yield s
878+ let servers = filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))
868879 forEach servers (\s -> s) (\server -> do
869− let serverFwds = takeRelation maxBatchSize (do f <- fwds; where f.toServer == server; yield f)
880+ let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))
870881 forEach serverFwds (\f -> f.messageId) (\f -> do
871882 t2 <- now
872883 auth <- generateFederationAuth ("forward:" ++ f.messageId) t2
@@ -888,7 +899,7 @@ retryPendingForwards = do
888899
889900sendAcksAndClean = \pendingAcks ->
890901 forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do
891− let mids = do a <- pendingAcks; where a.originServer == server; yield a.messageId
902+ let mids = pluck (filter pendingAcks (\a -> a.originServer == server)) (\a -> a.messageId)
892903 t <- now
893904 auth <- generateFederationAuth "ack" t
894905 result <- fetch (federationUrl server) (FederationAck {
@@ -901,20 +912,16 @@ sendAcksAndClean = \pendingAcks ->
901912 clearFailure server
902913 atomic do
903914 cur <- *ackedIds
904− set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a)
915+ set *ackedIds = filter cur (\a -> not (a.originServer == server && elem mids a.messageId)))
905916
906917-- Forward queued messages to a server when a remote key comes online.
907918forwardQueuedToServer = \pubkey server t -> atomic do
908919 msgs <- *messages
909920 fwds <- *forwards
910− let candidates = do
911− m <- msgs
912− where m.toKey == pubkey
913− where none fwds (\f -> f.messageId == m.messageId && f.toServer == server)
914− let origin = if m.originServer == "" then serverHostN else m.originServer
915− yield {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: origin, retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey}
916− let capped = takeRelation maxForwardsPerServer candidates
917− set *forwards = capList (union fwds capped) maxForwardsEntries (\f -> f.createdAt)
921+ let mine = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server))
922+ let candidates = pluck mine (\m ->
923+ {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: (if m.originServer == "" then serverHostN else m.originServer), retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey})
924+ set *forwards = capList (union fwds (takeRelation maxForwardsPerServer candidates)) maxForwardsEntries (\f -> f.createdAt)
918925
919926-- ============================================
920927-- Background loops
@@ -927,7 +934,7 @@ pruneAndRetry = \t -> do
927934 when federationEnabled (fork retryPendingForwards)
928935 when federationEnabled (do
929936 aids <- *ackedIds
930− let pendingAcks = do a <- aids; where not (isLocalServer a.originServer); yield a
937+ let pendingAcks = filter aids (\a -> not (isLocalServer a.originServer))
931938 fork (sendAcksAndClean pendingAcks))
932939 -- Refresh local presence for active sessions every ~5 minutes
933940 lastRefresh <- *lastPresenceRefresh
@@ -940,18 +947,15 @@ refreshLocalPresence = \t -> do
940947 refreshState <- atomic do
941948 sessions <- *sessions
942949 presence <- *presence
943− let activeKeys = unique (do s <- sessions; yield s.pubkey)
944− let updated = do
945− p <- presence
946− yield (if isLocalServer p.server && any activeKeys (\k -> k == p.pubkey) && p.expiresAt - t < sessionExpiry / 2
947− then {pubkey: p.pubkey, server: p.server, expiresAt: t + sessionExpiry, timestamp: t}
948− else p)
950+ let activeKeys = unique (pluck sessions (\s -> s.pubkey))
951+ let updated = pluck presence (\p ->
952+ if isLocalServer p.server && elem activeKeys p.pubkey && p.expiresAt - t < sessionExpiry / 2
953+ then {pubkey: p.pubkey, server: p.server, expiresAt: t + sessionExpiry, timestamp: t}
954+ else p)
949955 set *presence = updated
950− let refreshed = do p <- updated; where isLocalServer p.server && p.expiresAt == t + sessionExpiry; yield p
951− let peers = knownPeerServersFromPresence updated
952− yield {refreshed: refreshed, peers: peers}
956+ yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
953957 when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do
954− let events = do p <- refreshState.refreshed; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]}
958+ let events = pluck refreshState.refreshed (\p -> {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]})
955959 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))
956960
957961-- ============================================
@@ -972,9 +976,8 @@ handleChallenge = \pubkey clientIp respond -> do
972976 challenge <- genToken
973977 result <- atomic do
974978 ch <- *challenges
975− let active = do c <- ch; where c.expiresAt > t; yield c
976− let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)
977− let cappedActive = if activeForKey >= maxChallengesPerKey
979+ let active = filter ch (\c -> c.expiresAt > t)
980+ let cappedActive = if countWhere active (\c -> c.pubkey == pubkey) >= maxChallengesPerKey
978981 then evictOldest active (\c -> if c.pubkey == pubkey then c.expiresAt else 999999999999)
979982 else active
980983 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
@@ -1006,26 +1009,23 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10061009 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
10071010 then yield {valid: False {}, peers: []}
10081011 else do
1009− set *challenges = do c <- ch; where c.pubkey != pubkey || c.challenge != challenge; yield c
1012+ set *challenges = filter ch (\c -> c.pubkey != pubkey || c.challenge != challenge)
10101013 dp <- *drainPhase
10111014 if dp >= 1 then yield {valid: False {}, peers: []}
10121015 else do
10131016 ses <- *sessions
1014− let kept = if revokeOthers then (do s <- ses; where s.pubkey != pubkey; yield s) else ses
1015− let perKeyOver = count (do s <- kept; where s.pubkey == pubkey; yield s) >= maxSessionsPerKey
1016− let afterPerKey = if perKeyOver
1017− then let forKey = do s <- kept; where s.pubkey == pubkey; yield s in
1018− let others = do s <- kept; where s.pubkey != pubkey; yield s in
1019− union others (evictOldest forKey (\s -> s.expiresAt))
1017+ let kept = if revokeOthers then filter ses (\s -> s.pubkey != pubkey) else ses
1018+ let forKey = filter kept (\s -> s.pubkey == pubkey)
1019+ let afterPerKey = if count forKey >= maxSessionsPerKey
1020+ then union (filter kept (\s -> s.pubkey != pubkey)) (evictOldest forKey (\s -> s.expiresAt))
10201021 else kept
10211022 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
10221023 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
10231024 pres <- *presence
1024− let withoutLocal = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
1025+ let withoutLocal = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
10251026 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
10261027 set *presence = newPres
1027− let peers = knownPeerServersFromPresence newPres
1028− yield {valid: True {}, peers: peers}
1028+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10291029 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
10301030 else do
10311031 when federationEnabled
@@ -1044,7 +1044,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
10441044 let found = any ses (\s -> s.pubkey == caller && s.token == targetToken)
10451045 if not found then yield False
10461046 else do
1047− set *sessions = do s <- ses; where s.pubkey != caller || s.token != targetToken; yield s
1047+ set *sessions = filter ses (\s -> s.pubkey != caller || s.token != targetToken)
10481048 yield True
10491049 yield (respond {ok: result, authorized: True {}}))
10501050
@@ -1060,14 +1060,14 @@ handleLogout = \authorization clientIp respond -> do
10601060 Authorized {pubkey, ip} ->
10611061 if not (matchIp clientIp ip) then yield {ok: False {}, pubkey: "", peers: []}
10621062 else do
1063− set *sessions = do s <- ses; where s.token != token; yield s
1063+ set *sessions = filter ses (\s -> s.token != token)
10641064 -- If no remaining sessions for this key, drop local presence
10651065 after <- *sessions
10661066 let stillHas = any after (\s -> s.pubkey == pubkey)
10671067 peers <- if stillHas then yield []
10681068 else do
10691069 pres <- *presence
1070− set *presence = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
1070+ set *presence = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
10711071 yield (knownPeerServersFromPresence pres)
10721072 yield {ok: True {}, pubkey: pubkey, peers: peers}
10731073 when (removed.ok && removed.pubkey != "" && count removed.peers > 0 && federationEnabled)
@@ -1111,7 +1111,7 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11111111 Just {value: s} -> do
11121112 acquired <- atomic do
11131113 aps <- *activePolls
1114− let perKey = count (do p <- aps; where p.pubkey == s.pubkey; yield p)
1114+ let perKey = countWhere aps (\p -> p.pubkey == s.pubkey)
11151115 if perKey >= maxPollsPerKey then yield False
11161116 else do
11171117 set *activePolls = capList (union aps [{pubkey: s.pubkey, startedAt: t}]) maxActivePolls (\p -> p.startedAt)
@@ -1127,18 +1127,18 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11271127 then atomic do
11281128 t2 <- now
11291129 pres <- *presence
1130− yield (do k <- watchKeys; yield {pubkey: k, online: count (presenceForKey pres k t2) > 0})
1130+ yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))
11311131 else yield []
11321132 drains <- atomic do
11331133 se <- *senderExpirations
1134− let myExp = do x <- se; where x.senderKey == s.pubkey; yield {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt}
1135− set *senderExpirations = do x <- se; where x.senderKey != s.pubkey; yield x
1134+ let myExp = pluck (filter se (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1135+ set *senderExpirations = filter se (\x -> x.senderKey != s.pubkey)
11361136 sd <- *senderDeliveries
1137− let myDel = do x <- sd; where x.senderKey == s.pubkey; yield {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt}
1138− set *senderDeliveries = do x <- sd; where x.senderKey != s.pubkey; yield x
1137+ let myDel = pluck (filter sd (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1138+ set *senderDeliveries = filter sd (\x -> x.senderKey != s.pubkey)
11391139 cn <- *cancelNotifications
1140− let myCancels = do x <- cn; where x.toKey == s.pubkey; yield {messageId: x.messageId, cancelledAt: x.cancelledAt}
1141− set *cancelNotifications = do x <- cn; where x.toKey != s.pubkey; yield x
1140+ let myCancels = pluck (filter cn (\x -> x.toKey == s.pubkey)) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1141+ set *cancelNotifications = filter cn (\x -> x.toKey != s.pubkey)
11421142 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
11431143 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
11441144 yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: pollResult.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})
@@ -1160,27 +1160,26 @@ doPoll = \s t -> do
11601160 yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
11611161
11621162collectPollEvents = \messages pubkey cursor t ->
1163− let myMsgs = do m <- messages; where m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor; yield m in
1164− let sorted = sortBy (\m -> m.receivedAt) myMsgs in
1165− let limited = takeRelation maxPollEvents sorted in
1166− let evts = do m <- limited; yield {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}} in
1167− let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) cursor limited in
1168− {events: evts, cursor: lastReceivedAt, hasMore: count myMsgs > count limited}
1163+ let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
1164+ let limited = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs) in
1165+ let evts = pluck limited (\m -> {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}}) in
1166+ {events: evts, cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}
11691167
11701168ackPolledMessages = \pubkey cursor t -> do
11711169 msgs <- *messages
1172− let toAck = do m <- msgs; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
1170+ let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
1171+ let toAck = filter msgs matches
11731172 if count toAck == 0 then yield {pendingAcks: []}
11741173 else do
1175− set *messages = do m <- msgs; where not (m.toKey == pubkey && m.receivedAt <= cursor); yield m
1174+ set *messages = filter msgs (\m -> not (matches m))
11761175 fwds <- *forwards
1177− let mids = do m <- toAck; yield m.messageId
1178− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any mids (\mid -> mid == f.messageId)); yield f
1179− let ackedFederated = do m <- toAck; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
1176+ let mids = pluck toAck (\m -> m.messageId)
1177+ set *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1178+ let ackedFederated = pluck (filter toAck (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
11801179 aids <- *ackedIds
11811180 set *ackedIds = capList (union aids ackedFederated) maxAckedIds (\a -> a.ackedAt)
11821181 -- Sender delivery receipts for locally-originated messages
1183− let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
1182+ let localDelivered = pluck (filter toAck (\m -> m.senderKey != "" && m.originServer == "")) (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
11841183 sd <- *senderDeliveries
11851184 set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)
11861185 yield {pendingAcks: ackedFederated}
@@ -1230,7 +1229,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12301229 recvRates <- *recvRates
12311230 let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t
12321231 set *recvRates = rr.entries
1233− let recipientCount = count (do m <- msgs; where m.toKey == to; yield m)
1232+ let recipientCount = countWhere msgs (\m -> m.toKey == to)
12341233 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}
12351234 else if isDup then yield {status: MsDuplicate {}, remoteServers: []}
12361235 else if sr.limited || pr.limited || rr.limited then yield {status: MsRateLimited {}, remoteServers: []}
@@ -1250,7 +1249,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12501249 forwardsEnqueued <- if federationEnabled && count remoteServers > 0
12511250 then do
12521251 fwds <- *forwards
1253− let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}
1252+ let newFwds = pluck (filter remoteServers (\s -> none fwds (\f -> f.messageId == effMsgId && f.toServer == s))) (\s -> {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey})
12541253 set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
12551254 yield (count newFwds > 0)
12561255 else yield False
@@ -1263,7 +1262,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12631262 forEach servers (\s -> s) (\server -> do
12641263 info <- atomic do
12651264 fwds <- *forwards
1266− case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of
1265+ case headOf (filter fwds (\f -> f.messageId == effMsgId && f.toServer == server)) of
12671266 Nothing {} -> yield Nothing {}
12681267 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
12691268 case info of
@@ -1284,7 +1283,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12841283 clearFailure server
12851284 atomic do
12861285 fwds <- *forwards
1287− set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f
1286+ set *forwards = filter fwds (\f -> not (f.messageId == effMsgId && f.toServer == server))
12881287 else yield {})
12891288
12901289-- ============================================
@@ -1293,54 +1292,45 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12931292
12941293handleLookupPresence = \pubkey authorization clientIp respond -> do
12951294 t <- now
1296− g <- checkGlobalRate t
1297− if g then yield (respond {server: "", authorized: False {}, error: "rate_limited"})
1298− else withSessionAuth authorization t clientIp
1299− (yield (respond {server: "", authorized: False {}, error: ""}))
1300− (\senderKey -> do
1301− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
1302− if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})
1303− else do
1304− pres <- *presence
1305− let found = bestServerForKey (presenceForKey pres pubkey t)
1306− let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1307− yield (respond {server: masked, authorized: True {}, error: ""}))
1295+ let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1296+ gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do
1297+ lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
1298+ if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})
1299+ else do
1300+ pres <- *presence
1301+ let found = bestServerForKey (presenceForKey pres pubkey t)
1302+ let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1303+ yield (respond {server: masked, authorized: True {}, error: ""}))
13081304
13091305handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
13101306 t <- now
1311− g <- checkGlobalRate t
1312− if g then yield (respond {results: [], authorized: False {}, error: "rate_limited"})
1313− else withSessionAuth authorization t clientIp
1314− (yield (respond {results: [], authorized: False {}, error: ""}))
1315− (\senderKey -> do
1316− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
1317− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1318− else do
1319− let capped = takeRelation 50 pubkeys
1320− pres <- *presence
1321− let results = do
1322− k <- capped
1323− let found = bestServerForKey (presenceForKey pres k t)
1324− let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1325− yield {pubkey: k, server: masked}
1326− yield (respond {results: results, authorized: True {}, error: ""}))
1307+ let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}
1308+ gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do
1309+ lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t
1310+ if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1311+ else do
1312+ pres <- *presence
1313+ let results = do
1314+ k <- takeRelation 50 pubkeys
1315+ let found = bestServerForKey (presenceForKey pres k t)
1316+ let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1317+ yield {pubkey: k, server: masked}
1318+ yield (respond {results: results, authorized: True {}, error: ""}))
13271319
13281320handleMessageStatus = \to messageId authorization clientIp respond -> do
13291321 t <- now
1330− g <- checkGlobalRate t
1331− if g then yield (respond {status: "", authorized: False {}})
1332− else withSessionAuth authorization t clientIp
1333− (yield (respond {status: "", authorized: False {}}))
1334− (\_ -> do
1335− msgs <- *messages
1336− fwds <- *forwards
1337− aids <- *ackedIds
1338− dids <- *deliveredIds
1339− let s = if any msgs (\m -> m.messageId == messageId && m.toKey == to) then "pending"
1340− else if any fwds (\f -> f.messageId == messageId && f.toKey == to) then "forwarding"
1341− else if hasMessage messageId to aids || hasMessage messageId to dids then "delivered"
1342− else "unknown"
1343− yield (respond {status: s, authorized: True {}}))
1322+ let mkErr = \_ -> {status: "", authorized: False {}}
1323+ gateAuthNoDrain authorization clientIp t respond mkErr (\_ -> do
1324+ msgs <- *messages
1325+ fwds <- *forwards
1326+ aids <- *ackedIds
1327+ dids <- *deliveredIds
1328+ let matches = \x -> x.messageId == messageId && x.toKey == to
1329+ let s = if any msgs matches then "pending"
1330+ else if any fwds matches then "forwarding"
1331+ else if any aids matches || any dids matches then "delivered"
1332+ else "unknown"
1333+ yield (respond {status: s, authorized: True {}}))
13441334
13451335-- ============================================
13461336-- Handler: ack / list / delete / cancel
@@ -1348,125 +1338,99 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do
13481338
13491339handleAckMessages = \messageIds authorization clientIp respond -> do
13501340 t <- now
1351− dp <- *drainPhase
1352− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "draining"})
1353− else do
1354− g <- checkGlobalRate t
1355− if g then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "rate_limited"})
1356− else withSessionAuth authorization t clientIp
1357− (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "unauthorized"}))
1358− (\pubkey -> do
1359− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
1360− if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
1341+ let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
1342+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1343+ lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
1344+ if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})
1345+ else do
1346+ let dedupedIds = dedupMessageIds messageIds
1347+ let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1348+ result <- atomic do
1349+ msgs <- *messages
1350+ let toAck = filter msgs matches
1351+ if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}
13611352 else do
1362− let dedupedIds = dedupMessageIds messageIds
1363− result <- atomic do
1364− msgs <- *messages
1365− let toAck = do m <- msgs; where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid); yield m
1366− if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}
1367− else do
1368− set *messages = do m <- msgs; where not (m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)); yield m
1369− fwds <- *forwards
1370− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any dedupedIds (\mid -> f.messageId == mid)); yield f
1371− let ackedFed = do m <- toAck; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
1372− aids <- *ackedIds
1373− set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1374− let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
1375− sd <- *senderDeliveries
1376− set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)
1377− yield {ackedCount: count toAck, pendingAcks: ackedFed}
1378− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1379− yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
1353+ set *messages = filter msgs (\m -> not (matches m))
1354+ fwds <- *forwards
1355+ set *forwards = filter fwds (\f -> not (matches f))
1356+ let ackedFed = pluck (filter toAck (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
1357+ aids <- *ackedIds
1358+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1359+ let localDelivered = pluck (filter toAck (\m -> m.senderKey != "" && m.originServer == "")) (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
1360+ sd <- *senderDeliveries
1361+ set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)
1362+ yield {ackedCount: count toAck, pendingAcks: ackedFed}
1363+ fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1364+ yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
13801365
13811366handleListMessages = \offset authorization clientIp respond -> do
13821367 t <- now
1383− dp <- *drainPhase
1384− if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0, error: ""})
1385− else do
1386− g <- checkGlobalRate t
1387− if g then yield (respond {messages: [], authorized: False {}, count: 0, error: "rate_limited"})
1388− else withSessionAuth authorization t clientIp
1389− (yield (respond {messages: [], authorized: False {}, count: 0, error: ""}))
1390− (\pubkey -> do
1391− lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t
1392− if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1393− else do
1394− msgs <- *messages
1395− let myMsgs = sortBy (\m -> m.receivedAt) (do m <- msgs; where m.toKey == pubkey && m.expiresAt > t; yield m)
1396− let total = count myMsgs
1397− let effectiveOffset = clamp 0 total offset
1398− -- Index using a fold so we can drop items by position; simple O(n).
1399− let after = (fold (\acc m -> if acc.i < effectiveOffset then {i: acc.i + 1, out: acc.out} else {i: acc.i, out: union acc.out [m]}) {i: 0, out: []} myMsgs).out
1400− let capped = takeRelation maxPollEvents after
1401− let result = do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}
1402− yield (respond {messages: result, authorized: True {}, count: total, error: ""}))
1368+ let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}
1369+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1370+ lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t
1371+ if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1372+ else do
1373+ msgs <- *messages
1374+ let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1375+ let total = count myMsgs
1376+ let effectiveOffset = clamp 0 total offset
1377+ -- Drop the first `effectiveOffset` items via fold (O(n)).
1378+ let after = (fold (\acc m -> if acc.i < effectiveOffset then {i: acc.i + 1, out: acc.out} else {i: acc.i, out: union acc.out [m]}) {i: 0, out: []} myMsgs).out
1379+ let result = pluck (takeRelation maxPollEvents after) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
1380+ yield (respond {messages: result, authorized: True {}, count: total, error: ""}))
14031381
14041382handleDeleteMessages = \messageIds authorization clientIp respond -> do
14051383 t <- now
1406− dp <- *drainPhase
1407− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: ""})
1408− else do
1409− g <- checkGlobalRate t
1410− if g then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "rate_limited"})
1411− else withSessionAuth authorization t clientIp
1412− (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "unauthorized"}))
1413− (\pubkey -> do
1414− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
1415− if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
1384+ let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}
1385+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do
1386+ lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t
1387+ if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})
1388+ else do
1389+ let dedupedIds = dedupMessageIds messageIds
1390+ let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId
1391+ result <- atomic do
1392+ msgs <- *messages
1393+ let toDel = filter msgs matches
1394+ if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}
14161395 else do
1417− let dedupedIds = dedupMessageIds messageIds
1418− result <- atomic do
1419− msgs <- *messages
1420− let toDel = do m <- msgs; where m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId); yield m
1421− if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}
1422− else do
1423− set *messages = do m <- msgs; where not (m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId)); yield m
1424− fwds <- *forwards
1425− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any dedupedIds (\mid -> mid == f.messageId)); yield f
1426− let federatedDeleted = do m <- toDel; where m.originServer != ""; yield m
1427− let ackedFed = do m <- federatedDeleted; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
1428− aids <- *ackedIds
1429− set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1430− yield {deletedCount: count toDel, pendingAcks: ackedFed}
1431− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1432− yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
1396+ set *messages = filter msgs (\m -> not (matches m))
1397+ fwds <- *forwards
1398+ set *forwards = filter fwds (\f -> not (matches f))
1399+ let ackedFed = pluck (filter toDel (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
1400+ aids <- *ackedIds
1401+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1402+ yield {deletedCount: count toDel, pendingAcks: ackedFed}
1403+ fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1404+ yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
14331405
14341406handleCancelMessage = \messageId to authorization clientIp respond -> do
14351407 t <- now
1436− dp <- *drainPhase
1437− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: ""})
1438− else do
1439− g <- checkGlobalRate t
1440− if g then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "rate_limited"})
1441− else withSessionAuth authorization t clientIp
1442− (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "unauthorized"}))
1443− (\senderKey -> do
1444− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1445− if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})
1446− else do
1447− result <- atomic do
1448− msgs <- *messages
1449− let target = headOf (do m <- msgs; where m.messageId == messageId && m.toKey == to; yield m)
1450− case target of
1451− Nothing {} -> yield {cancelled: False {}, cancelServers: []}
1452− Just {value: msg} ->
1453− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1454− then yield {cancelled: False {}, cancelServers: []}
1455− else do
1456− set *messages = do m <- msgs; where not (m.messageId == messageId && m.toKey == to); yield m
1457− fwds <- *forwards
1458− let cancelledFwds = do f <- fwds; where f.messageId == messageId && f.toKey == to; yield f
1459− let fwdServers = collectPeers cancelledFwds (\f -> f.toServer)
1460− set *forwards = do f <- fwds; where not (f.messageId == messageId && f.toKey == to); yield f
1461− pres <- *presence
1462− let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)
1463− let allServers = unique (union presServers fwdServers)
1464− cn <- *cancelNotifications
1465− set *cancelNotifications = capList (union cn [{toKey: to, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1466− yield {cancelled: True {}, cancelServers: allServers}
1467− when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1468− (fork (sendCancelToEach messageId to result.cancelServers))
1469− yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1408+ let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}
1409+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1410+ lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1411+ if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})
1412+ else do
1413+ let matches = \x -> x.messageId == messageId && x.toKey == to
1414+ result <- atomic do
1415+ msgs <- *messages
1416+ case headOf (filter msgs matches) of
1417+ Nothing {} -> yield {cancelled: False {}, cancelServers: []}
1418+ Just {value: msg} ->
1419+ if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1420+ then yield {cancelled: False {}, cancelServers: []}
1421+ else do
1422+ set *messages = filter msgs (\m -> not (matches m))
1423+ fwds <- *forwards
1424+ let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)
1425+ set *forwards = filter fwds (\f -> not (matches f))
1426+ pres <- *presence
1427+ let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)
1428+ cn <- *cancelNotifications
1429+ set *cancelNotifications = capList (union cn [{toKey: to, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1430+ yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
1431+ when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1432+ (fork (sendCancelToEach messageId to result.cancelServers))
1433+ yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
14701434
14711435sendCancelToEach = \messageId to servers ->
14721436 forEach servers (\s -> s) (\server -> do
@@ -1483,57 +1447,45 @@ sendCancelToEach = \messageId to servers ->
14831447
14841448handleCancelBatch = \cancels authorization clientIp respond -> do
14851449 t <- now
1486− dp <- *drainPhase
1487− if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})
1488− else do
1489− g <- checkGlobalRate t
1490− if g then yield (respond {results: [], authorized: False {}, error: "rate_limited"})
1491− else withSessionAuth authorization t clientIp
1492− (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))
1493− (\senderKey -> do
1494− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1495− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1496− else do
1497− results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1498− r <- atomic do
1499− msgs <- *messages
1500− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)
1501− case target of
1502− Nothing {} -> yield False
1503− Just {value: msg} ->
1504− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey
1505− then yield False
1506− else do
1507− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m
1508− fwds <- *forwards
1509− set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.to); yield f
1510− cn <- *cancelNotifications
1511− set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1512− yield True
1513− yield {messageId: e.messageId, cancelled: r})
1514− yield (respond {results: results, authorized: True {}, error: ""}))
1450+ let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}
1451+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1452+ lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t
1453+ if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})
1454+ else do
1455+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1456+ let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
1457+ r <- atomic do
1458+ msgs <- *messages
1459+ case headOf (filter msgs matches) of
1460+ Nothing {} -> yield False
1461+ Just {value: msg} ->
1462+ if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1463+ then yield False
1464+ else do
1465+ set *messages = filter msgs (\m -> not (matches m))
1466+ fwds <- *forwards
1467+ set *forwards = filter fwds (\f -> not (matches f))
1468+ cn <- *cancelNotifications
1469+ set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1470+ yield True
1471+ yield {messageId: e.messageId, cancelled: r})
1472+ yield (respond {results: results, authorized: True {}, error: ""}))
15151473
15161474handleSendBatch = \messages authorization clientIp respond -> do
15171475 t <- now
1518− dp <- *drainPhase
1519− if dp >= 2 then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})
1520− else do
1521− g <- checkGlobalRate t
1522− if g then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})
1523− else withSessionAuth authorization t clientIp
1524− (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))
1525− (\senderKey -> do
1526− results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do
1527− effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1528− let clampedTs = clampTimestamp m.timestamp t
1529− let clampedPri = clamp 0 maxMessagePriority m.priority
1530− r <- if t - m.timestamp > messageMaxAge
1531− then yield {status: MsRejected {}, remoteServers: []}
1532− else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1533− when (federationEnabled && count r.remoteServers > 0)
1534− (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1535− yield {messageId: effMsgId, status: r.status})
1536− yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))
1476+ let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1477+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1478+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do
1479+ effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1480+ let clampedTs = clampTimestamp m.timestamp t
1481+ let clampedPri = clamp 0 maxMessagePriority m.priority
1482+ r <- if t - m.timestamp > messageMaxAge
1483+ then yield {status: MsRejected {}, remoteServers: []}
1484+ else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t
1485+ when (federationEnabled && count r.remoteServers > 0)
1486+ (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))
1487+ yield {messageId: effMsgId, status: r.status})
1488+ yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))
15371489
15381490-- ============================================
15391491-- Handler: federation receive
@@ -1551,9 +1503,9 @@ handleVerifyToken = \token scope clientIp respond -> do
15511503 else do
15521504 found <- atomic do
15531505 pending <- *pendingVerifications
1554− let candidates = do p <- pending; where p.scope == scope && t - p.createdAt < verificationTokenExpiry; yield p
1506+ let candidates = filter pending (\p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry)
15551507 let match = any candidates (\p -> p.token == token)
1556− when match (set *pendingVerifications = do p <- pending; where p.scope != scope || p.token != token; yield p)
1508+ when match (set *pendingVerifications = filter pending (\p -> p.scope != scope || p.token != token))
15571509 yield match
15581510 yield (respond {ok: found})
15591511
@@ -1572,19 +1524,13 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
15721524 let expiresAt = min e.timestamp t + clampTtl e.ttl * 1000
15731525 where expiresAt > t
15741526 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
1575− let online = do v <- validEvents; where (case v.eventType of GossipOnline {} -> True {}; _ -> False {}); yield {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp}
1576− let offline = do v <- validEvents; where (case v.eventType of GossipOffline {} -> True {}; _ -> False {}); yield {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp}
1577− let withoutOffline = do
1578− p <- pres
1579− where none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp)
1580− yield p
1581− let merged = mergePresence withoutOffline online
1582− set *presence = capList merged maxPresenceEntries (\p -> p.expiresAt)
1583− let newlyOnline = do
1584− e <- online
1585− where not (isLocalServer e.server)
1586− where none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)
1587− yield {pubkey: e.pubkey, server: e.server}
1527+ let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
1528+ let isOffline = \v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}
1529+ let online = pluck (filter validEvents isOnline) (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
1530+ let offline = pluck (filter validEvents isOffline) (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
1531+ let withoutOffline = filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))
1532+ set *presence = capList (mergePresence withoutOffline online) maxPresenceEntries (\p -> p.expiresAt)
1533+ let newlyOnline = pluck (filter online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt))) (\e -> {pubkey: e.pubkey, server: e.server})
15881534 yield {newlyOnline: newlyOnline}
15891535 when (count valid.newlyOnline > 0)
15901536 (fork (forwardForNewlyOnline valid.newlyOnline t))
@@ -1614,8 +1560,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
16141560 if isDup then yield "duplicate"
16151561 else if count msgs >= maxMessages then yield "global_full"
16161562 else
1617− let rcCount = count (do m <- msgs; where m.toKey == toKey; yield m) in
1618− if rcCount >= maxMessagesPerRecipient then yield "mailbox_full"
1563+ if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
16191564 else do
16201565 rr <- *recvRates
16211566 let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t
@@ -1641,22 +1586,21 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
16411586 let dedupedIds = dedupMessageIds messageIds
16421587 result <- atomic do
16431588 fwds <- *forwards
1644− let removed = do f <- fwds; where f.toServer == fromServer && any dedupedIds (\m -> m == f.messageId); yield f
1645− set *forwards = do f <- fwds; where not (f.toServer == fromServer && any dedupedIds (\m -> m == f.messageId)); yield f
1589+ let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
1590+ let removed = filter fwds isAcked
1591+ set *forwards = filter fwds (\f -> not (isAcked f))
16461592 -- Reclaim origin messages with no remaining forwards
16471593 msgs <- *messages
1648− let removedIds = do f <- removed; yield f.messageId
1594+ let removedIds = pluck removed (\f -> f.messageId)
16491595 pres <- *presence
16501596 remainingFwds <- *forwards
1651− let toReap = do
1652− m <- msgs
1653− where any removedIds (\mid -> mid == m.messageId)
1654− where t - m.receivedAt >= ackGracePeriod
1655− where none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1656− where none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey)
1657− yield m
1658− set *messages = do m <- msgs; where none toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey); yield m
1659− let perResult = do mid <- dedupedIds; yield {messageId: mid, found: any removed (\f -> f.messageId == mid)}
1597+ let toReap = filter msgs (\m ->
1598+ elem removedIds m.messageId
1599+ && t - m.receivedAt >= ackGracePeriod
1600+ && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1601+ && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1602+ set *messages = filter msgs (\m -> none toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
1603+ let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})
16601604 yield {ackedCount: count removed, results: perResult}
16611605 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
16621606
@@ -1664,15 +1608,15 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
16641608 t <- now
16651609 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}
16661610 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (do
1611+ let matches = \x -> x.messageId == messageId && x.toKey == toKey
16671612 cancelled <- atomic do
16681613 msgs <- *messages
1669− let target = headOf (do m <- msgs; where m.messageId == messageId && m.toKey == toKey && m.originServer == fromServer; yield m)
1670− case target of
1614+ case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of
16711615 Nothing {} -> yield False
16721616 Just {} -> do
1673− set *messages = do m <- msgs; where not (m.messageId == messageId && m.toKey == toKey); yield m
1617+ set *messages = filter msgs (\m -> not (matches m))
16741618 fwds <- *forwards
1675− set *forwards = do f <- fwds; where not (f.messageId == messageId && f.toKey == toKey); yield f
1619+ set *forwards = filter fwds (\f -> not (matches f))
16761620 cn <- *cancelNotifications
16771621 set *cancelNotifications = capList (union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
16781622 yield True
@@ -1683,15 +1627,15 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
16831627 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
16841628 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
16851629 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1630+ let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
16861631 cancelled <- atomic do
16871632 msgs <- *messages
1688− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.toKey && m.originServer == fromServer; yield m)
1689− case target of
1633+ case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of
16901634 Nothing {} -> yield False
16911635 Just {} -> do
1692− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.toKey); yield m
1636+ set *messages = filter msgs (\m -> not (matches m))
16931637 fwds <- *forwards
1694− set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.toKey); yield f
1638+ set *forwards = filter fwds (\f -> not (matches f))
16951639 yield True
16961640 yield {messageId: e.messageId, cancelled: cancelled})
16971641 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
@@ -1716,8 +1660,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
17161660 if isDup then yield True
17171661 else if count msgs >= maxMessages then yield False
17181662 else
1719− let rcCount = count (do m <- msgs; where m.toKey == e.toKey; yield m) in
1720− if rcCount >= maxMessagesPerRecipient then yield False
1663+ if countWhere msgs (\m -> m.toKey == e.toKey) >= maxMessagesPerRecipient then yield False
17211664 else do
17221665 seq <- *seqCounter
17231666 let seqTs = nextSeqTs t seq
@@ -1750,7 +1693,7 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
17501693 set *senderDeliveries = capList (union sd receipts) 10000 (\d -> d.deliveredAt)
17511694 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
17521695
1753−-- Pair up parallel lists by index. Builds index arrays and joins on i (O(n*m) but trivial code).
1696+-- Pair up parallel lists by index (O(n*m) join on i; trivial code).
17541697withIndex = \xs -> (fold (\acc x -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: x}]}) {i: 0, out: []} xs).out
17551698zipPairs = \mids keys ->
17561699 do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}
@@ -1770,13 +1713,13 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
17701713 if effectivePhase == 1 then yield {drainedCount: 0, peers: [], localKeys: []}
17711714 else do
17721715 pres <- *presence
1773− let localEntries = do p <- pres; where isLocalServer p.server && p.expiresAt > t; yield p
1774− set *presence = do p <- pres; where not (isLocalServer p.server) || p.expiresAt <= t; yield p
1716+ let localEntries = filter pres (\p -> isLocalServer p.server && p.expiresAt > t)
1717+ set *presence = filter pres (\p -> not (isLocalServer p.server) || p.expiresAt <= t)
17751718 set *pollWakeup = t
17761719 let peers = knownPeerServersFromPresence pres
17771720 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}
17781721 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)
1779− (fork (sendBatchedGossip drainResult.peers (do p <- drainResult.localKeys; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}) gossipOriginHops))
1722+ (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops))
17801723 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
17811724
17821725handleUndrain = \authorization respond -> withMaintAuth authorization
@@ -1860,7 +1803,7 @@ handleAdminListSessions = \pubkey authorization respond -> withMaintAuth authori
18601803 (yield (respond {sessions: [], count: 0, ok: False {}}))
18611804 (do
18621805 ses <- *sessions
1863− let mine = do s <- ses; where s.pubkey == pubkey; yield {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}
1806+ let mine = pluck (filter ses (\s -> s.pubkey == pubkey)) (\s -> {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt})
18641807 yield (respond {sessions: mine, count: count mine, ok: True {}}))
18651808
18661809handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization
@@ -1868,10 +1811,9 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
18681811 (do
18691812 revoked <- atomic do
18701813 ses <- *sessions
1871− let mine = do s <- ses; where s.pubkey == pubkey; yield s
18721814 pruneRef sessions (\s -> s.pubkey != pubkey)
18731815 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1874− yield (count mine)
1816+ yield (countWhere ses (\s -> s.pubkey == pubkey))
18751817 yield (respond {ok: True {}, revokedCount: revoked}))
18761818
18771819handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization
@@ -1880,10 +1822,10 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
18801822 t <- now
18811823 purged <- atomic do
18821824 msgs <- *messages
1883− let mine = do m <- msgs; where m.toKey == pubkey; yield m
1825+ let mine = filter msgs (\m -> m.toKey == pubkey)
18841826 pruneRef messages (\m -> m.toKey != pubkey)
18851827 pruneRef forwards (\f -> f.toKey != pubkey)
1886− let federated = do m <- mine; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
1828+ let federated = pluck (filter mine (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
18871829 aids <- *ackedIds
18881830 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)
18891831 yield (count mine)
@@ -1895,10 +1837,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
18951837 else do
18961838 result <- atomic do
18971839 ses <- *sessions
1898− let mine = do s <- ses; where s.ip == ip; yield s
1840+ let mine = filter ses (\s -> s.ip == ip)
18991841 pruneRef sessions (\s -> s.ip != ip)
1900− let keys = unique (do s <- mine; yield s.pubkey)
1901− yield {revoked: count mine, keys: count keys}
1842+ yield {revoked: count mine, keys: count (unique (pluck mine (\s -> s.pubkey)))}
19021843 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
19031844
19041845handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
@@ -1906,14 +1847,13 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
19061847 (do
19071848 t <- now
19081849 ses <- *sessions
1909− let active = do s <- ses; where s.expiresAt > t; yield s
1910− let grouped = do
1911− pk <- unique (do s <- active; yield s.pubkey)
1912− let mine = do s <- active; where s.pubkey == pk; yield s
1913− yield {pubkey: pk,
1914− count: count mine,
1915− ips: unique (do s <- mine; where s.ip != ""; yield s.ip),
1916− latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine}
1850+ let active = filter ses (\s -> s.expiresAt > t)
1851+ let grouped = pluck (unique (pluck active (\s -> s.pubkey))) (\pk ->
1852+ let mine = filter active (\s -> s.pubkey == pk) in
1853+ {pubkey: pk,
1854+ count: count mine,
1855+ ips: unique (pluck (filter mine (\s -> s.ip != "")) (\s -> s.ip)),
1856+ latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
19171857 let sorted = sortBy (\g -> 0 - g.count) grouped
19181858 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit
19191859 yield (respond {sessions: takeRelation effLimit sorted, ok: True {}, total: count grouped}))
@@ -1979,10 +1919,8 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization
19791919 t <- now
19801920 failures <- *forwardFailures
19811921 fwds <- *forwards
1982− let peers = do
1983− f <- failures
1984− let pending = count (do x <- fwds; where x.toServer == f.server; yield x)
1985− yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: pending, avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0}
1922+ let peers = pluck failures (\f ->
1923+ {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: countWhere fwds (\x -> x.toServer == f.server), avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0})
19861924 yield (respond {peers: peers, ok: True {}}))
19871925
19881926handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
@@ -1994,22 +1932,21 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
19941932 ses <- *sessions
19951933 fwds <- *forwards
19961934 aps <- *activePolls
1997− let m = count (do x <- msgs; where x.toKey == pubkey; yield x)
1998− let p = do x <- pres; where x.pubkey == pubkey && x.expiresAt > t; yield {server: x.server, expiresAt: x.expiresAt}
1999− let s = count (do x <- ses; where x.pubkey == pubkey; yield x)
2000− let myFwds = do x <- fwds; where x.toKey == pubkey; yield x
2001− let f = count myFwds
2002− let fd = do x <- myFwds; yield {server: x.toServer, retries: x.retries, createdAt: x.createdAt}
2003− let a = count (do x <- aps; where x.pubkey == pubkey; yield x)
2004− yield (respond {messages: m, presence: p, sessions: s, forwards: f, forwardDetails: fd, activePolls: a, ok: True {}}))
1935+ let myFwds = filter fwds (\x -> x.toKey == pubkey)
1936+ yield (respond {
1937+ messages: countWhere msgs (\x -> x.toKey == pubkey),
1938+ presence: pluck (filter pres (\x -> x.pubkey == pubkey && x.expiresAt > t)) (\x -> {server: x.server, expiresAt: x.expiresAt}),
1939+ sessions: countWhere ses (\x -> x.pubkey == pubkey),
1940+ forwards: count myFwds,
1941+ forwardDetails: pluck myFwds (\x -> {server: x.toServer, retries: x.retries, createdAt: x.createdAt}),
1942+ activePolls: countWhere aps (\x -> x.pubkey == pubkey),
1943+ ok: True {}}))
20051944
20061945handleTopRecipients = \limit authorization respond -> withMaintAuth authorization
20071946 (yield (respond {recipients: [], ok: False {}, total: 0}))
20081947 (do
20091948 msgs <- *messages
2010− let counts = do
2011− pk <- unique (do m <- msgs; yield m.toKey)
2012− yield {pubkey: pk, count: count (do m <- msgs; where m.toKey == pk; yield m)}
1949+ let counts = pluck (unique (pluck msgs (\m -> m.toKey))) (\pk -> {pubkey: pk, count: countWhere msgs (\m -> m.toKey == pk)})
20131950 let sorted = sortBy (\c -> 0 - c.count) counts
20141951 let effLimit = if limit <= 0 then 20 else clamp 1 100 limit
20151952 yield (respond {recipients: takeRelation effLimit sorted, ok: True {}, total: count counts}))
@@ -2020,12 +1957,10 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization
20201957 t <- now
20211958 fwds <- *forwards
20221959 let servers = collectPeers fwds (\f -> f.toServer)
2023− let result = do
2024− s <- servers
2025− let svFwds = do f <- fwds; where f.toServer == s; yield f
2026− let oldest = case minBy svFwds (\f -> f.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt
2027− let retries = fold (\acc f -> acc + f.retries) 0 svFwds
2028− yield {server: s, count: count svFwds, oldestAge: oldest, totalRetries: retries}
1960+ let result = pluck servers (\s ->
1961+ let svFwds = filter fwds (\f -> f.toServer == s) in
1962+ let oldest = case minBy svFwds (\f -> f.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt in
1963+ {server: s, count: count svFwds, oldestAge: oldest, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds})
20291964 yield (respond {servers: result, ok: True {}, total: count fwds}))
20301965
20311966handleStatsRates = \authorization respond -> withMaintAuth authorization