@@ -74,7 +74,7 @@ type PollResult = PollUnauthorized {} | PollAuthorized {events: [PollEvent], cur
7474type VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [Text]}
7575type SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [Text], fromKey: PubkeyHex}
7676type AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
77−type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: Text}
77+type LookupResult = LookupUnauthorized {} | LookupDone {foundServer: OptionalServerName}
7878
7979
8080-- ============================================
@@ -151,6 +151,10 @@ route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
151151any = \xs pred -> fold (\acc x -> if acc then acc else pred x) (False {}) xs
152152none = \xs pred -> not (any xs pred)
153153when = \cond action -> if cond then action else yield {}
154+whenOk = \result action -> case result of
155+ Ok _ -> action
156+ Err _ -> yield {}
157+clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x
154158headOf = \items -> fold (\acc x -> case acc of
155159 Nothing {} -> Just {value: x}
156160 _ -> acc) Nothing {} items
@@ -204,11 +208,6 @@ isAllowedServerChar = \c ->
204208 || c == "m" || c == "n" || c == "o" || c == "p" || c == "q" || c == "r"
205209 || c == "s" || c == "t" || c == "u" || c == "v" || c == "w" || c == "x"
206210 || c == "y" || c == "z"
207− || c == "A" || c == "B" || c == "C" || c == "D" || c == "E" || c == "F"
208− || c == "G" || c == "H" || c == "I" || c == "J" || c == "K" || c == "L"
209− || c == "M" || c == "N" || c == "O" || c == "P" || c == "Q" || c == "R"
210− || c == "S" || c == "T" || c == "U" || c == "V" || c == "W" || c == "X"
211− || c == "Y" || c == "Z"
212211
213212containsBadUrlChar = \s ->
214213 if s == "" then False {}
@@ -373,9 +372,9 @@ authenticateSession = \sessions token t ->
373372 Just {value: s} -> Authorized {pubkey: s.pubkey}
374373
375374extractToken = \authHeader ->
376− if toLower (take 7 authHeader) == "bearer "
377− && not (tooLong (drop 7 authHeader) maxTokenLen)
378− then drop 7 authHeader
375+ let token = drop 7 authHeader in
376+ if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
377+ then token
379378 else ""
380379
381380verifySecret = \token secret ->
@@ -388,9 +387,9 @@ verifyMaintenanceAuth = \authorization ->
388387 let token = extractToken authorization in
389388 verifySecret token federationSecret || verifySecret token maintenanceSecret
390389
391−clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
390+clampTtl = clamp 1 maxPresenceTtl
392391
393−clampTimestamp = \timestamp t -> if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
392+clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
394393
395394nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
396395
@@ -500,8 +499,7 @@ removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
500499countMessagesFor = \messages toKey t ->
501500 count (do
502501 m <- messages
503− where m.toKey == toKey
504− where m.receivedAt >= t - messageMaxAge
502+ where m.toKey == toKey && m.receivedAt >= t - messageMaxAge
505503 yield m)
506504
507505
@@ -595,7 +593,8 @@ sendBatchedGossip = \peerServers allEvents ->
595593broadcastOfflineGossip = \t peerServers expiredLocal -> do
596594 let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
597595 p <- expiredLocal
598− yield {pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else if rawTtl > maxPresenceTtl then maxPresenceTtl else rawTtl), expiresAt: p.expiresAt})
596+ let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
597+ yield {pubkey: p.pubkey, ttl: ttl, expiresAt: p.expiresAt})
599598 when (federationEnabled && count expiredEntries > 0 && count peerServers > 0) (do
600599 let gossipEvents = do
601600 e <- expiredEntries
@@ -605,6 +604,10 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
605604forkBroadcastOffline = \t result ->
606605 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
607606
607+maybePruneAndBroadcast = \t -> do
608+ pruneResult <- atomic do maybePrune t
609+ forkBroadcastOffline t pruneResult
610+
608611removeForward = \mid srv -> atomic do
609612 forwards <- *forwards
610613 set *forwards = do
@@ -624,9 +627,7 @@ retryForwardsForServer = \entries remaining ->
624627 toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob,
625628 timestamp: f.timestamp, originServer: serverHostN, messageId: f.messageId,
626629 authorization: "Bearer " ++ federationSecret})
627− case result of
628− Ok _ -> removeForward f.messageId f.toServer
629− Err _ -> yield {}
630+ whenOk result (removeForward f.messageId f.toServer)
630631 retryForwardsForServer rest (remaining - 1)
631632
632633retryPendingForwards : [ForwardEntry] -> IO {network} {}
@@ -643,15 +644,12 @@ sendBatchedAcksToServer = \serverAcks server ->
643644 authorization: "Bearer " ++ federationSecret,
644645 messageIds: (do a <- batch; yield a.messageId),
645646 originServer: server, fromServer: serverHostN})
646− case result of
647− Ok _ ->
648− atomic do
649− ackedIds <- *ackedIds
650− set *ackedIds = do
651− a <- ackedIds
652− where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
653− yield a
654− Err _ -> yield {})
647+ whenOk result (atomic do
648+ ackedIds <- *ackedIds
649+ set *ackedIds = do
650+ a <- ackedIds
651+ where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
652+ yield a))
655653
656654sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> [ServerName] -> IO {network} {}
657655sendAcksAndClean = \pendingAcks servers ->
@@ -685,12 +683,10 @@ pollOnce = \token cursor -> do
685683 seqCounter <- *seqCounter
686684 messages <- *messages
687685 let cursorCeil = if seqCounter > t then seqCounter else t
688− let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
686+ let clampedCursor = clamp 0 cursorCeil cursor
689687 let matchingMsgs = do
690688 m <- messages
691− where m.toKey == pubkey
692− where m.receivedAt >= t - messageMaxAge
693− where m.receivedAt > clampedCursor
689+ where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > clampedCursor
694690 yield m
695691 let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
696692 let evts = do
@@ -771,31 +767,32 @@ serve = \req -> case req of
771767 then collectPeerServers updatedPresence (\p -> p.server)
772768 else []
773769 yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}
774− fork (case result.verify of
775− VerifyOk {peerServers} -> do
776− when federationEnabled
777− (sendToEach peerServers (\server -> do
778− _ <- fetch (federationScheme ++ server) (RecvGossip {
779− authorization: "Bearer " ++ federationSecret,
780− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
781− yield {}))
782− broadcastOfflineGossip t result.pr.peerServers (do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p)
783− VerifyFailed {} ->
784− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
770+ fork (do
771+ case result.verify of
772+ VerifyOk {peerServers} ->
773+ when federationEnabled
774+ (sendToEach peerServers (\server -> do
775+ _ <- fetch (federationScheme ++ server) (RecvGossip {
776+ authorization: "Bearer " ++ federationSecret,
777+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
778+ yield {}))
779+ _ -> yield {}
780+ let expiredLocal = case result.verify of
781+ VerifyOk _ -> do p <- result.pr.expiredLocal; where not (p.pubkey == pubkey); yield p
782+ _ -> result.pr.expiredLocal
783+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)
785784 yield (case result.verify of
786785 VerifyFailed _ -> respond {token: "", expiresAt: 0}
787786 VerifyOk _ -> respond {token: token, expiresAt: expiry})
788787
789788 Poll {cursor, timeout, authorization, respond} -> do
790789 t <- now
791− let clampedTimeout = if timeout < 0 then 0 else if timeout > 60000 then 60000 else timeout
790+ let clampedTimeout = clamp 0 60000 timeout
792791 let deadline = t + clampedTimeout
793792 let token = extractToken authorization
794793 result <- pollLoop token cursor deadline
795794 tAfterPoll <- now
796− pruneResult <- atomic do
797− maybePrune tAfterPoll
798− forkBroadcastOffline tAfterPoll pruneResult
795+ maybePruneAndBroadcast tAfterPoll
799796 yield (case result of
800797 PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}
801798 PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})
@@ -825,7 +822,7 @@ serve = \req -> case req of
825822 set *messages = union messages
826823 [{messageId: msgId, toKey: to, fromKey: fromKey,
827824 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
828− let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
825+ let toPresence = do p <- presence; where p.pubkey == to && p.expiresAt > t; yield p
829826 let localOnline = any toPresence (\p -> p.server == serverHostN)
830827 let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
831828 forwards <- *forwards
@@ -847,13 +844,10 @@ serve = \req -> case req of
847844 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
848845 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
849846 authorization: "Bearer " ++ federationSecret})
850− case fwdResult of
851− Ok _ -> removeForward msgId server
852− Err _ -> yield {})
847+ whenOk fwdResult (removeForward msgId server))
853848 when (status == "queued")
854849 (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")))
855− SendUnauthorized {} -> yield {}
856− SendRejected {} -> yield {}
850+ _ -> yield {}
857851 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
858852 yield (case result.send of
859853 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
@@ -923,7 +917,7 @@ serve = \req -> case req of
923917 when (dropped > 0)
924918 (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
925919 _ -> yield {}))
926− AckUnauthorized {} -> yield {}
920+ _ -> yield {}
927921 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
928922 yield (case result.ack of
929923 AckUnauthorized _ -> respond {ok: False {}}
@@ -943,12 +937,13 @@ serve = \req -> case req of
943937 presence <- *presence
944938 let valid = do
945939 p <- presence
946− where p.pubkey == pubkey
947− where p.expiresAt > t
940+ where p.pubkey == pubkey && p.expiresAt > t
948941 yield p
949− let foundServer = if any valid (\p -> p.server == serverHostN)
950− then serverHostN
951− else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
942+ let foundServer = (fold (\acc p ->
943+ if acc.server == serverHostN then acc
944+ else if p.server == serverHostN then {server: serverHostN, best: p.expiresAt}
945+ else if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt}
946+ else acc) {server: "", best: 0} valid).server
952947 yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}
953948 forkBroadcastOffline t result.pr
954949 yield (case result.lookup of
@@ -962,8 +957,7 @@ serve = \req -> case req of
962957 let fromServer = normalizeServerName fromServer
963958 if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents
964959 then do
965− pruneResult <- atomic do maybePrune t
966− forkBroadcastOffline t pruneResult
960+ maybePruneAndBroadcast t
967961 yield (respond {ok: False {}})
968962 else do
969963 let allValidEntries = validGossipEvents events t