@@ -725,8 +725,8 @@ chargeRate = \rateRef key maxRate windowMs t -> do
725725checkPubkeyRateGate = \rateRef pubkey maxRate windowMs t -> atomic
726726 (chargeRate rateRef pubkey maxRate windowMs t)
727727
728−checkServerRateGate = \server t ->
729− checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs t
728+checkServerRateGate = \server ->
729+ checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs
730730
731731-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
732732authPreCheck = \clientIp t -> do
@@ -755,8 +755,9 @@ recordFailure = \server -> do
755755 t <- now
756756 atomic do
757757 failures <- *forwardFailures
758− set *forwardFailures = union (removeWhere failures (\f -> f.server == server))
759− [case findFirst failures (\f -> f.server == server) of
758+ let same = \f -> f.server == server
759+ set *forwardFailures = union (removeWhere failures same)
760+ [case findFirst failures same of
760761 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}
761762 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}]
762763
@@ -1126,9 +1127,10 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
11261127 dp <- *drainPhase
11271128 g <- checkGlobalRate t
11281129 effMsgId <- if messageId != "" then yield messageId else genToken
1129− if dp >= 2 then yield (respond (sendErrResp (MsDraining {}) "draining"))
1130− else if g then yield (respond (sendErrResp (MsRateLimited {}) "rate_limited"))
1131− else if t - timestamp > messageMaxAge then yield (respond (sendErrResp (MsRejected {}) "stale_timestamp"))
1130+ let err = \s e -> yield (respond (sendErrResp s e))
1131+ if dp >= 2 then err (MsDraining {}) "draining"
1132+ else if g then err (MsRateLimited {}) "rate_limited"
1133+ else if t - timestamp > messageMaxAge then err (MsRejected {}) "stale_timestamp"
11321134 else do
11331135 let clampedTs = clampTimestamp timestamp t
11341136 result <- atomic do
@@ -1149,11 +1151,12 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11491151 srLim <- chargeRate sendRates senderKey maxSendRatePerKey sendRateWindowMs t
11501152 prLim <- chargeRate pairRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
11511153 rrLim <- chargeRate recvRates to maxRecvRatePerKey recvRateWindowMs t
1152− if senderKey == to then yield {status: MsRejected {}, remoteServers: []}
1153− else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then yield {status: MsDuplicate {}, remoteServers: []}
1154− else if srLim || prLim || rrLim then yield {status: MsRateLimited {}, remoteServers: []}
1155− else if count msgs >= maxMessages then yield {status: MsRejected {}, remoteServers: []}
1156− else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then yield {status: MsQueued {}, remoteServers: []}
1154+ let reject = \s -> yield {status: s, remoteServers: []}
1155+ if senderKey == to then reject (MsRejected {})
1156+ else if hasMessage effMsgId to msgs || hasMessage effMsgId to aids || hasMessage effMsgId to dids then reject (MsDuplicate {})
1157+ else if srLim || prLim || rrLim then reject (MsRateLimited {})
1158+ else if count msgs >= maxMessages then reject (MsRejected {})
1159+ else if countWhere msgs (\m -> m.toKey == to) >= maxMessagesPerRecipient then reject (MsQueued {})
11571160 else do
11581161 pres <- *presence
11591162 seq <- *seqCounter
@@ -1175,7 +1178,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
11751178forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
11761179 forEach servers id (\server -> do
11771180 fwds <- *forwards
1178− case findFirst fwds (\f -> f.messageId == effMsgId && f.toServer == server) of
1181+ let same = \x -> x.messageId == effMsgId && x.toServer == server
1182+ case findFirst fwds same of
11791183 Nothing {} -> yield {}
11801184 Just {value: f} -> do
11811185 t <- now
@@ -1190,7 +1194,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
11901194 Err _ -> recordFailure server
11911195 Ok r -> when r.ok (do
11921196 clearFailure server
1193− atomic (pruneRef forwards (\x -> not (x.messageId == effMsgId && x.toServer == server)))))
1197+ atomic (pruneRef forwards (\x -> not (same x)))))
11941198
11951199-- ============================================
11961200-- Handler: lookup
@@ -1255,9 +1259,7 @@ handleListMessages = \offset authorization clientIp respond -> do
12551259 {messages: [], authorized: True {}, count: 0, error: "rate_limited"} (\pubkey -> do
12561260 msgs <- *messages
12571261 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))
1258− let after = drop (clamp 0 (count myMsgs) offset) myMsgs
1259− let result = pluck (takeRelation maxPollEvents after) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})
1260− yield (respond {messages: result, authorized: True {}, count: count myMsgs, error: ""}))
1262+ yield (respond {messages: pluck (takeRelation maxPollEvents (drop (clamp 0 (count myMsgs) offset) myMsgs)) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}), authorized: True {}, count: count myMsgs, error: ""}))
12611263
12621264handleDeleteMessages = \messageIds authorization clientIp respond -> do
12631265 t <- now
@@ -1345,8 +1347,9 @@ handleVerifyToken = \token scope clientIp respond -> do
13451347 else do
13461348 found <- atomic do
13471349 pending <- *pendingVerifications
1348− let match = any pending (\p -> p.scope == scope && p.token == token && t - p.createdAt < verificationTokenExpiry)
1349− when match (set *pendingVerifications = removeWhere pending (\p -> p.scope == scope && p.token == token))
1350+ let same = \p -> p.scope == scope && p.token == token
1351+ let match = any pending (\p -> same p && t - p.createdAt < verificationTokenExpiry)
1352+ when match (set *pendingVerifications = removeWhere pending same)
13501353 yield match
13511354 yield (respond {ok: found})
13521355
@@ -1424,7 +1427,7 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
14241427 && t - m.receivedAt >= ackGracePeriod
14251428 && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)
14261429 && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))
1427− set *messages = removeWhere msgs (\m -> any toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))
1430+ set *messages = removeWhere msgs (\m -> hasMessage m.messageId m.toKey toReap)
14281431 yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
14291432 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
14301433
@@ -1560,11 +1563,13 @@ handleUnblockPeer = \server authorization respond -> withMaintAuth authorization
15601563 (yield (respond {ok: False {}, unblocked: False {}}))
15611564 (do removed <- removeServerFromList blockedPeers server; yield (respond {ok: True {}, unblocked: removed}))
15621565
1563−handleListBlocked = \authorization respond -> do
1566+handleListPeersResp = \listRef authorization respond -> do
15641567 ok <- verifyMaintenanceAuth authorization
1565− list <- *blockedPeers
1568+ list <- *listRef
15661569 yield (respond {peers: if ok then list else [], ok: ok})
15671570
1571+handleListBlocked = handleListPeersResp blockedPeers
1572+
15681573handleAllowPeer = \server authorization respond -> withMaintAuth authorization
15691574 (yield (respond {ok: False {}, allowed: 0}))
15701575 (do n <- addServerToList allowedPeers server; yield (respond {ok: True {}, allowed: n}))
@@ -1573,10 +1578,7 @@ handleUnallowPeer = \server authorization respond -> withMaintAuth authorization
15731578 (yield (respond {ok: False {}, removed: False {}}))
15741579 (do removed <- removeServerFromList allowedPeers server; yield (respond {ok: True {}, removed: removed}))
15751580
1576−handleListAllowed = \authorization respond -> do
1577− ok <- verifyMaintenanceAuth authorization
1578− list <- *allowedPeers
1579− yield (respond {peers: if ok then list else [], ok: ok})
1581+handleListAllowed = handleListPeersResp allowedPeers
15801582
15811583handlePingPeer = \server authorization respond -> withMaintAuth authorization
15821584 (yield (respond {ok: False {}, latencyMs: 0, error: "unauthorized"}))
@@ -1603,7 +1605,7 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho
16031605 (do
16041606 revoked <- atomic do
16051607 ses <- *sessions
1606− pruneRef sessions (\s -> s.pubkey != pubkey)
1608+ set *sessions = removeWhere ses (\s -> s.pubkey == pubkey)
16071609 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))
16081610 yield (countWhere ses (\s -> s.pubkey == pubkey))
16091611 yield (respond {ok: True {}, revokedCount: revoked}))
@@ -1615,7 +1617,7 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author
16151617 purged <- atomic do
16161618 msgs <- *messages
16171619 let mine = filter msgs (\m -> m.toKey == pubkey)
1618− pruneRef messages (\m -> m.toKey != pubkey)
1620+ set *messages = removeWhere msgs (\m -> m.toKey == pubkey)
16191621 pruneRef forwards (\f -> f.toKey != pubkey)
16201622 appendCapped ackedIds (mapWhere mine (\m -> m.originServer != "") (\m -> mkAckedEntry m t)) maxAckedIds (\a -> a.ackedAt)
16211623 yield (count mine)
@@ -1626,7 +1628,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16261628 (do
16271629 keys <- if ip == "" then yield [] else atomic do
16281630 ses <- *sessions
1629− pruneRef sessions (\s -> s.ip != ip)
1631+ set *sessions = removeWhere ses (\s -> s.ip == ip)
16301632 yield (mapWhere ses (\s -> s.ip == ip) (\s -> s.pubkey))
16311633 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
16321634