@@ -439,8 +439,11 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
439439-- Generic helpers
440440-- ============================================
441441
442+id = \x -> x
442443filter = \xs pred -> do x <- xs; where pred x; yield x
444+removeWhere = \xs pred -> filter xs (\x -> not (pred x))
443445pluck = \xs f -> do x <- xs; yield (f x)
446+mapWhere = \xs pred f -> pluck (filter xs pred) f
444447countWhere = \xs pred -> count (filter xs pred)
445448any : [a] -> (a -> Bool) -> Bool
446449any = \xs pred -> countWhere xs pred > 0
@@ -449,6 +452,7 @@ none = \xs pred -> not (any xs pred)
449452all : [a] -> (a -> Bool) -> Bool
450453all = \xs pred -> none xs (\x -> not (pred x))
451454elem = \xs x -> any xs (\y -> y == x)
455+findFirst = \xs pred -> headOf (filter xs pred)
452456when = \cond action -> if cond then action else yield {}
453457clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
454458min = \a b -> if a < b then a else b
@@ -462,10 +466,15 @@ pruneRef = \ref keep -> do
462466 cur <- *ref
463467 set *ref = do x <- cur; where keep x; yield x
464468
469+-- Append `items` to a ref, then cap by `getKey` (drops oldest on overflow).
470+appendCapped = \ref items maxN getKey -> do
471+ cur <- *ref
472+ set *ref = capList (union cur items) maxN getKey
473+
465474-- Add `server` to the named list ref if absent. Returns the new total.
466475addServerToList = \listRef server -> atomic do
467476 current <- *listRef
468− let updated = if any current (\x -> x == server) then current else union current [server]
477+ let updated = if elem current server then current else union current [server]
469478 set *listRef = updated
470479 yield (count updated)
471480
@@ -473,7 +482,7 @@ addServerToList = \listRef server -> atomic do
473482removeServerFromList = \listRef server -> atomic do
474483 current <- *listRef
475484 pruneRef listRef (\x -> x != server)
476− yield (any current (\x -> x == server))
485+ yield (elem current server)
477486
478487headOf = \items -> fold (\acc x -> case acc of
479488 Nothing {} -> Just {value: x}
@@ -618,7 +627,7 @@ gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do
618627
619628authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
620629authenticateSession = \sessions token t ->
621− case headOf (filter sessions (\s -> s.expiresAt > t && s.token == token)) of
630+ case findFirst sessions (\s -> s.expiresAt > t && s.token == token) of
622631 Nothing {} -> Unauthorized {}
623632 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
624633
@@ -703,13 +712,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
703712-- Returns {limited, entries}. Increments count when the key fits within the window.
704713-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.
705714checkRate = \entries key maxRate windowMs maxEntries t ->
706− let cur = headOf (filter entries (\r -> r.key == key && t - r.windowStart < windowMs)) in
715+ let cur = findFirst entries (\r -> r.key == key && t - r.windowStart < windowMs) in
707716 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in
708717 if nextCount > maxRate
709718 then {limited: True {}, entries: entries}
710719 else
711720 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in
712− {limited: False {}, entries: capList (union (filter entries (\x -> x.key != key)) [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}
721+ {limited: False {}, entries: capList (union (removeWhere entries (\x -> x.key == key)) [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}
713722
714723checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
715724 ws <- *windowRef
@@ -757,7 +766,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
757766
758767shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
759768shouldRetryServer = \server t failures ->
760− case headOf (filter failures (\f -> f.server == server)) of
769+ case findFirst failures (\f -> f.server == server) of
761770 Nothing {} -> True {}
762771 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures
763772
@@ -765,10 +774,10 @@ recordFailure = \server -> do
765774 t <- now
766775 atomic do
767776 failures <- *forwardFailures
768− let new = case headOf (filter failures (\f -> f.server == server)) of
777+ let new = case findFirst failures (\f -> f.server == server) of
769778 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
770779 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
771− set *forwardFailures = union (filter failures (\f -> f.server != server)) [new]
780+ set *forwardFailures = union (removeWhere failures (\f -> f.server == server)) [new]
772781
773782clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
774783
@@ -817,7 +826,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
817826mergePresence = \existing incoming ->
818827 fold (\acc e ->
819828 if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt) then acc
820− else union (filter acc (\x -> x.pubkey != e.pubkey || x.server != e.server)) [e]) existing incoming
829+ else union (removeWhere acc (\x -> x.pubkey == e.pubkey && x.server == e.server)) [e]) existing incoming
821830
822831-- ============================================
823832-- Pruning
@@ -847,7 +856,26 @@ pruneExpired = \t -> atomic do
847856-- Federation gossip
848857-- ============================================
849858
850−sendBatchedGossip = \peers events hops -> forEach peers (\s -> s) (\server -> do
859+-- Build a gossip event with this server's identity.
860+mkGossip = \etype pk t ttl ->
861+ {eventType: etype, pubkey: pk, server: serverHostN, timestamp: t, ttl: ttl, seenBy: [serverHostN]}
862+
863+-- Build an acked entry for a federated message.
864+mkAckedEntry = \m t -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
865+
866+-- Build a sender-delivery entry for a locally-originated message.
867+mkDeliveryEntry = \m t -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
868+
869+-- Record federation acks and sender deliveries for a batch of just-removed messages.
870+-- Returns the federated-ack entries (the caller flushes them remotely later).
871+recordAcksAndDeliveries = \toAck t -> do
872+ let ackedFed = mapWhere toAck (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
873+ appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
874+ let localDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "") (\m -> mkDeliveryEntry m t)
875+ appendCapped senderDeliveries localDel 10000 (\d -> d.deliveredAt)
876+ yield ackedFed
877+
878+sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do
851879 t <- now
852880 failures <- *forwardFailures
853881 when (shouldRetryServer server t failures) (do
@@ -864,7 +892,7 @@ broadcastOfflineGossip = \t expiredLocal -> do
864892 presence <- *presence
865893 let peers = knownPeerServersFromPresence presence
866894 when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do
867− sendBatchedGossip peers (pluck expiredLocal (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops)
895+ sendBatchedGossip peers (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops)
868896
869897-- ============================================
870898-- Forward retries and acks
@@ -876,7 +904,7 @@ retryPendingForwards = do
876904 failures <- *forwardFailures
877905 blocked <- *blockedPeers
878906 let servers = filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))
879− forEach servers (\s -> s) (\server -> do
907+ forEach servers id (\server -> do
880908 let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))
881909 forEach serverFwds (\f -> f.messageId) (\f -> do
882910 t2 <- now
@@ -898,8 +926,8 @@ retryPendingForwards = do
898926 else {messageId: x.messageId, toServer: x.toServer, createdAt: x.createdAt, toKey: x.toKey, encryptedBlob: x.encryptedBlob, timestamp: x.timestamp, originServer: x.originServer, retries: x.retries + 1, expiresAt: x.expiresAt, priority: x.priority, senderKey: x.senderKey})))
899927
900928sendAcksAndClean = \pendingAcks ->
901− forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do
902− let mids = pluck (filter pendingAcks (\a -> a.originServer == server)) (\a -> a.messageId)
929+ forEach (collectPeers pendingAcks (\a -> a.originServer)) id (\server -> do
930+ let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
903931 t <- now
904932 auth <- generateFederationAuth "ack" t
905933 result <- fetch (federationUrl server) (FederationAck {
@@ -910,16 +938,13 @@ sendAcksAndClean = \pendingAcks ->
910938 Err _ -> recordFailure server
911939 Ok _ -> do
912940 clearFailure server
913− atomic do
914− cur <- *ackedIds
915− set *ackedIds = filter cur (\a -> not (a.originServer == server && elem mids a.messageId)))
941+ atomic (pruneRef ackedIds (\a -> not (a.originServer == server && elem mids a.messageId))))
916942
917943-- Forward queued messages to a server when a remote key comes online.
918944forwardQueuedToServer = \pubkey server t -> atomic do
919945 msgs <- *messages
920946 fwds <- *forwards
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 ->
947+ let candidates = mapWhere msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server)) (\m ->
923948 {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})
924949 set *forwards = capList (union fwds (takeRelation maxForwardsPerServer candidates)) maxForwardsEntries (\f -> f.createdAt)
925950
@@ -955,7 +980,7 @@ refreshLocalPresence = \t -> do
955980 set *presence = updated
956981 yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
957982 when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do
958− let events = pluck refreshState.refreshed (\p -> {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]})
983+ let events = pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey t (clampTtl (sessionExpiry / 1000)))
959984 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))
960985
961986-- ============================================
@@ -1009,27 +1034,27 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10091034 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
10101035 then yield {valid: False {}, peers: []}
10111036 else do
1012− set *challenges = filter ch (\c -> c.pubkey != pubkey || c.challenge != challenge)
1037+ set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
10131038 dp <- *drainPhase
10141039 if dp >= 1 then yield {valid: False {}, peers: []}
10151040 else do
10161041 ses <- *sessions
1017− let kept = if revokeOthers then filter ses (\s -> s.pubkey != pubkey) else ses
1042+ let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses
10181043 let forKey = filter kept (\s -> s.pubkey == pubkey)
10191044 let afterPerKey = if count forKey >= maxSessionsPerKey
1020− then union (filter kept (\s -> s.pubkey != pubkey)) (evictOldest forKey (\s -> s.expiresAt))
1045+ then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))
10211046 else kept
10221047 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
10231048 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
10241049 pres <- *presence
1025− let withoutLocal = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1050+ let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
10261051 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
10271052 set *presence = newPres
10281053 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10291054 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
10301055 else do
10311056 when federationEnabled
1032− (fork (sendBatchedGossip result.peers [{eventType: GossipOnline {}, pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl, seenBy: [serverHostN]}] gossipOriginHops))
1057+ (fork (sendBatchedGossip result.peers [mkGossip (GossipOnline {}) pubkey t onlineGossipTtl] gossipOriginHops))
10331058 yield (respond {token: token, expiresAt: expiry, error: ""})
10341059
10351060handleRevoke = \targetToken authorization clientIp respond -> do
@@ -1044,7 +1069,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
10441069 let found = any ses (\s -> s.pubkey == caller && s.token == targetToken)
10451070 if not found then yield False
10461071 else do
1047− set *sessions = filter ses (\s -> s.pubkey != caller || s.token != targetToken)
1072+ set *sessions = removeWhere ses (\s -> s.pubkey == caller && s.token == targetToken)
10481073 yield True
10491074 yield (respond {ok: result, authorized: True {}}))
10501075
@@ -1060,18 +1085,18 @@ handleLogout = \authorization clientIp respond -> do
10601085 Authorized {pubkey, ip} ->
10611086 if not (matchIp clientIp ip) then yield {ok: False {}, pubkey: "", peers: []}
10621087 else do
1063− set *sessions = filter ses (\s -> s.token != token)
1088+ set *sessions = removeWhere ses (\s -> s.token == token)
10641089 -- If no remaining sessions for this key, drop local presence
10651090 after <- *sessions
10661091 let stillHas = any after (\s -> s.pubkey == pubkey)
10671092 peers <- if stillHas then yield []
10681093 else do
10691094 pres <- *presence
1070− set *presence = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
1095+ set *presence = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
10711096 yield (knownPeerServersFromPresence pres)
10721097 yield {ok: True {}, pubkey: pubkey, peers: peers}
10731098 when (removed.ok && removed.pubkey != "" && count removed.peers > 0 && federationEnabled)
1074− (fork (sendBatchedGossip removed.peers [{eventType: GossipOffline {}, pubkey: removed.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}] gossipOriginHops))
1099+ (fork (sendBatchedGossip removed.peers [mkGossip (GossipOffline {}) removed.pubkey t offlineGossipMinTtl] gossipOriginHops))
10751100 yield (respond {ok: removed.ok, authorized: removed.ok})
10761101
10771102-- ============================================
@@ -1111,10 +1136,9 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11111136 Just {value: s} -> do
11121137 acquired <- atomic do
11131138 aps <- *activePolls
1114− let perKey = countWhere aps (\p -> p.pubkey == s.pubkey)
1115− if perKey >= maxPollsPerKey then yield False
1139+ if countWhere aps (\p -> p.pubkey == s.pubkey) >= maxPollsPerKey then yield False
11161140 else do
1117− set *activePolls = capList (union aps [{pubkey: s.pubkey, startedAt: t}]) maxActivePolls (\p -> p.startedAt)
1141+ appendCapped activePolls [{pubkey: s.pubkey, startedAt: t}] maxActivePolls (\p -> p.startedAt)
11181142 yield True
11191143 if not acquired
11201144 then yield (respond {events: [], cursor: cursor, authorized: True {}, rateLimited: True {}, draining: False {}, generation: serverGen, presenceUpdates: [], heartbeat: False {}, retryAfterMs: pollTimeoutMs, cursorReset: cursorWasReset, hasMore: False {}, cursorStale: False {}, messageExpirations: [], messageDeliveries: [], droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: []})
@@ -1131,14 +1155,14 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11311155 else yield []
11321156 drains <- atomic do
11331157 se <- *senderExpirations
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)
1158+ let myExp = mapWhere se (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1159+ pruneRef senderExpirations (\x -> x.senderKey != s.pubkey)
11361160 sd <- *senderDeliveries
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)
1161+ let myDel = mapWhere sd (\x -> x.senderKey == s.pubkey) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1162+ pruneRef senderDeliveries (\x -> x.senderKey != s.pubkey)
11391163 cn <- *cancelNotifications
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)
1164+ let myCancels = mapWhere cn (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1165+ pruneRef cancelNotifications (\x -> x.toKey != s.pubkey)
11421166 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
11431167 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
11441168 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})
@@ -1171,17 +1195,10 @@ ackPolledMessages = \pubkey cursor t -> do
11711195 let toAck = filter msgs matches
11721196 if count toAck == 0 then yield {pendingAcks: []}
11731197 else do
1174− set *messages = filter msgs (\m -> not (matches m))
1175− fwds <- *forwards
1198+ set *messages = removeWhere msgs matches
11761199 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})
1179− aids <- *ackedIds
1180− set *ackedIds = capList (union aids ackedFederated) maxAckedIds (\a -> a.ackedAt)
1181− -- Sender delivery receipts for locally-originated messages
1182− let localDelivered = pluck (filter toAck (\m -> m.senderKey != "" && m.originServer == "")) (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
1183− sd <- *senderDeliveries
1184− set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)
1200+ pruneRef forwards (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1201+ ackedFederated <- recordAcksAndDeliveries toAck t
11851202 yield {pendingAcks: ackedFederated}
11861203
11871204-- ============================================
@@ -1249,8 +1266,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12491266 forwardsEnqueued <- if federationEnabled && count remoteServers > 0
12501267 then do
12511268 fwds <- *forwards
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})
1253− set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
1269+ let newFwds = mapWhere 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})
1270+ appendCapped forwards newFwds maxForwardsEntries (\f -> f.createdAt)
12541271 yield (count newFwds > 0)
12551272 else yield False
12561273 let status = if localOnline then MsDelivered {}
@@ -1259,10 +1276,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12591276 yield {status: status, remoteServers: remoteServers}
12601277
12611278forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1262− forEach servers (\s -> s) (\server -> do
1279+ forEach servers id (\server -> do
12631280 info <- atomic do
12641281 fwds <- *forwards
1265− case headOf (filter fwds (\f -> f.messageId == effMsgId && f.toServer == server)) of
1282+ case findFirst fwds (\f -> f.messageId == effMsgId && f.toServer == server) of
12661283 Nothing {} -> yield Nothing {}
12671284 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
12681285 case info of
@@ -1281,9 +1298,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12811298 Ok r ->
12821299 if r.ok then do
12831300 clearFailure server
1284− atomic do
1285− fwds <- *forwards
1286− set *forwards = filter fwds (\f -> not (f.messageId == effMsgId && f.toServer == server))
1301+ atomic (pruneRef forwards (\f -> not (f.messageId == effMsgId && f.toServer == server)))
12871302 else yield {})
12881303
12891304-- ============================================
@@ -1350,15 +1365,9 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
13501365 let toAck = filter msgs matches
13511366 if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}
13521367 else do
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)
1368+ set *messages = removeWhere msgs matches
1369+ pruneRef forwards (\f -> not (matches f))
1370+ ackedFed <- recordAcksAndDeliveries toAck t
13621371 yield {ackedCount: count toAck, pendingAcks: ackedFed}
13631372 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
13641373 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
@@ -1374,8 +1383,7 @@ handleListMessages = \offset authorization clientIp respond -> do
13741383 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
13751384 let total = count myMsgs
13761385 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
1386+ let after = mapWhere (withIndex myMsgs) (\p -> p.i >= effectiveOffset) (\p -> p.v)
13791387 let result = pluck (takeRelation maxPollEvents after) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
13801388 yield (respond {messages: result, authorized: True {}, count: total, error: ""}))
13811389
@@ -1393,12 +1401,10 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
13931401 let toDel = filter msgs matches
13941402 if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}
13951403 else do
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)
1404+ set *messages = removeWhere msgs matches
1405+ pruneRef forwards (\f -> not (matches f))
1406+ let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1407+ appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
14021408 yield {deletedCount: count toDel, pendingAcks: ackedFed}
14031409 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
14041410 yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
@@ -1413,27 +1419,26 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
14131419 let matches = \x -> x.messageId == messageId && x.toKey == to
14141420 result <- atomic do
14151421 msgs <- *messages
1416− case headOf (filter msgs matches) of
1422+ case findFirst msgs matches of
14171423 Nothing {} -> yield {cancelled: False {}, cancelServers: []}
14181424 Just {value: msg} ->
14191425 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
14201426 then yield {cancelled: False {}, cancelServers: []}
14211427 else do
1422− set *messages = filter msgs (\m -> not (matches m))
1428+ set *messages = removeWhere msgs matches
14231429 fwds <- *forwards
14241430 let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)
1425− set *forwards = filter fwds (\f -> not (matches f))
1431+ pruneRef forwards (\f -> not (matches f))
14261432 pres <- *presence
14271433 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)
1434+ appendCapped cancelNotifications [{toKey: to, messageId: messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
14301435 yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
14311436 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
14321437 (fork (sendCancelToEach messageId to result.cancelServers))
14331438 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
14341439
14351440sendCancelToEach = \messageId to servers ->
1436− forEach servers (\s -> s) (\server -> do
1441+ forEach servers id (\server -> do
14371442 t <- now
14381443 auth <- generateFederationAuth ("cancel:" ++ messageId) t
14391444 result <- fetch (federationUrl server) (FederationCancel {
@@ -1441,9 +1446,8 @@ sendCancelToEach = \messageId to servers ->
14411446 requestTimestamp: t, correlationId: "", authorization: auth,
14421447 protocolVersion: federationProtocolVersion})
14431448 let failed = case result of Err _ -> True {}; Ok r -> not r.ok
1444− when failed (atomic do
1445− pfc <- *pendingFederationCancels
1446− set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)))
1449+ when failed (atomic
1450+ (appendCapped pendingFederationCancels [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}] 10000 (\c -> c.createdAt))))
14471451
14481452handleCancelBatch = \cancels authorization clientIp respond -> do
14491453 t <- now
@@ -1456,17 +1460,15 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
14561460 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
14571461 r <- atomic do
14581462 msgs <- *messages
1459− case headOf (filter msgs matches) of
1463+ case findFirst msgs matches of
14601464 Nothing {} -> yield False
14611465 Just {value: msg} ->
14621466 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
14631467 then yield False
14641468 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)
1469+ set *messages = removeWhere msgs matches
1470+ pruneRef forwards (\f -> not (matches f))
1471+ appendCapped cancelNotifications [{toKey: e.to, messageId: e.messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
14701472 yield True
14711473 yield {messageId: e.messageId, cancelled: r})
14721474 yield (respond {results: results, authorized: True {}, error: ""}))
@@ -1475,7 +1477,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
14751477 t <- now
14761478 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
14771479 gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1478− results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do
1480+ results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do
14791481 effMsgId <- if m.messageId != "" then yield m.messageId else genToken
14801482 let clampedTs = clampTimestamp m.timestamp t
14811483 let clampedPri = clamp 0 maxMessagePriority m.priority
@@ -1526,11 +1528,11 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
15261528 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
15271529 let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
15281530 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 online = mapWhere validEvents isOnline (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})
1532+ let offline = mapWhere validEvents isOffline (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
15311533 let withoutOffline = filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))
15321534 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})
1535+ let newlyOnline = mapWhere 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})
15341536 yield {newlyOnline: newlyOnline}
15351537 when (count valid.newlyOnline > 0)
15361538 (fork (forwardForNewlyOnline valid.newlyOnline t))
@@ -1588,7 +1590,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
15881590 fwds <- *forwards
15891591 let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
15901592 let removed = filter fwds isAcked
1591− set *forwards = filter fwds (\f -> not (isAcked f))
1593+ set *forwards = removeWhere fwds isAcked
15921594 -- Reclaim origin messages with no remaining forwards
15931595 msgs <- *messages
15941596 let removedIds = pluck removed (\f -> f.messageId)
@@ -1599,7 +1601,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
15991601 && t - m.receivedAt >= ackGracePeriod
16001602 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
16011603 && 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))
1604+ set *messages = removeWhere msgs (\m -> any toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
16031605 let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})
16041606 yield {ackedCount: count removed, results: perResult}
16051607 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
@@ -1611,14 +1613,12 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
16111613 let matches = \x -> x.messageId == messageId && x.toKey == toKey
16121614 cancelled <- atomic do
16131615 msgs <- *messages
1614− case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of
1616+ case findFirst msgs (\m -> matches m && m.originServer == fromServer) of
16151617 Nothing {} -> yield False
16161618 Just {} -> do
1617− set *messages = filter msgs (\m -> not (matches m))
1618− fwds <- *forwards
1619− set *forwards = filter fwds (\f -> not (matches f))
1620− cn <- *cancelNotifications
1621− set *cancelNotifications = capList (union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1619+ set *messages = removeWhere msgs matches
1620+ pruneRef forwards (\f -> not (matches f))
1621+ appendCapped cancelNotifications [{toKey: toKey, messageId: messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
16221622 yield True
16231623 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
16241624
@@ -1630,12 +1630,11 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
16301630 let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
16311631 cancelled <- atomic do
16321632 msgs <- *messages
1633− case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of
1633+ case findFirst msgs (\m -> matches m && m.originServer == fromServer) of
16341634 Nothing {} -> yield False
16351635 Just {} -> do
1636− set *messages = filter msgs (\m -> not (matches m))
1637− fwds <- *forwards
1638− set *forwards = filter fwds (\f -> not (matches f))
1636+ set *messages = removeWhere msgs matches
1637+ pruneRef forwards (\f -> not (matches f))
16391638 yield True
16401639 yield {messageId: e.messageId, cancelled: cancelled})
16411640 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
@@ -1684,13 +1683,12 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
16841683 let pairs = zipPairs messageIds toKeys
16851684 atomic do
16861685 msgs <- *messages
1687− sd <- *senderDeliveries
16881686 let receipts = do
16891687 p <- pairs
16901688 m <- msgs
16911689 where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
16921690 yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}
1693− set *senderDeliveries = capList (union sd receipts) 10000 (\d -> d.deliveredAt)
1691+ appendCapped senderDeliveries receipts 10000 (\d -> d.deliveredAt)
16941692 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
16951693
16961694-- Pair up parallel lists by index (O(n*m) join on i; trivial code).
@@ -1719,7 +1717,7 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
17191717 let peers = knownPeerServersFromPresence pres
17201718 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}
17211719 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)
1722− (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops))
1720+ (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey t offlineGossipMinTtl)) gossipOriginHops))
17231721 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
17241722
17251723handleUndrain = \authorization respond -> withMaintAuth authorization
@@ -1803,7 +1801,7 @@ handleAdminListSessions = \pubkey authorization respond -> withMaintAuth authori
18031801 (yield (respond {sessions: [], count: 0, ok: False {}}))
18041802 (do
18051803 ses <- *sessions
1806− let mine = pluck (filter ses (\s -> s.pubkey == pubkey)) (\s -> {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt})
1804+ let mine = mapWhere ses (\s -> s.pubkey == pubkey) (\s -> {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt})
18071805 yield (respond {sessions: mine, count: count mine, ok: True {}}))
18081806
18091807handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization
@@ -1825,9 +1823,8 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
18251823 let mine = filter msgs (\m -> m.toKey == pubkey)
18261824 pruneRef messages (\m -> m.toKey != pubkey)
18271825 pruneRef forwards (\f -> f.toKey != pubkey)
1828− let federated = pluck (filter mine (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
1829− aids <- *ackedIds
1830− set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)
1826+ let federated = mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1827+ appendCapped ackedIds federated maxAckedIds (\a -> a.ackedAt)
18311828 yield (count mine)
18321829 yield (respond {ok: True {}, purgedCount: purged}))
18331830
@@ -1837,9 +1834,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
18371834 else do
18381835 result <- atomic do
18391836 ses <- *sessions
1840− let mine = filter ses (\s -> s.ip == ip)
1837+ let myKeys = mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey)
18411838 pruneRef sessions (\s -> s.ip != ip)
1842− yield {revoked: count mine, keys: count (unique (pluck mine (\s -> s.pubkey)))}
1839+ yield {revoked: count myKeys, keys: count (unique myKeys)}
18431840 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
18441841
18451842handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
@@ -1852,7 +1849,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
18521849 let mine = filter active (\s -> s.pubkey == pk) in
18531850 {pubkey: pk,
18541851 count: count mine,
1855− ips: unique (pluck (filter mine (\s -> s.ip != "")) (\s -> s.ip)),
1852+ ips: unique (mapWhere mine (\s -> s.ip != "") (\s -> s.ip)),
18561853 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
18571854 let sorted = sortBy (\g -> 0 - g.count) grouped
18581855 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit
@@ -1935,7 +1932,7 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
19351932 let myFwds = filter fwds (\x -> x.toKey == pubkey)
19361933 yield (respond {
19371934 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}),
1935+ presence: mapWhere pres (\x -> x.pubkey == pubkey && x.expiresAt > t) (\x -> {server: x.server, expiresAt: x.expiresAt}),
19391936 sessions: countWhere ses (\x -> x.pubkey == pubkey),
19401937 forwards: count myFwds,
19411938 forwardDetails: pluck myFwds (\x -> {server: x.toServer, retries: x.retries, createdAt: x.createdAt}),