@@ -638,14 +638,6 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
638638 if matchIp clientIp ip then onAuth pubkey else onUnauth
639639 _ -> onUnauth
640640
641−splitOnColon : Text -> {before: Text, after: Text}
642−splitOnColon = \s ->
643− if s == "" then {before: "", after: ""}
644− else if take 1 s == ":" then {before: "", after: drop 1 s}
645− else
646− let r = splitOnColon (drop 1 s) in
647− {before: take 1 s ++ r.before, after: r.after}
648−
649641generateFederationAuth : Text -> Timestamp -> IO {network} Text
650642generateFederationAuth = \scope requestTimestamp ->
651643 if serverFederationPrivateKeyHex != ""
@@ -669,14 +661,18 @@ verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
669661 let payload = drop 7 authorization
670662 if payload == "" || tooLong payload (maxSignatureLen + 25) then yield False
671663 else do
672− let parts = splitOnColon payload
664+ -- Split payload on first ':' into {sig, nonce} via a fold over chars.
665+ let parts = fold (\acc c ->
666+ if acc.split then {sig: acc.sig, nonce: acc.nonce ++ c, split: True {}}
667+ else if c == ":" then {sig: acc.sig, nonce: "", split: True {}}
668+ else {sig: acc.sig ++ c, nonce: "", split: False {}}) {sig: "", nonce: "", split: False {}} (chars payload)
673669 result <- fetch (federationUrl fromServer) (FederationPubkey {})
674670 case result of
675671 Ok r ->
676672 if not r.ok || r.pubkey == "" then yield False
677673 else
678− let msg = scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if parts.after == "" then "" else "|" ++ parts.after) in
679− yield (verify (bytesFromHex r.pubkey) (textToBytes msg) (bytesFromHex parts.before))
674+ let msg = scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if parts.nonce == "" then "" else "|" ++ parts.nonce) in
675+ yield (verify (bytesFromHex r.pubkey) (textToBytes msg) (bytesFromHex parts.sig))
680676 _ -> yield False
681677 else case extractToken authorization of
682678 Just {value: token} -> do
@@ -749,23 +745,18 @@ checkPubkeyRateGate = \rateRef pubkey maxRate windowMs t -> atomic
749745checkServerRateGate = \server t ->
750746 checkPubkeyRateGate federationRates server maxFederationRatePerServer federationRateWindowMs t
751747
752−checkPreAuth = \clientIp t -> do
748+-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
749+authPreCheck = \clientIp t -> do
753750 dp <- *drainPhase
754751 g <- checkGlobalRate t
755752 i <- checkIpRateGate clientIp t
753+ a <- checkAuthGlobalRate t
756754 yield (if dp >= 1 then "draining"
757755 else if g then "global_rate_limited"
758756 else if i then "ip_rate_limited"
757+ else if a then "auth_rate_limited"
759758 else "")
760759
761−-- Pre-auth check + auth-global rate. Returns "" when ok, else an error code.
762−authPreCheck = \clientIp t -> do
763− p <- checkPreAuth clientIp t
764− if p != "" then yield p
765− else do
766− a <- checkAuthGlobalRate t
767− yield (if a then "auth_rate_limited" else "")
768−
769760-- ============================================
770761-- Federation failure tracking (simple, no circuit breaker)
771762-- ============================================
@@ -819,7 +810,7 @@ bestServerForKey = \entries ->
819810 if any entries (\p -> isLocalServer p.server) then serverHostN
820811 else case maxBy entries (\p -> p.expiresAt) of
821812 Just {value: p} -> p.server
822− Nothing {} -> ""
813+ _ -> ""
823814
824815hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
825816hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
@@ -873,15 +864,13 @@ mkGossip = \etype pk t ttl ->
873864-- Build an acked entry for a federated message.
874865mkAckedEntry = \m t -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}
875866
876−-- Build a sender-delivery entry for a locally-originated message.
877−mkDeliveryEntry = \m t -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}
878−
879867-- Record federation acks and sender deliveries for a batch of just-removed messages.
880868-- Returns the federated-ack entries (the caller flushes them remotely later).
881869recordAcksAndDeliveries = \toAck t -> do
882870 let ackedFed = mapWhere toAck (\m -> m.originServer != "") (\m -> mkAckedEntry m t)
883871 appendCapped ackedIds ackedFed maxAckedIds (\a -> a.ackedAt)
884− let localDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "") (\m -> mkDeliveryEntry m t)
872+ let localDel = mapWhere toAck (\m -> m.senderKey != "" && m.originServer == "")
873+ (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})
885874 appendCapped senderDeliveries localDel 10000 (\d -> d.deliveredAt)
886875 yield ackedFed
887876
@@ -927,13 +916,17 @@ retryPendingForwards = do
927916 expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
928917 let success = case result of Ok r -> r.ok; _ -> False {}
929918 case result of Err _ -> recordFailure server; Ok _ -> when success (clearFailure server)
919+ let isMatch = \x -> x.messageId == f.messageId && x.toServer == server
930920 atomic do
931− cur <- *forwards
932− set *forwards = do
933− x <- cur
934− where x.messageId != f.messageId || x.toServer != server || (not success && x.retries + 1 < maxForwardRetries)
935− yield (if success || x.messageId != f.messageId || x.toServer != server then x
936− 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})))
921+ if success then pruneRef forwards (\x -> not (isMatch x))
922+ else do
923+ cur <- *forwards
924+ set *forwards = do
925+ x <- cur
926+ where not (isMatch x) || x.retries + 1 < maxForwardRetries
927+ yield (if isMatch x
928+ then {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}
929+ else x)))
937930
938931sendAcksAndClean = \pendingAcks ->
939932 forEach (collectPeers pendingAcks (\a -> a.originServer)) id (\server -> do
@@ -1000,28 +993,25 @@ refreshLocalPresence = \t -> do
1000993handleChallenge = \pubkey clientIp respond -> do
1001994 t <- now
1002995 let mkResp = \err -> respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: err}
1003− if requireClientIp && clientIp == "" then yield (mkResp "client_ip_required")
996+ preErr <- if requireClientIp && clientIp == "" then yield "client_ip_required" else authPreCheck clientIp t
997+ if preErr != "" then yield (mkResp preErr)
1004998 else do
1005− preErr <- authPreCheck clientIp t
1006− if preErr != "" then yield (mkResp preErr)
1007− else do
1008− challenge <- genToken
1009− result <- atomic do
1010− ch <- *challenges
1011− let active = filter ch (\c -> c.expiresAt > t)
1012− let cappedActive = if countWhere active (\c -> c.pubkey == pubkey) >= maxChallengesPerKey
1013− then evictOldest active (\c -> if c.pubkey == pubkey then c.expiresAt else 999999999999)
1014− else active
1015− if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
1016− else do
1017− rates <- *challengeRates
1018− let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
1019− if r.limited then yield {ok: False {}, error: "rate_limited"}
1020− else do
1021− set *challengeRates = r.entries
1022− set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
1023− yield {ok: True {}, error: ""}
1024− yield (respond {challenge: if result.ok then challenge else "", ok: result.ok, expiresAt: if result.ok then t + challengeExpiry else 0, retryAfterMs: 0, error: result.error})
999+ challenge <- genToken
1000+ result <- atomic do
1001+ ch <- *challenges
1002+ let active = filter ch (\c -> c.expiresAt > t)
1003+ let cappedActive = if countWhere active (\c -> c.pubkey == pubkey) >= maxChallengesPerKey
1004+ then evictOldest active (\c -> if c.pubkey == pubkey then c.expiresAt else 999999999999)
1005+ else active
1006+ rates <- *challengeRates
1007+ let r = checkRate rates pubkey maxChallengeRatePerKey challengeRateWindowMs 5000 t
1008+ if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}
1009+ else if r.limited then yield {ok: False {}, error: "rate_limited"}
1010+ else do
1011+ set *challengeRates = r.entries
1012+ set *challenges = union cappedActive [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry, ip: clientIp}]
1013+ yield {ok: True {}, error: ""}
1014+ yield (respond {challenge: if result.ok then challenge else "", ok: result.ok, expiresAt: if result.ok then t + challengeExpiry else 0, retryAfterMs: 0, error: result.error})
10251015
10261016handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
10271017 t <- now
@@ -1032,29 +1022,24 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->
10321022 token <- genToken
10331023 result <- atomic do
10341024 ch <- *challenges
1025+ dp <- *drainPhase
10351026 let hasMatch = any ch (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && matchIp clientIp c.ip)
1036− if not hasMatch then yield {valid: False {}, peers: []}
1037− else
1038− if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))
1039− then yield {valid: False {}, peers: []}
1040− else do
1041− set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1042− dp <- *drainPhase
1043− if dp >= 1 then yield {valid: False {}, peers: []}
1044− else do
1045− ses <- *sessions
1046− let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses
1047− let forKey = filter kept (\s -> s.pubkey == pubkey)
1048− let afterPerKey = if count forKey >= maxSessionsPerKey
1049− then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))
1050− else kept
1051− let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1052− set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)
1053− pres <- *presence
1054− let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1055− let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1056− set *presence = newPres
1057− yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
1027+ let sigOk = verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
1028+ if dp >= 1 || not hasMatch || not sigOk then yield {valid: False {}, peers: []}
1029+ else do
1030+ set *challenges = removeWhere ch (\c -> c.pubkey == pubkey && c.challenge == challenge)
1031+ ses <- *sessions
1032+ let kept = if revokeOthers then removeWhere ses (\s -> s.pubkey == pubkey) else ses
1033+ let forKey = filter kept (\s -> s.pubkey == pubkey)
1034+ let afterPerKey = if count forKey >= maxSessionsPerKey
1035+ then union (removeWhere kept (\s -> s.pubkey == pubkey)) (evictOldest forKey (\s -> s.expiresAt))
1036+ else kept
1037+ set *sessions = capList (union afterPerKey [{token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}]) maxSessions (\s -> s.expiresAt)
1038+ pres <- *presence
1039+ let withoutLocal = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1040+ let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)
1041+ set *presence = newPres
1042+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres}
10581043 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})
10591044 else do
10601045 when federationEnabled
@@ -1108,18 +1093,16 @@ handleLogout = \authorization clientIp respond -> do
11081093handlePoll = \cursor generation watchKeys authorization clientIp respond -> do
11091094 t <- now
11101095 serverGen <- *serverGeneration
1111− let generationMismatch = generation != serverGen
1112− let effectiveCursor = if generationMismatch then 0 else cursor
1113− let cursorWasReset = generationMismatch && cursor > 0
1114− let cappedWatchKeys = takeRelation maxWatchKeys watchKeys
11151096 dp <- *drainPhase
1097+ g <- checkGlobalRate t
1098+ let cursorWasReset = generation != serverGen && cursor > 0
11161099 if dp >= 2 then yield (respond (emptyPollResp 0 (True {}) (False {}) (True {}) serverGen 0 cursorWasReset))
1117− else do
1118− g <- checkGlobalRate t
1119− if g then yield (respond (emptyPollResp 0 (False {}) (True {}) (False {}) serverGen globalRateWindowMs cursorWasReset))
1120− else case extractToken authorization of
1121− Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1122− Just {value: token} -> pollOnce token effectiveCursor cursorWasReset cappedWatchKeys clientIp serverGen respond t
1100+ else if g then yield (respond (emptyPollResp 0 (False {}) (True {}) (False {}) serverGen globalRateWindowMs cursorWasReset))
1101+ else case extractToken authorization of
1102+ Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1103+ Just {value: token} ->
1104+ let effectiveCursor = if generation != serverGen then 0 else cursor in
1105+ pollOnce token effectiveCursor cursorWasReset (takeRelation maxWatchKeys watchKeys) clientIp serverGen respond t
11231106
11241107pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -> do
11251108 setup <- atomic do
@@ -1170,20 +1153,18 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -
11701153 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})
11711154
11721155doPoll = \s t -> do
1173− initial <- atomic do
1156+ let collectAt = \tnow -> atomic do
11741157 msgs <- *messages
1175− yield (collectPollEvents msgs s.pubkey s.clampedCursor t)
1176− if count initial.events > 0 then yield {authorized: True {}, events: initial.events, cursor: initial.cursor, hasMore: initial.hasMore}
1177− else do
1178− -- Wait for signal or heartbeat
1179− atomic do
1180− wakeup <- *pollWakeup
1181− if wakeup > s.startWakeup then yield {}
1182− else retry
1183− t2 <- now
1184− msgs2 <- *messages
1185− let r = collectPollEvents msgs2 s.pubkey s.clampedCursor t2
1186− yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
1158+ yield (collectPollEvents msgs s.pubkey s.clampedCursor tnow)
1159+ initial <- collectAt t
1160+ r <- if count initial.events > 0 then yield initial
1161+ else do
1162+ atomic do
1163+ wakeup <- *pollWakeup
1164+ if wakeup > s.startWakeup then yield {} else retry
1165+ t2 <- now
1166+ collectAt t2
1167+ yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}
11871168
11881169collectPollEvents = \messages pubkey cursor t ->
11891170 let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in
@@ -1211,26 +1192,23 @@ handleSendMessage = \to encryptedBlob timestamp messageId expiresInMs priority a
12111192 let clampedPriority = clamp 0 maxMessagePriority priority
12121193 t <- now
12131194 dp <- *drainPhase
1195+ g <- checkGlobalRate t
1196+ effMsgId <- if messageId != "" then yield messageId else genToken
12141197 if dp >= 2 then yield (respond (sendErrResp (MsDraining {}) "draining"))
1198+ else if g then yield (respond (sendErrResp (MsRateLimited {}) "rate_limited"))
1199+ else if t - timestamp > messageMaxAge then yield (respond (sendErrResp (MsRejected {}) "stale_timestamp"))
12151200 else do
1216− g <- checkGlobalRate t
1217− if g then yield (respond (sendErrResp (MsRateLimited {}) "rate_limited"))
1218− else do
1219− effMsgId <- if messageId != "" then yield messageId else genToken
1220− if t - timestamp > messageMaxAge
1221− then yield (respond (sendErrResp (MsRejected {}) "stale_timestamp"))
1222− else do
1223− let clampedTs = clampTimestamp timestamp t
1224− result <- atomic do
1225− withSessionAuth authorization t clientIp
1226− (yield {status: MsUnauthorized {}, remoteServers: []})
1227− (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs clampedPriority t)
1228− when (federationEnabled && count result.remoteServers > 0)
1229− (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
1230− let ok = case result.status of
1231− MsDelivered {} -> True {}; MsFederated {} -> True {}
1232− MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}
1233− yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})
1201+ let clampedTs = clampTimestamp timestamp t
1202+ result <- atomic do
1203+ withSessionAuth authorization t clientIp
1204+ (yield {status: MsUnauthorized {}, remoteServers: []})
1205+ (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs effMsgId expiresInMs clampedPriority t)
1206+ when (federationEnabled && count result.remoteServers > 0)
1207+ (fork (forkForwardsTo effMsgId to encryptedBlob clampedTs result.remoteServers))
1208+ let ok = case result.status of
1209+ MsDelivered {} -> True {}; MsFederated {} -> True {}
1210+ MsQueued {} -> True {}; MsDuplicate {} -> True {}; _ -> False {}
1211+ yield (respond {status: result.status, messageId: if ok then effMsgId else "", retryAfterMs: 0, error: if ok then "" else "rejected", loadPct: 0})
12341212
12351213sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msgPriority t -> do
12361214 msgs <- *messages
@@ -1272,14 +1250,10 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12721250
12731251forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12741252 forEach servers id (\server -> do
1275− info <- atomic do
1276− fwds <- *forwards
1277− case findFirst fwds (\f -> f.messageId == effMsgId && f.toServer == server) of
1278− Nothing {} -> yield Nothing {}
1279− Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})
1280− case info of
1253+ fwds <- *forwards
1254+ case findFirst fwds (\f -> f.messageId == effMsgId && f.toServer == server) of
12811255 Nothing {} -> yield {}
1282− Just {value: i} -> do
1256+ Just {value: f} -> do
12831257 t <- now
12841258 auth <- generateFederationAuth ("forward:" ++ effMsgId) t
12851259 result <- fetch (federationUrl server) (ForwardMessage {
@@ -1287,14 +1261,12 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
12871261 timestamp: clampedTs, originServer: serverHostN, messageId: effMsgId,
12881262 fromServer: serverHostN, requestTimestamp: t,
12891263 authorization: auth, correlationId: "",
1290− expiresAt: i.expiresAt, priority: i.priority, protocolVersion: federationProtocolVersion})
1264+ expiresAt: f.expiresAt, priority: f.priority, protocolVersion: federationProtocolVersion})
12911265 case result of
12921266 Err _ -> recordFailure server
1293− Ok r ->
1294− if r.ok then do
1295− clearFailure server
1296− atomic (pruneRef forwards (\f -> not (f.messageId == effMsgId && f.toServer == server)))
1297− else yield {})
1267+ Ok r -> when r.ok (do
1268+ clearFailure server
1269+ atomic (pruneRef forwards (\x -> not (x.messageId == effMsgId && x.toServer == server)))))
12981270
12991271-- ============================================
13001272-- Handler: lookup
@@ -1678,8 +1650,10 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp
16781650 appendCapped senderDeliveries receipts 10000 (\d -> d.deliveredAt)
16791651 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))
16801652
1681−-- Pair up parallel lists by index (O(n*m) join on i; trivial code).
1682−withIndex = \xs -> (fold (\acc x -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: x}]}) {i: 0, out: []} xs).out
1653+-- Index a list (1-based by virtue of `count` of the partial fold).
1654+withIndex = \xs -> fold (\acc x -> union acc [{i: count acc, v: x}]) [] xs
1655+
1656+-- Pair parallel lists by index (O(n*m) cross-filter; trivial code).
16831657zipPairs = \mids keys ->
16841658 do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}
16851659