@@ -47,6 +47,8 @@ type GossipEventType = Text where \s -> s == "online" || s == "offline"
4747
4848type GossipEvent = {eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}
4949
50+type MessageStatus = Text where \s -> s == "unauthorized" || s == "rejected" || s == "delivered" || s == "federated" || s == "queued"
51+
5052type PubkeyHex = Text where \s -> isValidHex s && not (tooLong s maxPubkeyLen)
5153type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)
5254type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)
@@ -95,7 +97,7 @@ route ClientApi where
9597 = Poll
9698 /messages
9799 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}
98− -> {status: Text, messageId: Text}
100+ -> {status: MessageStatus, messageId: Text}
99101 = SendMessage
100102 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}
101103 -> {ok: Bool}
@@ -133,6 +135,8 @@ isJust = \m -> case m of
133135 Nothing {} -> False {}
134136 _ -> True {}
135137
138+firstOf = \items -> fold (\acc s -> if acc == "" then s else acc) "" items
139+
136140msToS : Int<Ms> -> Int<S>
137141msToS = \x -> x / 1000
138142
@@ -357,6 +361,10 @@ lookupSession = \sessions token t ->
357361 _ -> acc
358362 else acc) Nothing {} sessions
359363
364+sessionPubkey = \session -> case session of
365+ Nothing {} -> ""
366+ Just {value: s} -> s.pubkey
367+
360368extractToken = \authHeader ->
361369 if toLower (take 7 authHeader) == "bearer "
362370 && not (tooLong (drop 7 authHeader) maxTokenLen)
@@ -478,6 +486,7 @@ countMessagesFor = \messages toKey t ->
478486 where m.receivedAt >= t - messageMaxAge
479487 yield m)
480488
489+validGossipEvents : [GossipEvent] -> Bool -> Timestamp -> GossipEventType -> [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}]
481490validGossipEvents = \events validRequest t eventType ->
482491 do
483492 e <- events
@@ -499,7 +508,7 @@ sendToEach = \servers action ->
499508 if count servers == 0
500509 then yield {}
501510 else do
502− let server = fold (\acc s -> if acc == "" then s else acc) "" servers
511+ let server = firstOf servers
503512 let rest = do s <- servers; where not (s == server); yield s
504513 if server == ""
505514 then yield {}
@@ -586,6 +595,16 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do
586595 sendBatchedGossip peerServers gossipEvents
587596 else yield {}
588597
598+forkBroadcastOffline = \t result ->
599+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
600+
601+removeForward = \mid srv -> atomic do
602+ forwards <- *forwards
603+ set *forwards = do
604+ x <- forwards
605+ where not (x.messageId == mid && x.toServer == srv)
606+ yield x
607+
589608retryForwardsForServer : [{encryptedBlob: BlobHex, fromKey: PubkeyHex, messageId: MessageIdHex, timestamp: Timestamp, toKey: PubkeyHex, toServer: ServerName}] -> Int -> IO {network} {}
590609retryForwardsForServer = \entries remaining ->
591610 if count entries == 0 || remaining <= 0
@@ -602,12 +621,7 @@ retryForwardsForServer = \entries remaining ->
602621 authorization: "Bearer " ++ federationSecret})
603622 case result of
604623 Ok _ -> do
605− atomic do
606− forwards <- *forwards
607− set *forwards = do
608− x <- forwards
609− where not (x.messageId == f.messageId && x.toServer == f.toServer)
610− yield x
624+ removeForward f.messageId f.toServer
611625 yield {}
612626 Err _ -> yield {}
613627 retryForwardsForServer rest (remaining - 1)
@@ -736,7 +750,7 @@ serve = \req -> case req of
736750 globalCapped
737751 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]
738752 yield pr
739− fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)
753+ forkBroadcastOffline t pruneResult
740754 yield (respond {challenge: challenge})
741755
742756 VerifyAuth {pubkey, challenge, signature, respond} -> do
@@ -813,7 +827,7 @@ serve = \req -> case req of
813827 tAfterPoll <- now
814828 pruneResult <- atomic do
815829 maybePrune tAfterPoll
816− fork (broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal)
830+ forkBroadcastOffline tAfterPoll pruneResult
817831 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
818832
819833 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
@@ -828,9 +842,7 @@ serve = \req -> case req of
828842 messages <- *messages
829843 presence <- *presence
830844 let session = lookupSession sessions (extractToken authorization) t
831− let fromKey = case session of
832− Nothing {} -> ""
833− Just {value: s} -> s.pubkey
845+ let fromKey = sessionPubkey session
834846 let authorized = isJust session
835847 seqCounter <- *seqCounter
836848 let seqTs = nextSeqTs t seqCounter
@@ -865,24 +877,20 @@ serve = \req -> case req of
865877 else forwards
866878 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
867879 let finalStatus = if result.status == "federated" && (not federationEnabled || result.forwardsAtCapacity) then "queued" else result.status
880+ let actionable = not (result.status == "unauthorized") && not (result.status == "rejected")
868881 fork (do
869− if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "rejected") && federationEnabled
882+ if count result.remoteServers > 0 && actionable && federationEnabled
870883 then sendToEach result.remoteServers (\server -> do
871884 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
872885 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
873886 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
874887 authorization: "Bearer " ++ federationSecret})
875888 case fwdResult of
876− Ok _ -> atomic do
877− forwards <- *forwards
878− set *forwards = do
879− x <- forwards
880− where not (x.messageId == msgId && x.toServer == server)
881− yield x
889+ Ok _ -> removeForward msgId server
882890 Err _ -> yield {}
883891 )
884892 else yield {}
885− if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "rejected")
893+ if result.forwardsAtCapacity && count result.remoteServers > 0 && actionable
886894 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
887895 else yield {}
888896 broadcastOfflineGossip t result.peerServers result.expiredLocal)
@@ -992,7 +1000,7 @@ serve = \req -> case req of
9921000 then serverHostN
9931001 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
9941002 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
995− fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
1003+ forkBroadcastOffline t result
9961004 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
9971005
9981006 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -1142,7 +1150,7 @@ serve = \req -> case req of
11421150 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
11431151 else messages
11441152 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}
1145− fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
1153+ forkBroadcastOffline t result
11461154 yield (respond {ok: result.inserted || (authorized && validOrigin && result.alreadyExists)})
11471155
11481156 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
@@ -1166,7 +1174,7 @@ serve = \req -> case req of
11661174 then diff forwards forwardsToRemove
11671175 else forwards
11681176 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1169− fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)
1177+ forkBroadcastOffline t pruneResult
11701178 yield (respond {ok: pruneResult.validAck})
11711179
11721180 Prune {authorization, respond} -> do