@@ -655,8 +655,9 @@ pruneAndRetry = \t -> do
655655 forwards <- *forwards
656656 ackedIds <- *ackedIds
657657 let pendingAcks = do a <- ackedIds; where isFederatedOrigin a.originServer; yield a
658− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: forwards, pendingAcks: pendingAcks}
659− broadcastOfflineGossip t result.peerServers result.expiredLocal
658+ yield {pr: pr, pendingForwards: forwards, pendingAcks: pendingAcks}
659+ let pr = result.pr
660+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
660661 when (federationEnabled && count result.pendingForwards > 0)
661662 (retryPendingForwards result.pendingForwards)
662663 when (federationEnabled && count result.pendingAcks > 0) (do
@@ -748,7 +749,7 @@ serve = \req -> case req of
748749 Nothing {} -> False {}
749750 _ -> verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)
750751 if not sigValid
751− then yield {sigValid: False {}, peerServers: [], expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
752+ then yield VerifyFailed {pr: pr}
752753 else do
753754 set *challenges = do
754755 c <- challenges
@@ -768,21 +769,24 @@ serve = \req -> case req of
768769 let peerServers = if federationEnabled
769770 then collectPeerServers updatedPresence (\p -> p.server)
770771 else []
771− yield {sigValid: True {}, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}
772− fork (do
773− when (result.sigValid && federationEnabled)
774− (sendToEach result.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− broadcastOfflineGossip t result.offlinePeers (do
780− p <- result.expiredLocal
781− where not (result.sigValid && p.pubkey == pubkey)
782− yield p))
783− yield (if result.sigValid
784− then respond {token: token, expiresAt: expiry}
785− else respond {token: "", expiresAt: 0})
772+ yield VerifyOk {peerServers: peerServers, pr: pr}
773+ fork (case result of
774+ VerifyFailed {pr} ->
775+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
776+ VerifyOk {peerServers, pr} -> do
777+ when federationEnabled
778+ (sendToEach peerServers (\server -> do
779+ _ <- fetch (federationScheme ++ server) (RecvGossip {
780+ authorization: "Bearer " ++ federationSecret,
781+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})
782+ yield {}))
783+ broadcastOfflineGossip t pr.peerServers (do
784+ p <- pr.expiredLocal
785+ where not (p.pubkey == pubkey)
786+ yield p))
787+ yield (case result of
788+ VerifyFailed _ -> respond {token: "", expiresAt: 0}
789+ VerifyOk _ -> respond {token: token, expiresAt: expiry})
786790
787791 Poll {cursor, timeout, authorization, respond} -> do
788792 t <- now
@@ -808,7 +812,7 @@ serve = \req -> case req of
808812 let auth = authenticateSession sessions (extractToken authorization) t
809813 case auth of
810814 Unauthorized {} ->
811− yield {status: "unauthorized", shouldInsert: False {}, remoteServers: [], fromKey: "", expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
815+ yield SendUnauthorized {pr: pr}
812816 Authorized {pubkey: fromKey} -> do
813817 messages <- *messages
814818 presence <- *presence
@@ -833,21 +837,26 @@ serve = \req -> case req of
833837 set *forwards = if shouldInsert && federationEnabled && count forwardServers > 0
834838 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
835839 else forwards
836− yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
837− fork (do
838− when (count result.remoteServers > 0 && result.shouldInsert && federationEnabled)
839− (sendToEach result.remoteServers (\server -> do
840− fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
841− toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
842− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
843− authorization: "Bearer " ++ federationSecret})
844− case fwdResult of
845− Ok _ -> removeForward msgId server
846− Err _ -> yield {}))
847− when (result.status == "queued" && count result.remoteServers > 0 && result.shouldInsert && federationEnabled)
848− (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
849− broadcastOfflineGossip t result.peerServers result.expiredLocal)
850− yield (respond {status: result.status, messageId: if result.shouldInsert then msgId else ""})
840+ yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
841+ fork (case result of
842+ SendUnauthorized {pr} ->
843+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
844+ SendDone {status, remoteServers, fromKey, pr} -> do
845+ when (count remoteServers > 0 && not (status == "rejected") && federationEnabled)
846+ (sendToEach remoteServers (\server -> do
847+ fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
848+ toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
849+ timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
850+ authorization: "Bearer " ++ federationSecret})
851+ case fwdResult of
852+ Ok _ -> removeForward msgId server
853+ Err _ -> yield {}))
854+ when (status == "queued" && count remoteServers > 0 && federationEnabled)
855+ (println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking"))
856+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
857+ yield (case result of
858+ SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
859+ SendDone {status, remoteServers, fromKey, pr} -> respond {status: status, messageId: if status == "rejected" then "" else msgId})
851860
852861 AckMessages {messageIds, authorization, respond} -> do
853862 t <- now
@@ -858,7 +867,7 @@ serve = \req -> case req of
858867 let auth = authenticateSession sessions (extractToken authorization) t
859868 case auth of
860869 Unauthorized {} ->
861− yield {ok: False {}, pendingAcks: [], expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: []}
870+ yield AckUnauthorized {pr: pr}
862871 Authorized {pubkey} -> do
863872 messages <- *messages
864873 let ackedMessages = do
@@ -880,15 +889,17 @@ serve = \req -> case req of
880889 let evictedEntries = if excessCount <= 0 then []
881890 else diff combinedAckedIds afterEviction
882891 set *ackedIds = afterEviction
883− yield {ok: True {}, pendingAcks: ackedFederated, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
884− let originServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
885− fork (do
886− when result.ok (do
892+ yield AckDone {pendingAcks: ackedFederated, pr: pr, evictedEntries: evictedEntries}
893+ fork (case result of
894+ AckUnauthorized {pr} ->
895+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
896+ AckDone {pendingAcks, pr, evictedEntries} -> do
897+ let originServers = collectPeerServers pendingAcks (\a -> a.originServer)
887898 when (count originServers > 0 && federationEnabled)
888− (sendAcksAndClean result.pendingAcks originServers)
899+ (sendAcksAndClean pendingAcks originServers)
889900 let evictedToRetry = do
890− a <- result.evictedEntries
891− where none result.pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
901+ a <- evictedEntries
902+ where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)
892903 yield a
893904 when (count evictedToRetry > 0 && federationEnabled) (do
894905 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
@@ -910,9 +921,11 @@ serve = \req -> case req of
910921 yield excess
911922 when (dropped > 0)
912923 (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))
913− _ -> yield {})))
914− broadcastOfflineGossip t result.peerServers result.expiredLocal)
915− yield (respond {ok: result.ok})
924+ _ -> yield {}))
925+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
926+ yield (case result of
927+ AckUnauthorized _ -> respond {ok: False {}}
928+ AckDone _ -> respond {ok: True {}})
916929
917930 LookupPresence {pubkey, authorization, respond} -> do
918931 let pubkey = normalizeHex pubkey
@@ -923,7 +936,7 @@ serve = \req -> case req of
923936 let auth = authenticateSession sessions (extractToken authorization) t
924937 case auth of
925938 Unauthorized {} ->
926− yield {foundServer: "", authorized: False {}, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
939+ yield {foundServer: "", authorized: False {}, pr: pr}
927940 Authorized _ -> do
928941 presence <- *presence
929942 let valid = do
@@ -934,8 +947,8 @@ serve = \req -> case req of
934947 let foundServer = if any valid (\p -> p.server == serverHostN)
935948 then serverHostN
936949 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
937− yield {foundServer: foundServer, authorized: True {}, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
938− forkBroadcastOffline t result
950+ yield {foundServer: foundServer, authorized: True {}, pr: pr}
951+ forkBroadcastOffline t result.pr
939952 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
940953
941954 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -998,7 +1011,7 @@ serve = \req -> case req of
9981011 let finalPresence = if validRequest then capResult.finalPresence else presence
9991012 set *presence = finalPresence
10001013 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
1001− yield {appliedKeys: union capResult.appliedOnline appliedOffline, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1014+ yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
10021015 let effectiveEvents = do
10031016 e <- union filteredOnline filteredOffline
10041017 where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
@@ -1016,7 +1029,8 @@ serve = \req -> case req of
10161029 authorization: "Bearer " ++ federationSecret,
10171030 events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
10181031 yield {}))
1019− broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal)
1032+ let pr = gossipResult.pr
1033+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
10201034 yield (respond {ok: validRequest})
10211035
10221036 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
@@ -1047,8 +1061,8 @@ serve = \req -> case req of
10471061 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
10481062 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
10491063 else messages
1050− yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, ok: shouldInsert || alreadyExists}
1051− forkBroadcastOffline t result
1064+ yield {pr: pr, ok: shouldInsert || alreadyExists}
1065+ forkBroadcastOffline t result.pr
10521066 yield (respond {ok: result.ok})
10531067
10541068 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
@@ -1070,7 +1084,7 @@ serve = \req -> case req of
10701084 set *forwards = if validAck
10711085 then diff forwards forwardsToRemove
10721086 else forwards
1073− yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal}
1087+ yield pr
10741088 forkBroadcastOffline t pruneResult
10751089 yield (respond {ok: validAck})
10761090