@@ -439,8 +439,11 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
439-- Generic helpers
439-- Generic helpers
440-- ============================================
440-- ============================================
442filter = \xs pred -> do x <- xs; where pred x; yield x
443filter = \xs pred -> do x <- xs; where pred x; yield x
444+removeWhere = \xs pred -> filter xs (\x -> not (pred x))
443pluck = \xs f -> do x <- xs; yield (f x)
445pluck = \xs f -> do x <- xs; yield (f x)
446+mapWhere = \xs pred f -> pluck (filter xs pred) f
444countWhere = \xs pred -> count (filter xs pred)
447countWhere = \xs pred -> count (filter xs pred)
445any : [a] -> (a -> Bool) -> Bool
448any : [a] -> (a -> Bool) -> Bool
446any = \xs pred -> countWhere xs pred > 0
449any = \xs pred -> countWhere xs pred > 0
@@ -449,6 +452,7 @@ none = \xs pred -> not (any xs pred)
449all : [a] -> (a -> Bool) -> Bool
452all : [a] -> (a -> Bool) -> Bool
450all = \xs pred -> none xs (\x -> not (pred x))
453all = \xs pred -> none xs (\x -> not (pred x))
451elem = \xs x -> any xs (\y -> y == x)
454elem = \xs x -> any xs (\y -> y == x)
455+findFirst = \xs pred -> headOf (filter xs pred)
452when = \cond action -> if cond then action else yield {}
456when = \cond action -> if cond then action else yield {}
453clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
457clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
454min = \a b -> if a < b then a else b
458min = \a b -> if a < b then a else b
@@ -462,10 +466,15 @@ pruneRef = \ref keep -> do
462 cur <- *ref
466 cur <- *ref
463 set *ref = do x <- cur; where keep x; yield x
467 set *ref = do x <- cur; where keep x; yield x
469+-- Append `items` to a ref, then cap by `getKey` (drops oldest on overflow).
470+appendCapped = \ref items maxN getKey -> do
472+ set *ref = capList (union cur items) maxN getKey
465-- Add `server` to the named list ref if absent. Returns the new total.
474-- Add `server` to the named list ref if absent. Returns the new total.
466addServerToList = \listRef server -> atomic do
475addServerToList = \listRef server -> atomic do
467 current <- *listRef
476 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]
469 set *listRef = updated
478 set *listRef = updated
470 yield (count updated)
479 yield (count updated)
@@ -473,7 +482,7 @@ addServerToList = \listRef server -> atomic do
473removeServerFromList = \listRef server -> atomic do
482removeServerFromList = \listRef server -> atomic do
474 current <- *listRef
483 current <- *listRef
475 pruneRef listRef (\x -> x != server)
484 pruneRef listRef (\x -> x != server)
476− yield (any current (\x -> x == server))
485+ yield (elem current server)
478headOf = \items -> fold (\acc x -> case acc of
487headOf = \items -> fold (\acc x -> case acc of
479 Nothing {} -> Just {value: x}
488 Nothing {} -> Just {value: x}
@@ -618,7 +627,7 @@ gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do
619authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
628authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
620authenticateSession = \sessions token t ->
629authenticateSession = \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
622 Nothing {} -> Unauthorized {}
631 Nothing {} -> Unauthorized {}
623 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
632 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}
@@ -703,13 +712,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp
703-- Returns {limited, entries}. Increments count when the key fits within the window.
712-- Returns {limited, entries}. Increments count when the key fits within the window.
704-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.
713-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.
705checkRate = \entries key maxRate windowMs maxEntries t ->
714checkRate = \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
707 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in
716 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in
708 if nextCount > maxRate
717 if nextCount > maxRate
709 then {limited: True {}, entries: entries}
718 then {limited: True {}, entries: entries}
711 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in
720 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)}
714checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
723checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do
715 ws <- *windowRef
724 ws <- *windowRef
@@ -757,7 +766,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur
758shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
767shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
759shouldRetryServer = \server t failures ->
768shouldRetryServer = \server t failures ->
760− case headOf (filter failures (\f -> f.server == server)) of
769+ case findFirst failures (\f -> f.server == server) of
761 Nothing {} -> True {}
770 Nothing {} -> True {}
762 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures
771 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures
@@ -765,10 +774,10 @@ recordFailure = \server -> do
766 atomic do
775 atomic do
767 failures <- *forwardFailures
776 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
769 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
778 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
770 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}
779 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]
773clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
782clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))
@@ -817,7 +826,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve
817mergePresence = \existing incoming ->
826mergePresence = \existing incoming ->
818 fold (\acc e ->
827 fold (\acc e ->
819 if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt) then acc
828 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
822-- ============================================
831-- ============================================
823-- Pruning
832-- Pruning
@@ -847,7 +856,26 @@ pruneExpired = \t -> atomic do
847-- Federation gossip
856-- Federation gossip
848-- ============================================
857-- ============================================
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]}
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}
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}
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)
850−sendBatchedGossip = \peers events hops -> forEach peers (\s -> s) (\server -> do
878+sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do
852 failures <- *forwardFailures
880 failures <- *forwardFailures
853 when (shouldRetryServer server t failures) (do
881 when (shouldRetryServer server t failures) (do
@@ -864,7 +892,7 @@ broadcastOfflineGossip = \t expiredLocal -> do
864 presence <- *presence
892 presence <- *presence
865 let peers = knownPeerServersFromPresence presence
893 let peers = knownPeerServersFromPresence presence
866 when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do
894 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)
869-- ============================================
897-- ============================================
870-- Forward retries and acks
898-- Forward retries and acks
@@ -876,7 +904,7 @@ retryPendingForwards = do
876 failures <- *forwardFailures
904 failures <- *forwardFailures
877 blocked <- *blockedPeers
905 blocked <- *blockedPeers
878 let servers = filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))
906 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
880 let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))
908 let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))
881 forEach serverFwds (\f -> f.messageId) (\f -> do
909 forEach serverFwds (\f -> f.messageId) (\f -> do
882 t2 <- now
910 t2 <- now
@@ -898,8 +926,8 @@ retryPendingForwards = do
898 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})))
926 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})))
900sendAcksAndClean = \pendingAcks ->
928sendAcksAndClean = \pendingAcks ->
901− forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do
929+ forEach (collectPeers pendingAcks (\a -> a.originServer)) id (\server -> do
902− let mids = pluck (filter pendingAcks (\a -> a.originServer == server)) (\a -> a.messageId)
930+ let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
904 auth <- generateFederationAuth "ack" t
932 auth <- generateFederationAuth "ack" t
905 result <- fetch (federationUrl server) (FederationAck {
933 result <- fetch (federationUrl server) (FederationAck {
@@ -910,16 +938,13 @@ sendAcksAndClean = \pendingAcks ->
910 Err _ -> recordFailure server
938 Err _ -> recordFailure server
911 Ok _ -> do
939 Ok _ -> do
912 clearFailure server
940 clearFailure server
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))))
917-- Forward queued messages to a server when a remote key comes online.
943-- Forward queued messages to a server when a remote key comes online.
918forwardQueuedToServer = \pubkey server t -> atomic do
944forwardQueuedToServer = \pubkey server t -> atomic do
919 msgs <- *messages
945 msgs <- *messages
920 fwds <- *forwards
946 fwds <- *forwards
921− let mine = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server))
947+ let candidates = mapWhere msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server)) (\m ->
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})
948 {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)
949 set *forwards = capList (union fwds (takeRelation maxForwardsPerServer candidates)) maxForwardsEntries (\f -> f.createdAt)
@@ -955,7 +980,7 @@ refreshLocalPresence = \t -> do
955 set *presence = updated
980 set *presence = updated
956 yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
981 yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}
957 when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do
982 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)))
959 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))
984 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))
961-- ============================================
986-- ============================================
@@ -1009,27 +1034,27 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
1009 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1034 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1010 then yield {valid: False {}, peers: []}
1035 then yield {valid: False {}, peers: []}
1012− set *challenges = filter ch (\c -> c.pubkey != pubkey || c.challenge != challenge)
1037+ set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1013 dp <- *drainPhase
1038 dp <- *drainPhase
1014 if dp >= 1 then yield {valid: False {}, peers: []}
1039 if dp >= 1 then yield {valid: False {}, peers: []}
1016 ses <- *sessions
1041 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
1018 let forKey = filter kept (\s -> s.pubkey == pubkey)
1043 let forKey = filter kept (\s -> s.pubkey == pubkey)
1019 let afterPerKey = if count forKey >= maxSessionsPerKey
1044 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))
1021 else kept
1046 else kept
1022 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1047 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1023 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
1048 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
1024 pres <- *presence
1049 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)
1026 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1051 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1027 set *presence = newPres
1052 set *presence = newPres
1028 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
1053 yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
1029 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
1054 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
1031 when federationEnabled
1056 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))
1033 yield (respond {token: token, expiresAt: expiry, error: ""})
1058 yield (respond {token: token, expiresAt: expiry, error: ""})
1035handleRevoke = \targetToken authorization clientIp respond -> do
1060handleRevoke = \targetToken authorization clientIp respond -> do
@@ -1044,7 +1069,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
1044 let found = any ses (\s -> s.pubkey == caller && s.token == targetToken)
1069 let found = any ses (\s -> s.pubkey == caller && s.token == targetToken)
1045 if not found then yield False
1070 if not found then yield False
1047− set *sessions = filter ses (\s -> s.pubkey != caller || s.token != targetToken)
1072+ set *sessions = removeWhere ses (\s -> s.pubkey == caller && s.token == targetToken)
1048 yield True
1073 yield True
1049 yield (respond {ok: result, authorized: True {}}))
1074 yield (respond {ok: result, authorized: True {}}))
@@ -1060,18 +1085,18 @@ handleLogout = \authorization clientIp respond -> do
1060 Authorized {pubkey, ip} ->
1085 Authorized {pubkey, ip} ->
1061 if not (matchIp clientIp ip) then yield {ok: False {}, pubkey: "", peers: []}
1086 if not (matchIp clientIp ip) then yield {ok: False {}, pubkey: "", peers: []}
1063− set *sessions = filter ses (\s -> s.token != token)
1088+ set *sessions = removeWhere ses (\s -> s.token == token)
1064 -- If no remaining sessions for this key, drop local presence
1089 -- If no remaining sessions for this key, drop local presence
1065 after <- *sessions
1090 after <- *sessions
1066 let stillHas = any after (\s -> s.pubkey == pubkey)
1091 let stillHas = any after (\s -> s.pubkey == pubkey)
1067 peers <- if stillHas then yield []
1092 peers <- if stillHas then yield []
1069 pres <- *presence
1094 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)
1071 yield (knownPeerServersFromPresence pres)
1096 yield (knownPeerServersFromPresence pres)
1072 yield {ok: True {}, pubkey: pubkey, peers: peers}
1097 yield {ok: True {}, pubkey: pubkey, peers: peers}
1073 when (removed.ok && removed.pubkey != "" && count removed.peers > 0 && federationEnabled)
1098 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))
1075 yield (respond {ok: removed.ok, authorized: removed.ok})
1100 yield (respond {ok: removed.ok, authorized: removed.ok})
1077-- ============================================
1102-- ============================================
@@ -1111,10 +1136,9 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
1111 Just {value: s} -> do
1136 Just {value: s} -> do
1112 acquired <- atomic do
1137 acquired <- atomic do
1113 aps <- *activePolls
1138 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
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)
1118 yield True
1142 yield True
1119 if not acquired
1143 if not acquired
1120 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: []})
1144 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 -
1131 else yield []
1155 else yield []
1132 drains <- atomic do
1156 drains <- atomic do
1133 se <- *senderExpirations
1157 se <- *senderExpirations
1134− let myExp = pluck (filter se (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})
1158+ let myExp = mapWhere 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)
1159+ pruneRef senderExpirations (\x -> x.senderKey != s.pubkey)
1136 sd <- *senderDeliveries
1160 sd <- *senderDeliveries
1137− let myDel = pluck (filter sd (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})
1161+ let myDel = mapWhere 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)
1162+ pruneRef senderDeliveries (\x -> x.senderKey != s.pubkey)
1139 cn <- *cancelNotifications
1163 cn <- *cancelNotifications
1140− let myCancels = pluck (filter cn (\x -> x.toKey == s.pubkey)) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1164+ let myCancels = mapWhere cn (\x -> x.toKey == s.pubkey) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})
1141− set *cancelNotifications = filter cn (\x -> x.toKey != s.pubkey)
1165+ pruneRef cancelNotifications (\x -> x.toKey != s.pubkey)
1142 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
1166 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}
1143 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
1167 let isHeartbeat = pollResult.authorized && count pollResult.events == 0
1144 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})
1168 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
1171 let toAck = filter msgs matches
1195 let toAck = filter msgs matches
1172 if count toAck == 0 then yield {pendingAcks: []}
1196 if count toAck == 0 then yield {pendingAcks: []}
1174− set *messages = filter msgs (\m -> not (matches m))
1198+ set *messages = removeWhere msgs matches
1176 let mids = pluck toAck (\m -> m.messageId)
1199 let mids = pluck toAck (\m -> m.messageId)
1177− set *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))
1200+ pruneRef forwards (\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})
1201+ ackedFederated <- recordAcksAndDeliveries toAck t
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)
1185 yield {pendingAcks: ackedFederated}
1202 yield {pendingAcks: ackedFederated}
1187-- ============================================
1204-- ============================================
@@ -1249,8 +1266,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
1249 forwardsEnqueued <- if federationEnabled && count remoteServers > 0
1266 forwardsEnqueued <- if federationEnabled && count remoteServers > 0
1251 fwds <- *forwards
1268 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})
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})
1253− set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)
1270+ appendCapped forwards newFwds maxForwardsEntries (\f -> f.createdAt)
1254 yield (count newFwds > 0)
1271 yield (count newFwds > 0)
1255 else yield False
1272 else yield False
1256 let status = if localOnline then MsDelivered {}
1273 let status = if localOnline then MsDelivered {}
@@ -1259,10 +1276,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
1259 yield {status: status, remoteServers: remoteServers}
1276 yield {status: status, remoteServers: remoteServers}
1261forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1278forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1262− forEach servers (\s -> s) (\server -> do
1279+ forEach servers id (\server -> do
1263 info <- atomic do
1280 info <- atomic do
1264 fwds <- *forwards
1281 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
1266 Nothing {} -> yield Nothing {}
1283 Nothing {} -> yield Nothing {}
1267 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1284 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1268 case info of
1285 case info of
@@ -1281,9 +1298,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1282 if r.ok then do
1299 if r.ok then do
1283 clearFailure server
1300 clearFailure server
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)))
1287 else yield {})
1302 else yield {})
1289-- ============================================
1304-- ============================================
@@ -1350,15 +1365,9 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
1350 let toAck = filter msgs matches
1365 let toAck = filter msgs matches
1351 if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}
1366 if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}
1353− set *messages = filter msgs (\m -> not (matches m))
1368+ set *messages = removeWhere msgs matches
1355− set *forwards = filter fwds (\f -> not (matches f))
1369+ pruneRef forwards (\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})
1370+ ackedFed <- recordAcksAndDeliveries toAck t
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}
1371 yield {ackedCount: count toAck, pendingAcks: ackedFed}
1363 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1372 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1364 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
1373 yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))
@@ -1374,8 +1383,7 @@ handleListMessages = \offset authorization clientIp respond -> do
1374 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1383 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1375 let total = count myMsgs
1384 let total = count myMsgs
1376 let effectiveOffset = clamp 0 total offset
1385 let effectiveOffset = clamp 0 total offset
1377− -- Drop the first `effectiveOffset` items via fold (O(n)).
1386+ let after = mapWhere (withIndex myMsgs) (\p -> p.i >= effectiveOffset) (\p -> p.v)
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})
1387 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: ""}))
1388 yield (respond {messages: result, authorized: True {}, count: total, error: ""}))
@@ -1393,12 +1401,10 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
1393 let toDel = filter msgs matches
1401 let toDel = filter msgs matches
1394 if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}
1402 if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}
1396− set *messages = filter msgs (\m -> not (matches m))
1404+ set *messages = removeWhere msgs matches
1397− fwds <- *forwards
1405+ pruneRef forwards (\f -> not (matches f))
1398− set *forwards = filter fwds (\f -> not (matches f))
1406+ let ackedFed = mapWhere toDel (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1399− let ackedFed = pluck (filter toDel (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})
1401− set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)
1407+ appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
1402 yield {deletedCount: count toDel, pendingAcks: ackedFed}
1408 yield {deletedCount: count toDel, pendingAcks: ackedFed}
1403 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1409 fork (when federationEnabled (sendAcksAndClean result.pendingAcks))
1404 yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
1410 yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))
@@ -1413,27 +1419,26 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
1413 let matches = \x -> x.messageId == messageId && x.toKey == to
1419 let matches = \x -> x.messageId == messageId && x.toKey == to
1414 result <- atomic do
1420 result <- atomic do
1415 msgs <- *messages
1421 msgs <- *messages
1416− case headOf (filter msgs matches) of
1422+ case findFirst msgs matches of
1417 Nothing {} -> yield {cancelled: False {}, cancelServers: []}
1423 Nothing {} -> yield {cancelled: False {}, cancelServers: []}
1418 Just {value: msg} ->
1424 Just {value: msg} ->
1419 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1425 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1420 then yield {cancelled: False {}, cancelServers: []}
1426 then yield {cancelled: False {}, cancelServers: []}
1422− set *messages = filter msgs (\m -> not (matches m))
1428+ set *messages = removeWhere msgs matches
1423 fwds <- *forwards
1429 fwds <- *forwards
1424 let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)
1430 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))
1426 pres <- *presence
1432 pres <- *presence
1427 let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)
1433 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)
1430 yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
1435 yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}
1431 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1436 when (result.cancelled && count result.cancelServers > 0 && federationEnabled)
1432 (fork (sendCancelToEach messageId to result.cancelServers))
1437 (fork (sendCancelToEach messageId to result.cancelServers))
1433 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1438 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
1435sendCancelToEach = \messageId to servers ->
1440sendCancelToEach = \messageId to servers ->
1436− forEach servers (\s -> s) (\server -> do
1441+ forEach servers id (\server -> do
1437 t <- now
1442 t <- now
1438 auth <- generateFederationAuth ("cancel:" ++ messageId) t
1443 auth <- generateFederationAuth ("cancel:" ++ messageId) t
1439 result <- fetch (federationUrl server) (FederationCancel {
1444 result <- fetch (federationUrl server) (FederationCancel {
@@ -1441,9 +1446,8 @@ sendCancelToEach = \messageId to servers ->
1441 requestTimestamp: t, correlationId: "", authorization: auth,
1446 requestTimestamp: t, correlationId: "", authorization: auth,
1442 protocolVersion: federationProtocolVersion})
1447 protocolVersion: federationProtocolVersion})
1443 let failed = case result of Err _ -> True {}; Ok r -> not r.ok
1448 let failed = case result of Err _ -> True {}; Ok r -> not r.ok
1444− when failed (atomic do
1449+ when failed (atomic
1445− pfc <- *pendingFederationCancels
1446− set *pendingFederationCancels = capList (union pfc [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}]) 10000 (\c -> c.createdAt)))
1450+ (appendCapped pendingFederationCancels [{messageId: messageId, toKey: to, toServer: server, createdAt: t, retries: 0}] 10000 (\c -> c.createdAt))))
1448handleCancelBatch = \cancels authorization clientIp respond -> do
1452handleCancelBatch = \cancels authorization clientIp respond -> do
1449 t <- now
1453 t <- now
@@ -1456,17 +1460,15 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
1456 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
1460 let matches = \x -> x.messageId == e.messageId && x.toKey == e.to
1457 r <- atomic do
1461 r <- atomic do
1458 msgs <- *messages
1462 msgs <- *messages
1459− case headOf (filter msgs matches) of
1463+ case findFirst msgs matches of
1460 Nothing {} -> yield False
1464 Nothing {} -> yield False
1461 Just {value: msg} ->
1465 Just {value: msg} ->
1462 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1466 if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""
1463 then yield False
1467 then yield False
1465− set *messages = filter msgs (\m -> not (matches m))
1469+ set *messages = removeWhere msgs matches
1467− set *forwards = filter fwds (\f -> not (matches f))
1470+ pruneRef forwards (\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)
1471+ appendCapped cancelNotifications [{toKey: e.to, messageId: e.messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
1470 yield True
1472 yield True
1471 yield {messageId: e.messageId, cancelled: r})
1473 yield {messageId: e.messageId, cancelled: r})
1472 yield (respond {results: results, authorized: True {}, error: ""}))
1474 yield (respond {results: results, authorized: True {}, error: ""}))
@@ -1475,7 +1477,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
1475 t <- now
1477 t <- now
1476 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1478 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1477 gateAuth authorization clientIp t respond mkErr (\senderKey -> do
1479 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
1479 effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1481 effMsgId <- if m.messageId != "" then yield m.messageId else genToken
1480 let clampedTs = clampTimestamp m.timestamp t
1482 let clampedTs = clampTimestamp m.timestamp t
1481 let clampedPri = clamp 0 maxMessagePriority m.priority
1483 let clampedPri = clamp 0 maxMessagePriority m.priority
@@ -1526,11 +1528,11 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
1526 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
1528 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}
1527 let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
1529 let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}
1528 let isOffline = \v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}
1530 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})
1531+ let online = mapWhere 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})
1532+ let offline = mapWhere 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))
1533 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)
1534 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})
1534 yield {newlyOnline: newlyOnline}
1536 yield {newlyOnline: newlyOnline}
1535 when (count valid.newlyOnline > 0)
1537 when (count valid.newlyOnline > 0)
1536 (fork (forwardForNewlyOnline valid.newlyOnline t))
1538 (fork (forwardForNewlyOnline valid.newlyOnline t))
@@ -1588,7 +1590,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
1588 fwds <- *forwards
1590 fwds <- *forwards
1589 let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
1591 let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId
1590 let removed = filter fwds isAcked
1592 let removed = filter fwds isAcked
1591− set *forwards = filter fwds (\f -> not (isAcked f))
1593+ set *forwards = removeWhere fwds isAcked
1592 -- Reclaim origin messages with no remaining forwards
1594 -- Reclaim origin messages with no remaining forwards
1593 msgs <- *messages
1595 msgs <- *messages
1594 let removedIds = pluck removed (\f -> f.messageId)
1596 let removedIds = pluck removed (\f -> f.messageId)
@@ -1599,7 +1601,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
1599 && t - m.receivedAt >= ackGracePeriod
1601 && t - m.receivedAt >= ackGracePeriod
1600 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
1602 && 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))
1603 && 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))
1603 let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})
1605 let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})
1604 yield {ackedCount: count removed, results: perResult}
1606 yield {ackedCount: count removed, results: perResult}
1605 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
1607 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
@@ -1611,14 +1613,12 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio
1611 let matches = \x -> x.messageId == messageId && x.toKey == toKey
1613 let matches = \x -> x.messageId == messageId && x.toKey == toKey
1612 cancelled <- atomic do
1614 cancelled <- atomic do
1613 msgs <- *messages
1615 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
1615 Nothing {} -> yield False
1617 Nothing {} -> yield False
1616 Just {} -> do
1618 Just {} -> do
1617− set *messages = filter msgs (\m -> not (matches m))
1619+ set *messages = removeWhere msgs matches
1619− set *forwards = filter fwds (\f -> not (matches f))
1620+ pruneRef forwards (\f -> not (matches f))
1620− cn <- *cancelNotifications
1621− set *cancelNotifications = capList (union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)
1621+ appendCapped cancelNotifications [{toKey: toKey, messageId: messageId, cancelledAt: t}] 10000 (\c -> c.cancelledAt)
1622 yield True
1622 yield True
1623 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
1623 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
@@ -1630,12 +1630,11 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId
1630 let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
1630 let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey
1631 cancelled <- atomic do
1631 cancelled <- atomic do
1632 msgs <- *messages
1632 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
1634 Nothing {} -> yield False
1634 Nothing {} -> yield False
1635 Just {} -> do
1635 Just {} -> do
1636− set *messages = filter msgs (\m -> not (matches m))
1636+ set *messages = removeWhere msgs matches
1637− fwds <- *forwards
1637+ pruneRef forwards (\f -> not (matches f))
1638− set *forwards = filter fwds (\f -> not (matches f))
1639 yield True
1638 yield True
1640 yield {messageId: e.messageId, cancelled: cancelled})
1639 yield {messageId: e.messageId, cancelled: cancelled})
1641 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1640 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
@@ -1684,13 +1683,12 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
1684 let pairs = zipPairs messageIds toKeys
1683 let pairs = zipPairs messageIds toKeys
1685 atomic do
1684 atomic do
1686 msgs <- *messages
1685 msgs <- *messages
1687− sd <- *senderDeliveries
1688 let receipts = do
1686 let receipts = do
1689 p <- pairs
1687 p <- pairs
1690 m <- msgs
1688 m <- msgs
1691 where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1689 where m.messageId == p.messageId && m.toKey == p.toKey && m.senderKey != "" && m.originServer == ""
1692 yield {senderKey: m.senderKey, messageId: p.messageId, toKey: p.toKey, deliveredAt: t}
1690 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)
1694 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1692 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
1696-- Pair up parallel lists by index (O(n*m) join on i; trivial code).
1694-- Pair up parallel lists by index (O(n*m) join on i; trivial code).
@@ -1719,7 +1717,7 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization
1719 let peers = knownPeerServersFromPresence pres
1717 let peers = knownPeerServersFromPresence pres
1720 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}
1718 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}
1721 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)
1719 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))
1723 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
1721 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))
1725handleUndrain = \authorization respond -> withMaintAuth authorization
1723handleUndrain = \authorization respond -> withMaintAuth authorization
@@ -1803,7 +1801,7 @@ handleAdminListSessions = \pubkey authorization respond -> withMaintAuth authori
1803 (yield (respond {sessions: [], count: 0, ok: False {}}))
1801 (yield (respond {sessions: [], count: 0, ok: False {}}))
1805 ses <- *sessions
1803 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})
1807 yield (respond {sessions: mine, count: count mine, ok: True {}}))
1805 yield (respond {sessions: mine, count: count mine, ok: True {}}))
1809handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization
1807handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization
@@ -1825,9 +1823,8 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
1825 let mine = filter msgs (\m -> m.toKey == pubkey)
1823 let mine = filter msgs (\m -> m.toKey == pubkey)
1826 pruneRef messages (\m -> m.toKey != pubkey)
1824 pruneRef messages (\m -> m.toKey != pubkey)
1827 pruneRef forwards (\f -> f.toKey != pubkey)
1825 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})
1826+ let federated = mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
1830− set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)
1827+ appendCapped ackedIds federated maxAckedIds (\a -> a.ackedAt)
1831 yield (count mine)
1828 yield (count mine)
1832 yield (respond {ok: True {}, purgedCount: purged}))
1829 yield (respond {ok: True {}, purgedCount: purged}))
@@ -1837,9 +1834,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
1838 result <- atomic do
1835 result <- atomic do
1839 ses <- *sessions
1836 ses <- *sessions
1840− let mine = filter ses (\s -> s.ip == ip)
1837+ let myKeys = mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey)
1841 pruneRef sessions (\s -> s.ip != ip)
1838 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)}
1843 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
1840 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))
1845handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
1842handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
@@ -1852,7 +1849,7 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho
1852 let mine = filter active (\s -> s.pubkey == pk) in
1849 let mine = filter active (\s -> s.pubkey == pk) in
1853 {pubkey: pk,
1850 {pubkey: pk,
1854 count: count mine,
1851 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)),
1856 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
1853 latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})
1857 let sorted = sortBy (\g -> 0 - g.count) grouped
1854 let sorted = sortBy (\g -> 0 - g.count) grouped
1858 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit
1855 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit
@@ -1935,7 +1932,7 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization
1935 let myFwds = filter fwds (\x -> x.toKey == pubkey)
1932 let myFwds = filter fwds (\x -> x.toKey == pubkey)
1936 yield (respond {
1933 yield (respond {
1937 messages: countWhere msgs (\x -> x.toKey == pubkey),
1934 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}),
1939 sessions: countWhere ses (\x -> x.pubkey == pubkey),
1936 sessions: countWhere ses (\x -> x.pubkey == pubkey),
1940 forwards: count myFwds,
1937 forwards: count myFwds,
1941 forwardDetails: pluck myFwds (\x -> {server: x.toServer, retries: x.retries, createdAt: x.createdAt}),
1938 forwardDetails: pluck myFwds (\x -> {server: x.toServer, retries: x.retries, createdAt: x.createdAt}),