@@ -68,7 +68,7 @@ type Message = {messageId: MessageIdHex, toKey: PubkeyHex, fromKey: PubkeyHex, e
6868type PresenceEntry = {pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp}
6969type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
7070type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
71−type AuthResult = {authorized: Bool, pubkey: Text}
71+type AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
7272type PruneResult = {expiredLocal: [PresenceEntry], peerServers: [Text]}
7373
7474
@@ -149,9 +149,6 @@ headOf = \items -> fold (\acc x -> case acc of
149149 Nothing {} -> Just {value: x}
150150 _ -> acc) Nothing {} items
151151
152−msToS : Int<Ms> -> Int<S>
153−msToS = \x -> x / 1000
154−
155152hexDigit = \n ->
156153 let n = if n < 0 then 0 else if n > 15 then 15 else n in
157154 if n == 0 then "0" else if n == 1 then "1" else if n == 2 then "2" else if n == 3 then "3"
@@ -366,12 +363,12 @@ isBadServerName = \s ->
366363authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
367364authenticateSession = \sessions token t ->
368365 fold (\acc s ->
369− if acc.authorized then acc
370− else
371− let matches = safeCompare s.token token in
372− if matches && s.expiresAt > t
373− then {authorized: True {}, pubkey: s.pubkey}
374− else acc) {authorized: False {}, pubkey: ""} sessions
366+ case acc of
367+ Authorized _ -> acc
368+ Unauthorized {} ->
369+ if safeCompare s.token token && s.expiresAt > t
370+ then Authorized {pubkey: s.pubkey}
371+ else acc) Unauthorized {} sessions
375372
376373extractToken = \authHeader ->
377374 if toLower (take 7 authHeader) == "bearer "
@@ -495,10 +492,9 @@ countMessagesFor = \messages toKey t ->
495492 yield m)
496493
497494
498−validGossipEvents : [GossipEvent] -> Timestamp -> GossipEventType -> [ValidGossipEntry]
499−validGossipEvents = \events t eventType -> do
495+validGossipEvents : [GossipEvent] -> Timestamp -> [ValidGossipEntry]
496+validGossipEvents = \events t -> do
500497 e <- events
501− where e.eventType == eventType
502498 where not (tooLong e.pubkey maxPubkeyLen)
503499 where not (tooLong e.server maxServerNameLen)
504500 where not (isBadServerName e.server)
@@ -507,8 +503,9 @@ validGossipEvents = \events t eventType -> do
507503 where not (server == serverHostN)
508504 where e.timestamp <= t + clockSkewTolerance
509505 let ttl = clampTtl e.ttl
510− where e.timestamp + ttl * 1000 > t
511− yield {eventType: eventType, pubkey: normalizeHex e.pubkey, server: server, expiresAt: e.timestamp + ttl * 1000, timestamp: e.timestamp, ttl: ttl}
506+ let expiresAt = e.timestamp + ttl * 1000
507+ where expiresAt > t
508+ yield {eventType: e.eventType, pubkey: normalizeHex e.pubkey, server: server, expiresAt: expiresAt, timestamp: e.timestamp, ttl: ttl}
512509
513510sendToEach : [Text] -> (Text -> (a {})) -> a {}
514511sendToEach = \servers action ->
@@ -562,17 +559,21 @@ maybePrune = \t -> do
562559 set *lastPrunedAt = t
563560 pruneExpired t
564561
562+forEachBatch = \items batchSize process ->
563+ if count items == 0 then yield {}
564+ else do
565+ let batch = takeN items batchSize
566+ let rest = diff items batch
567+ process batch
568+ forEachBatch rest batchSize process
569+
565570sendBatchedGossipToServer : Text -> [GossipEvent] -> IO {network} {}
566571sendBatchedGossipToServer = \server allEvents ->
567− if count allEvents == 0
568− then yield {}
569− else do
570− let batch = takeN allEvents maxGossipEvents
571− let rest = diff allEvents batch
572− _ <- fetch (federationScheme ++ server) (RecvGossip {
573− authorization: "Bearer " ++ federationSecret,
574− events: batch, hopsRemaining: 2, fromServer: serverHostN})
575− sendBatchedGossipToServer server rest
572+ forEachBatch allEvents maxGossipEvents (\batch -> do
573+ _ <- fetch (federationScheme ++ server) (RecvGossip {
574+ authorization: "Bearer " ++ federationSecret,
575+ events: batch, hopsRemaining: 2, fromServer: serverHostN})
576+ yield {})
576577
577578sendBatchedGossip : [Text] -> [GossipEvent] -> IO {network} {}
578579sendBatchedGossip = \peerServers allEvents ->
@@ -628,25 +629,20 @@ retryPendingForwards = \pending -> do
628629
629630sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> Text -> IO {network} {}
630631sendBatchedAcksToServer = \serverAcks server ->
631− if count serverAcks == 0
632− then yield {}
633− else do
634− let batch = takeN serverAcks maxMessageIds
635− let rest = diff serverAcks batch
636− result <- fetch (federationScheme ++ server) (FederationAck {
637− authorization: "Bearer " ++ federationSecret,
638− messageIds: (do a <- batch; yield a.messageId),
639− originServer: server, fromServer: serverHostN})
640− case result of
641− Ok _ ->
642− atomic do
643− ackedIds <- *ackedIds
644− set *ackedIds = do
645− a <- ackedIds
646− where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
647− yield a
648− Err _ -> yield {}
649− sendBatchedAcksToServer rest server
632+ forEachBatch serverAcks maxMessageIds (\batch -> do
633+ result <- fetch (federationScheme ++ server) (FederationAck {
634+ authorization: "Bearer " ++ federationSecret,
635+ messageIds: (do a <- batch; yield a.messageId),
636+ originServer: server, fromServer: serverHostN})
637+ case result of
638+ Ok _ ->
639+ atomic do
640+ ackedIds <- *ackedIds
641+ set *ackedIds = do
642+ a <- ackedIds
643+ where not (a.originServer == server && any batch (\sa -> sa.messageId == a.messageId && sa.toKey == a.toKey))
644+ yield a
645+ Err _ -> yield {})
650646
651647sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: Text, | a}] -> [Text] -> IO {network} {}
652648sendAcksAndClean = \pendingAcks servers ->
@@ -676,28 +672,30 @@ pollOnce = \token cursor -> do
676672 t <- now
677673 atomic do
678674 sessions <- *sessions
679− messages <- *messages
680675 seqCounter <- *seqCounter
681676 let auth = authenticateSession sessions token t
682677 let cursorCeil = if seqCounter > t then seqCounter else t
683678 let clampedCursor = if cursor < 0 then 0 else if cursor > cursorCeil then cursorCeil else cursor
684− let matchingMsgs = if not auth.authorized then []
685− else do
686− m <- messages
687− where m.toKey == auth.pubkey
688− where m.receivedAt >= t - messageMaxAge
689− where m.receivedAt > clampedCursor
690− yield m
691− let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
692− let evts = do
693− m <- limitedMsgs
694− yield {id: m.messageId, eventType: "message",
695− payload: {toKey: m.toKey, fromKey: m.fromKey,
696− encryptedBlob: m.encryptedBlob,
697− timestamp: m.timestamp}}
698− let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
699− let newCursor = if not auth.authorized then -1 else lastReceivedAt
700− yield {events: evts, cursor: newCursor, authorized: auth.authorized}
679+ case auth of
680+ Unauthorized {} ->
681+ yield {events: [], cursor: -1, authorized: False {}}
682+ Authorized {pubkey} -> do
683+ messages <- *messages
684+ let matchingMsgs = do
685+ m <- messages
686+ where m.toKey == pubkey
687+ where m.receivedAt >= t - messageMaxAge
688+ where m.receivedAt > clampedCursor
689+ yield m
690+ let limitedMsgs = boundedSmallest matchingMsgs maxPollEvents (\m -> m.receivedAt)
691+ let evts = do
692+ m <- limitedMsgs
693+ yield {id: m.messageId, eventType: "message",
694+ payload: {toKey: m.toKey, fromKey: m.fromKey,
695+ encryptedBlob: m.encryptedBlob,
696+ timestamp: m.timestamp}}
697+ let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs
698+ yield {events: evts, cursor: lastReceivedAt, authorized: True {}}
701699
702700pollLoop : Text -> Int -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}
703701pollLoop = \token cursor deadline -> do
@@ -816,38 +814,37 @@ serve = \req -> case req of
816814 result <- atomic do
817815 pr <- maybePrune t
818816 sessions <- *sessions
819− messages <- *messages
820− presence <- *presence
821817 let auth = authenticateSession sessions (extractToken authorization) t
822− let fromKey = auth.pubkey
823− let authorized = auth.authorized
824− seqCounter <- *seqCounter
825− let seqTs = nextSeqTs t seqCounter
826− let shouldInsert = authorized && countMessagesFor messages to t < maxMessagesPerRecipient
827− set *seqCounter = if shouldInsert then seqTs else seqCounter
828− set *messages = if shouldInsert
829− then union messages
830− [{messageId: msgId, toKey: to, fromKey: fromKey,
831− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
832− else messages
833− let toPresence = if authorized
834− then do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
835− else []
836− let localOnline = any toPresence (\p -> p.server == serverHostN)
837− let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
838− forwards <- *forwards
839− let forwardsAtCapacity = count forwards >= maxForwardsEntries
840− let status = if not authorized then "unauthorized"
841− else if not shouldInsert then "rejected"
842− else if localOnline then "delivered"
843− else if count remoteServers > 0 && federationEnabled && not forwardsAtCapacity then "federated"
844− else "queued"
845− let forwardSlots = maxForwardsEntries - count forwards
846− let forwardServers = takeN remoteServers forwardSlots
847− set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled
848− then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
849− else forwards
850− yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
818+ case auth of
819+ Unauthorized {} ->
820+ yield {status: "unauthorized", shouldInsert: False {}, remoteServers: [], fromKey: "", expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: False {}}
821+ Authorized {pubkey: fromKey} -> do
822+ messages <- *messages
823+ presence <- *presence
824+ seqCounter <- *seqCounter
825+ let seqTs = nextSeqTs t seqCounter
826+ let shouldInsert = countMessagesFor messages to t < maxMessagesPerRecipient
827+ set *seqCounter = if shouldInsert then seqTs else seqCounter
828+ set *messages = if shouldInsert
829+ then union messages
830+ [{messageId: msgId, toKey: to, fromKey: fromKey,
831+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
832+ else messages
833+ let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
834+ let localOnline = any toPresence (\p -> p.server == serverHostN)
835+ let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
836+ forwards <- *forwards
837+ let forwardsAtCapacity = count forwards >= maxForwardsEntries
838+ let status = if not shouldInsert then "rejected"
839+ else if localOnline then "delivered"
840+ else if count remoteServers > 0 && federationEnabled && not forwardsAtCapacity then "federated"
841+ else "queued"
842+ let forwardSlots = maxForwardsEntries - count forwards
843+ let forwardServers = takeN remoteServers forwardSlots
844+ set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled
845+ then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
846+ else forwards
847+ yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
851848 fork (do
852849 if count result.remoteServers > 0 && result.shouldInsert && federationEnabled
853850 then sendToEach result.remoteServers (\server -> do
@@ -872,29 +869,32 @@ serve = \req -> case req of
872869 result <- atomic do
873870 pr <- maybePrune t
874871 sessions <- *sessions
875− messages <- *messages
876872 let auth = authenticateSession sessions (extractToken authorization) t
877− let ackedMessages = if not auth.authorized then []
878− else do
879− m <- messages
880− mid <- dedupedIds
881− where m.messageId == mid
882− where m.toKey == auth.pubkey
883− yield m
884− set *messages = diff messages ackedMessages
885− let ackedFederated = do
886− m <- ackedMessages
887− where isFederatedOrigin m.originServer
888− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
889− ackedIds <- *ackedIds
890− let combinedAckedIds = union ackedIds ackedFederated
891− let excessCount = count combinedAckedIds - maxAckedIds
892− let afterEviction = if excessCount <= 0 then combinedAckedIds
893− else evictOldestAckedIds combinedAckedIds excessCount
894− let evictedEntries = if excessCount <= 0 then []
895− else diff combinedAckedIds afterEviction
896− set *ackedIds = afterEviction
897− yield {ok: auth.authorized, pendingAcks: ackedFederated, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
873+ case auth of
874+ Unauthorized {} ->
875+ yield {ok: False {}, pendingAcks: [], expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: []}
876+ Authorized {pubkey} -> do
877+ messages <- *messages
878+ let ackedMessages = do
879+ m <- messages
880+ mid <- dedupedIds
881+ where m.messageId == mid
882+ where m.toKey == pubkey
883+ yield m
884+ set *messages = diff messages ackedMessages
885+ let ackedFederated = do
886+ m <- ackedMessages
887+ where isFederatedOrigin m.originServer
888+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
889+ ackedIds <- *ackedIds
890+ let combinedAckedIds = union ackedIds ackedFederated
891+ let excessCount = count combinedAckedIds - maxAckedIds
892+ let afterEviction = if excessCount <= 0 then combinedAckedIds
893+ else evictOldestAckedIds combinedAckedIds excessCount
894+ let evictedEntries = if excessCount <= 0 then []
895+ else diff combinedAckedIds afterEviction
896+ set *ackedIds = afterEviction
897+ yield {ok: True {}, pendingAcks: ackedFederated, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, evictedEntries: evictedEntries}
898898 let originServers = collectPeerServers result.pendingAcks (\a -> a.originServer)
899899 fork (do
900900 if result.ok
@@ -941,20 +941,21 @@ serve = \req -> case req of
941941 result <- atomic do
942942 pr <- maybePrune t
943943 sessions <- *sessions
944− presence <- *presence
945944 let auth = authenticateSession sessions (extractToken authorization) t
946− let authorized = auth.authorized
947− let valid = do
948− p <- presence
949− where p.pubkey == pubkey
950− where p.expiresAt > t
951− yield p
952− let foundServer = if not authorized
953− then ""
954− else if any valid (\p -> p.server == serverHostN)
955− then serverHostN
956− else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
957− yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
945+ case auth of
946+ Unauthorized {} ->
947+ yield {foundServer: "", authorized: False {}, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
948+ Authorized _ -> do
949+ presence <- *presence
950+ let valid = do
951+ p <- presence
952+ where p.pubkey == pubkey
953+ where p.expiresAt > t
954+ yield p
955+ let foundServer = if any valid (\p -> p.server == serverHostN)
956+ then serverHostN
957+ else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server
958+ yield {foundServer: foundServer, authorized: True {}, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
958959 forkBroadcastOffline t result
959960 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})
960961
@@ -964,8 +965,9 @@ serve = \req -> case req of
964965 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
965966 let fromServer = normalizeServerName fromServer
966967 let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents
967− let onlineEntries = if validRequest then validGossipEvents events t "online" else []
968− let offlineEntries = if validRequest then validGossipEvents events t "offline" else []
968+ let allValidEntries = if validRequest then validGossipEvents events t else []
969+ let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
970+ let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
969971 let filteredOnline = do
970972 e <- onlineEntries
971973 where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
@@ -1000,37 +1002,23 @@ serve = \req -> case req of
10001002 yield {pubkey: k.pubkey, server: k.server}
10011003 let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
10021004 let withinCapacity = count afterOffline <= maxPresenceEntries
1003− let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e
1004− let updatesOnlyMerged = if withinCapacity then afterOffline
1005− else mergeByPubkeyServer presence updatesOnly
1006− let updatesOnlyAfterOffline = if withinCapacity then afterOffline
1007− else removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t
1008− let reclaimCapacity = if withinCapacity then 0 else maxPresenceEntries - count updatesOnlyAfterOffline
1009− let newEntries = if withinCapacity then []
1010− else do
1011− e <- trustedDeduped
1012− where none presence (\p -> samePubkeyServer p e)
1013− where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e)
1014− yield e
1015− let reclaimedEntries = if withinCapacity then []
1016− else takeN newEntries reclaimCapacity
1017− let afterReclaim = if withinCapacity then afterOffline
1018− else union updatesOnlyAfterOffline reclaimedEntries
1019− let appliedOnlineUpdates = if withinCapacity then appliedOnline
1020− else union
1021− (do
1022− e <- updatesOnly
1023− where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
1024− yield {pubkey: e.pubkey, server: e.server})
1025− (do
1026− e <- reclaimedEntries
1027− yield {pubkey: e.pubkey, server: e.server})
1028− let finalPresence = if validRequest
1029− then if withinCapacity then afterOffline else afterReclaim
1030− else presence
1005+ let capResult = if withinCapacity
1006+ then {finalPresence: afterOffline, appliedOnline: appliedOnline}
1007+ else
1008+ let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e in
1009+ let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
1010+ let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
1011+ let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
1012+ let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e); where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in
1013+ let reclaimedEntries = takeN newEntries reclaimSlots in
1014+ {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
1015+ appliedOnline: union
1016+ (do e <- updatesOnly; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield {pubkey: e.pubkey, server: e.server})
1017+ (do e <- reclaimedEntries; yield {pubkey: e.pubkey, server: e.server})}
1018+ let finalPresence = if validRequest then capResult.finalPresence else presence
10311019 set *presence = finalPresence
10321020 let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
1033− yield {appliedKeys: (if withinCapacity then union appliedOnline appliedOffline else union appliedOnlineUpdates appliedOffline), expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1021+ yield {appliedKeys: union capResult.appliedOnline appliedOffline, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
10341022 let filteredOffline = do
10351023 e <- offlineEntries
10361024 where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)