@@ -312,25 +312,11 @@ route Api = AuthApi | ClientApi | FederationApi
312312-- Everything is intentionally O(n) over flat lists; performance is sacrificed
313313-- for clarity per the file header.
314314
315−-- Keep elements satisfying `pred`. Uses Knot's relation comprehension (`where`).
316−filter : a b -> (b -> Bool) -> a b
317−filter = \xs pred -> do
318− x <- xs
319− where pred x
320− yield x
321−-- Inverse of `filter`: drop elements satisfying `pred`.
322−removeWhere : a b -> (b -> Bool) -> a b
323−removeWhere = \xs pred -> filter xs (\x -> not pred x)
324−-- Map without filtering (a.k.a. classic `map`).
325−pluck : a b -> (b -> c) -> a c
326−pluck = \xs f -> do
327− x <- xs
328− yield (f x)
329315-- Filter then map in one pass.
330316mapWhere : a b -> (b -> Bool) -> (b -> c) -> a c
331−mapWhere = \xs pred -> pluck (filter xs pred)
317+mapWhere = \xs pred fn -> map fn (filter pred xs)
332318countWhere : a b -> (b -> Bool) -> Int
333−countWhere = \xs pred -> count (filter xs pred)
319+countWhere = \xs pred -> count (filter pred xs)
334320-- True when at least one element matches.
335321any : [a] -> (a -> Bool) -> Bool
336322any = \xs pred -> countWhere xs pred > 0
@@ -341,7 +327,7 @@ none = \xs pred -> not any xs pred
341327all : [a] -> (a -> Bool) -> Bool
342328all = \xs pred -> none xs (\x -> not pred x)
343329findFirst : [a] -> (a -> Bool) -> Maybe a
344−findFirst = \xs pred -> headOf (filter xs pred)
330+findFirst = \xs pred -> headOf (filter pred xs)
345331-- Run `action` only when `cond` is True; otherwise no-op.
346332when : Bool -> IO {} {} -> IO {} {}
347333when = \cond action -> if cond then action else yield {}
@@ -368,7 +354,7 @@ capList = \items maxN getKey -> dropRelation (max 0 (count items - maxN)) (sortB
368354-- single key can't exhaust a global pool — see `maxChallengesPerKey` and
369355-- `maxSessionsPerKey`.
370356capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]
371−capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
357+capPerKey = \items pred maxN getKey -> union (filter (\x -> not pred x) items) (capList (filter pred items) (maxN - 1) getKey)
372358
373359-- Sequence an IO action over each item, in order. Recurses through the list
374360-- one element at a time; this is the closest equivalent of `mapM_` here.
@@ -520,7 +506,7 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
520506-- `entries` list is what the caller must write back to the corresponding ref;
521507-- it has been capped at 5000 rows (LRU by `windowStart`) to bound memory.
522508checkRate : [RateEntry] -> Text -> Int -> Int<Ms> -> Timestamp -> {limited: Bool, entries: [RateEntry]}
523−checkRate = \entries key maxRate windowMs t -> (\cur -> if cur.count > maxRate then {limited: True {}, entries} else {limited: False {}, entries: capList (union (removeWhere entries (\x -> x.key == key)) [{key, cur.windowStart, cur.count}]) 5000 (\r -> r.windowStart)}) (case findFirst entries (\r -> r.key == key && t - r.windowStart < windowMs) of
509+checkRate = \entries key maxRate windowMs t -> (\cur -> if cur.count > maxRate then {limited: True {}, entries} else {limited: False {}, entries: capList (union (filter (\x -> not (x.key == key)) entries) [{key, cur.windowStart, cur.count}]) 5000 (\r -> r.windowStart)}) (case findFirst entries (\r -> r.key == key && t - r.windowStart < windowMs) of
524510 Just {value: r} -> {r.windowStart, count: r.count + 1}
525511 _ -> {windowStart: t, count: 1})
526512
@@ -608,7 +594,7 @@ recordFailure = \server -> do
608594 let prev = case findFirst failures same of
609595 Just {value: f} -> {failures: f.failures + 1, f.firstFailedAt}
610596 _ -> {failures: 1, firstFailedAt: t}
611− *forwardFailures = union (removeWhere failures same) [
597+ *forwardFailures = union (filter (\f -> not same f) failures) [
612598 {server, failedAt: t, prev.failures, prev.firstFailedAt}
613599 ])
614600
@@ -617,7 +603,7 @@ recordFailure = \server -> do
617603clearFailure : Text -> IO {rw *forwardFailures} {}
618604clearFailure = \server -> atomic (do
619605 forwardFailures <- *forwardFailures
620− *forwardFailures = filter forwardFailures (\f -> f.server != server))
606+ *forwardFailures = filter (\f -> f.server != server) forwardFailures)
621607
622608-- ============================================
623609-- Common utility queries
@@ -633,12 +619,12 @@ nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
633619-- All unexpired presence rows for a given pubkey (may include both local and
634620-- remote entries — the recipient may live on multiple servers simultaneously).
635621presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
636−presenceForKey = \presence pubkey t -> filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)
622+presenceForKey = \presence pubkey t -> filter (\p -> p.pubkey == pubkey && p.expiresAt > t) presence
637623
638624-- Project items through `getServer`, drop our own hostname, and dedupe.
639625-- Used to derive peer recipient lists from forwards/presence/etc.
640626collectPeers : [a] -> (a -> Text) -> [Text]
641−collectPeers = \items getServer -> filter (pluck items getServer) isRemote
627+collectPeers = \items getServer -> filter isRemote (map getServer items)
642628
643629-- All distinct remote hostnames currently appearing in `presence` (the gossip
644630-- broadcast set we know about, less ourselves).
@@ -658,18 +644,18 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
658644pruneExpired : Int<Ms> -> IO {rw *challenges, rw *forwardFailures, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: ServerName}]
659645pruneExpired = \t -> atomic do
660646 ch <- *challenges
661− *challenges = filter ch (\x -> x.expiresAt > t)
647+ *challenges = filter (\x -> x.expiresAt > t) ch
662648 ses <- *sessions
663− *sessions = filter ses (\x -> x.expiresAt > t)
649+ *sessions = filter (\x -> x.expiresAt > t) ses
664650 msgs <- *messages
665− *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
651+ *messages = filter (\m -> t - m.receivedAt < messageMaxAge) msgs
666652 ff <- *forwardFailures
667− *forwardFailures = filter ff (\f -> t - f.failedAt < 86400000)
653+ *forwardFailures = filter (\f -> t - f.failedAt < 86400000) ff
668654 -- presence: remember locally-expired keys for offline gossip
669655 pres <- *presence
670− *presence = filter pres (\p -> p.expiresAt > t)
656+ *presence = filter (\p -> p.expiresAt > t) pres
671657 let stillLive = \pk -> any pres (\k -> isLocalServer k.server && k.pubkey == pk && k.expiresAt > t)
672− yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)))
658+ yield (filter (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)) pres)
673659
674660-- ============================================
675661-- Federation gossip
@@ -695,9 +681,9 @@ sendBatchedGossip = \peers events -> forEach peers (\server -> do
695681broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
696682broadcastOfflineGossip = \expiredLocal -> do
697683 presence <- *presence
698− when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> GossipOffline {
684+ when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (map (\p -> GossipOffline {
699685 p.pubkey
700− })))
686+ }) expiredLocal))
701687
702688-- ============================================
703689-- Forward retries
@@ -716,11 +702,8 @@ retryPendingForwards = do
716702 t <- now
717703 fwds <- *forwards
718704 failures <- *forwardFailures
719− forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures)) (\server -> forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> do
720− result <- fetch (federationUrl server) (ForwardMessage {
721− f.toKey,
722− f.encryptedBlob
723− })
705+ forEach (filter (\s -> shouldRetryServer s t failures) (collectPeers fwds (\f -> f.toServer))) (\server -> forEach (takeRelation maxBatchSize (filter (\f -> f.toServer == server) fwds)) (\f -> do
706+ result <- fetch (federationUrl server) (ForwardMessage {f.toKey, f.encryptedBlob})
724707 success <- case result of
725708 Err _ -> do
726709 recordFailure server
@@ -732,22 +715,22 @@ retryPendingForwards = do
732715 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
733716 atomic (do
734717 cur <- *forwards
735− *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
718+ *forwards = map (\x -> if isMatch x
736719 then {x.toServer, x.toKey, x.encryptedBlob, retries: x.retries + 1}
737− else x))))
720+ else x) (filter (\x -> not isDone x) cur))))
738721
739722-- Forward queued messages to a server when a remote key comes online.
740723forwardQueuedToServer : PubkeyHex -> ServerName -> IO {rw *forwards, r *messages} {}
741724forwardQueuedToServer = \pubkey server -> atomic (do
742725 msgs <- *messages
743726 fwds <- *forwards
744− let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server))
745− let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m -> {
727+ let candidates = filter (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server)) msgs
728+ let newFwds = map (\m -> {
746729 toServer: server,
747730 m.toKey,
748731 m.encryptedBlob,
749732 retries: 0
750− })
733+ }) (takeRelation maxForwardsPerServer candidates)
751734 *forwards = union fwds newFwds)
752735
753736-- ============================================
@@ -789,7 +772,7 @@ handleChallenge = \pubkey clientIp -> do
789772 challenge <- genToken
790773 result <- atomic (do
791774 ch <- *challenges
792− let cappedActive = capPerKey (filter ch (\c -> c.expiresAt > t)) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
775+ let cappedActive = capPerKey (filter (\c -> c.expiresAt > t) ch) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
793776 if count cappedActive >= maxChallenges
794777 then yield "capacity"
795778 else do
@@ -832,16 +815,16 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
832815 if not valid
833816 then yield {valid: False {}, peers: []}
834817 else do
835− *challenges = removeWhere ch same
818+ *challenges = filter (\c -> not same c) ch
836819 ses <- *sessions
837820 let mine = \s -> s.pubkey == pubkey
838821 let pruned = if revokeOthers
839− then removeWhere ses mine
822+ then filter (\s -> not mine s) ses
840823 else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)
841824 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
842825 *sessions = capList (union pruned [newSession]) maxSessions (\s -> s.expiresAt)
843826 pres <- *presence
844− let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
827+ let prunedPres = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
845828 let newPres = union prunedPres [{pubkey, server: serverHost, expiresAt: expiry}]
846829 *presence = newPres
847830 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
@@ -881,20 +864,22 @@ pollOnce = \authorization cursor clientIp t -> do
881864 startWakeup <- *pollWakeup
882865 pres <- *presence
883866 let mine = \p -> p.pubkey == pubkey && isLocalServer p.server
884− let stale = not (any pres (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2))
867+ let stale = not any pres (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2)
885868 refreshPeers <- if stale
886− then do
887− let kept = removeWhere pres mine
888− let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
889− *presence = newPres
890− yield (Just {value: knownPeerServersFromPresence newPres})
891− else yield (Nothing {})
869+ then do
870+ let kept = filter (\p -> not mine p) pres
871+ let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
872+ *presence = newPres
873+ yield (Just {value: knownPeerServersFromPresence newPres})
874+ else yield (Nothing {})
892875 yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
893876 case setup of
894877 Nothing {} -> yield (Err {error: {status: 401, message: "unauthorized"}})
895878 Just {value: s} -> do
896879 case s.refreshPeers of
897− Just {value: peers} -> when federationEnabled (fork (sendBatchedGossip peers [GossipOnline {pubkey: s.pubkey}]))
880+ Just {value: peers} -> when federationEnabled (fork (sendBatchedGossip peers [
881+ GossipOnline {s.pubkey}
882+ ]))
898883 Nothing {} -> yield {}
899884 pollResult <- doPoll s t
900885 yield (Ok {value: {pollResult.events, pollResult.cursor}})
@@ -924,7 +909,7 @@ doPoll = \s t -> do
924909-- which the client echoes on the next /poll to advance — and which the
925910-- server treats as an implicit ack.
926911collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp}
927−collectPollEvents = \messages pubkey cursor t -> (\limited -> {events: pluck limited (\m -> {m.encryptedBlob}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))))
912+collectPollEvents = \messages pubkey cursor t -> (\limited -> {events: map (\m -> {m.encryptedBlob}) limited, cursor: fold (\acc m -> max acc m.receivedAt) cursor limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (filter (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor) messages)))
928913
929914-- Treat the supplied cursor as an ack: drop all messages with
930915-- receivedAt <= cursor, plus their pending forwards.
@@ -932,11 +917,11 @@ ackPolledMessages : PubkeyHex -> Int -> IO {rw *forwards, rw *messages} {}
932917ackPolledMessages = \pubkey cursor -> do
933918 msgs <- *messages
934919 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
935− let toAck = filter msgs matches
936− let blobs = pluck toAck (\m -> m.encryptedBlob)
937− *messages = removeWhere msgs matches
920+ let toAck = filter matches msgs
921+ let blobs = map (\m -> m.encryptedBlob) toAck
922+ *messages = filter (\m -> not matches m) msgs
938923 fwds <- *forwards
939− *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem f.encryptedBlob blobs))
924+ *forwards = filter (\f -> not (f.toKey == pubkey && elem f.encryptedBlob blobs)) fwds
940925
941926-- ============================================
942927-- Handler: send
@@ -1036,24 +1021,21 @@ forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
10361021 Nothing {} -> yield {}
10371022 Just {value: f} -> do
10381023 t <- now
1039− result <- fetch (federationUrl server) (ForwardMessage {
1040− toKey: to,
1041− encryptedBlob
1042− })
1024+ result <- fetch (federationUrl server) (ForwardMessage {toKey: to, encryptedBlob})
10431025 case result of
10441026 Err _ -> recordFailure server
10451027 Ok _ -> do
10461028 clearFailure server
10471029 atomic (do
10481030 fwds2 <- *forwards
1049− let remaining = filter fwds2 (\x -> not same x)
1031+ let remaining = filter (\x -> not same x) fwds2
10501032 *forwards = remaining
10511033 pres <- *presence
10521034 let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)
10531035 let othersPending = any remaining (\x -> x.encryptedBlob == encryptedBlob)
10541036 when (not stillLocal && not othersPending) (do
10551037 msgs <- *messages
1056− *messages = filter msgs (\m -> m.encryptedBlob != encryptedBlob))))
1038+ *messages = filter (\m -> m.encryptedBlob != encryptedBlob) msgs)))
10571039
10581040-- ============================================
10591041-- Handler: federation receive
@@ -1086,12 +1068,12 @@ handleRecvGossip = \events fromServer -> do
10861068 GossipOnline {pubkey} -> pubkey
10871069 GossipOffline {pubkey} -> pubkey
10881070 _ -> ""
1089− let onlinePubkeys = pluck (filter events isOnline) pubkeyOf
1090− let touchedPubkeys = pluck (filter events (\e -> isOnline e || isOffline e)) pubkeyOf
1091− let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)
1071+ let onlinePubkeys = map pubkeyOf (filter isOnline events)
1072+ let touchedPubkeys = map pubkeyOf (filter (\e -> isOnline e || isOffline e) events)
1073+ let kept = filter (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys) pres
10921074 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
10931075 *presence = union kept added
1094− yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))
1076+ yield (filter (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t)) onlinePubkeys))
10951077 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer))
10961078 yield (Ok {value: {}})
10971079