@@ -815,34 +815,35 @@ serve = \req -> case req of
815815 yield SendUnauthorized {pr: pr}
816816 Authorized {pubkey: fromKey} -> do
817817 messages <- *messages
818− presence <- *presence
819− seqCounter <- *seqCounter
820− let seqTs = nextSeqTs t seqCounter
821− let shouldInsert = countMessagesFor messages to t < maxMessagesPerRecipient
822− set *seqCounter = if shouldInsert then seqTs else seqCounter
823− set *messages = if shouldInsert
824− then union messages
825− [{messageId: msgId, toKey: to, fromKey: fromKey,
826− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
827− else messages
828− let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
829− let localOnline = any toPresence (\p -> p.server == serverHostN)
830− let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
831− forwards <- *forwards
832− let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
833− let status = if not shouldInsert then "rejected"
834− else if localOnline then "delivered"
835− else if federationEnabled && count forwardServers > 0 then "federated"
836− else "queued"
837− set *forwards = if shouldInsert && federationEnabled && count forwardServers > 0
838− then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
839− else forwards
840− yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
818+ if countMessagesFor messages to t >= maxMessagesPerRecipient
819+ then yield SendRejected {pr: pr}
820+ else do
821+ presence <- *presence
822+ seqCounter <- *seqCounter
823+ let seqTs = nextSeqTs t seqCounter
824+ set *seqCounter = seqTs
825+ set *messages = union messages
826+ [{messageId: msgId, toKey: to, fromKey: fromKey,
827+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
828+ let toPresence = do p <- presence; where p.pubkey == to; where p.expiresAt > t; yield p
829+ let localOnline = any toPresence (\p -> p.server == serverHostN)
830+ let remoteServers = do p <- toPresence; where not (p.server == serverHostN); yield p.server
831+ forwards <- *forwards
832+ let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
833+ let status = if localOnline then "delivered"
834+ else if federationEnabled && count forwardServers > 0 then "federated"
835+ else "queued"
836+ set *forwards = if federationEnabled && count forwardServers > 0
837+ then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
838+ else forwards
839+ yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}
841840 fork (case result of
842841 SendUnauthorized {pr} ->
843842 broadcastOfflineGossip t pr.peerServers pr.expiredLocal
843+ SendRejected {pr} ->
844+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal
844845 SendDone {status, remoteServers, fromKey, pr} -> do
845− when (count remoteServers > 0 && not (status == "rejected") && federationEnabled)
846+ when (count remoteServers > 0 && federationEnabled)
846847 (sendToEach remoteServers (\server -> do
847848 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
848849 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
@@ -856,7 +857,8 @@ serve = \req -> case req of
856857 broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
857858 yield (case result of
858859 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}
859− SendDone {status, remoteServers, fromKey, pr} -> respond {status: status, messageId: if status == "rejected" then "" else msgId})
860+ SendRejected _ -> respond {status: "rejected", messageId: ""}
861+ SendDone {status} -> respond {status: status, messageId: msgId})
860862
861863 AckMessages {messageIds, authorization, respond} -> do
862864 t <- now
@@ -956,82 +958,86 @@ serve = \req -> case req of
956958 let authorized = verifyFederationAuth authorization
957959 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining
958960 let fromServer = normalizeServerName fromServer
959− let validRequest = authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents
960− let allValidEntries = if validRequest then validGossipEvents events t else []
961− let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
962− let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
963− let filteredOnline = do
964− e <- onlineEntries
965− where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
966− yield e
967− let filteredOffline = do
968− e <- offlineEntries
969− where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
970− yield e
971− let trustedOffline = do
972− e <- offlineEntries
973− where e.server == fromServer
974− yield e
975− let deduped = mergeByPubkeyServer [] filteredOnline
976− gossipResult <- atomic do
977− pr <- maybePrune t
978− presence <- *presence
979− let trustedDeduped = do
980− e <- deduped
981− where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
982− yield e
983− let appliedOnline = do
984− e <- trustedDeduped
985− where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
986− yield {pubkey: e.pubkey, server: e.server}
987− let merged = mergeByPubkeyServer presence trustedDeduped
988− let offlineOnlyKeys = do
989− e <- allValidEntries
990− where none trustedDeduped (\d -> samePubkeyServer d e)
991− yield e
992− let appliedOffline = do
993− k <- offlineOnlyKeys
994− where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
995− yield {pubkey: k.pubkey, server: k.server}
996− let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
997− let withinCapacity = count afterOffline <= maxPresenceEntries
998− let capResult = if withinCapacity
999− then {finalPresence: afterOffline, appliedOnline: appliedOnline}
1000− else
1001− let updatesOnly = do e <- trustedDeduped; where any presence (\p -> samePubkeyServer p e); yield e in
1002− let updatesOnlyMerged = mergeByPubkeyServer presence updatesOnly in
1003− let updatesOnlyAfterOffline = removeOffline updatesOnlyMerged offlineOnlyKeys trustedOffline t in
1004− let reclaimSlots = maxPresenceEntries - count updatesOnlyAfterOffline in
1005− let newEntries = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e); where none updatesOnlyAfterOffline (\p -> samePubkeyServer p e); yield e in
1006− let reclaimedEntries = takeN newEntries reclaimSlots in
1007− {finalPresence: union updatesOnlyAfterOffline reclaimedEntries,
1008− appliedOnline: union
1009− (do e <- updatesOnly; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield {pubkey: e.pubkey, server: e.server})
1010− (do e <- reclaimedEntries; yield {pubkey: e.pubkey, server: e.server})}
1011− let finalPresence = if validRequest then capResult.finalPresence else presence
1012− set *presence = finalPresence
1013− let gossipPeerServers = collectPeerServers finalPresence (\p -> p.server)
1014− yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1015− let effectiveEvents = do
1016− e <- union filteredOnline filteredOffline
1017− where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1018− yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1019− when (not gossipResult.withinCapacity)
1020− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1021− fork (do
1022− when (validRequest && cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1023− let targets = do
1024− s <- gossipResult.gossipPeerServers
1025− where not (s == fromServer)
1026− yield s
1027− sendToEach targets (\server -> do
1028− _ <- fetch (federationScheme ++ server) (RecvGossip {
1029− authorization: "Bearer " ++ federationSecret,
1030− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1031− yield {}))
1032− let pr = gossipResult.pr
1033− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1034− yield (respond {ok: validRequest})
961+ if not (authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents)
962+ then do
963+ pruneResult <- atomic do maybePrune t
964+ forkBroadcastOffline t pruneResult
965+ yield (respond {ok: False {}})
966+ else do
967+ let allValidEntries = validGossipEvents events t
968+ let onlineEntries = do e <- allValidEntries; where e.eventType == "online"; yield e
969+ let offlineEntries = do e <- allValidEntries; where e.eventType == "offline"; yield e
970+ let filteredOnline = do
971+ e <- onlineEntries
972+ where none offlineEntries (\o -> o.server == fromServer && samePubkeyServer o e && o.timestamp > e.timestamp)
973+ yield e
974+ let filteredOffline = do
975+ e <- offlineEntries
976+ where none onlineEntries (\o -> samePubkeyServer o e && o.timestamp > e.timestamp)
977+ yield e
978+ let trustedOffline = do
979+ e <- offlineEntries
980+ where e.server == fromServer
981+ yield e
982+ let deduped = mergeByPubkeyServer [] filteredOnline
983+ gossipResult <- atomic do
984+ pr <- maybePrune t
985+ presence <- *presence
986+ let trustedDeduped = do
987+ e <- deduped
988+ where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
989+ yield e
990+ let appliedOnline = do
991+ e <- trustedDeduped
992+ where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
993+ yield {pubkey: e.pubkey, server: e.server}
994+ let merged = mergeByPubkeyServer presence trustedDeduped
995+ let offlineOnlyKeys = do
996+ e <- allValidEntries
997+ where none trustedDeduped (\d -> samePubkeyServer d e)
998+ yield e
999+ let appliedOffline = do
1000+ k <- offlineOnlyKeys
1001+ where any presence (\p -> samePubkeyServer p k && any trustedOffline (\o -> samePubkeyServer o k && (if o.timestamp > t then t else o.timestamp) > p.timestamp))
1002+ yield {pubkey: k.pubkey, server: k.server}
1003+ let afterOffline = removeOffline merged offlineOnlyKeys trustedOffline t
1004+ let withinCapacity = count afterOffline <= maxPresenceEntries
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+ set *presence = capResult.finalPresence
1019+ let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
1020+ yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: withinCapacity}
1021+ let effectiveEvents = do
1022+ e <- union filteredOnline filteredOffline
1023+ where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
1024+ yield {eventType: e.eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
1025+ when (not gossipResult.withinCapacity)
1026+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
1027+ fork (do
1028+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (do
1029+ let targets = do
1030+ s <- gossipResult.gossipPeerServers
1031+ where not (s == fromServer)
1032+ yield s
1033+ sendToEach targets (\server -> do
1034+ _ <- fetch (federationScheme ++ server) (RecvGossip {
1035+ authorization: "Bearer " ++ federationSecret,
1036+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})
1037+ yield {}))
1038+ let pr = gossipResult.pr
1039+ broadcastOfflineGossip t pr.peerServers pr.expiredLocal)
1040+ yield (respond {ok: True {}})
10351041
10361042 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
10371043 let toKey = normalizeHex toKey
@@ -1045,23 +1051,24 @@ serve = \req -> case req of
10451051 let clampedTs = clampTimestamp timestamp t
10461052 result <- atomic do
10471053 pr <- maybePrune t
1048− messages <- *messages
1049− seqCounter <- *seqCounter
1050− let seqTs = nextSeqTs t seqCounter
1051− ackedIds <- *ackedIds
1052− let valid = authorized && validOrigin
1053− let alreadyExists = valid
1054− && (any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1055− || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey))
1056− let shouldInsert = valid && not alreadyExists
1057− && countMessagesFor messages toKey t < maxMessagesPerRecipient
1058− set *seqCounter = if shouldInsert then seqTs else seqCounter
1059− set *messages = if shouldInsert
1060− then union messages
1061− [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1062− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1063− else messages
1064− yield {pr: pr, ok: shouldInsert || alreadyExists}
1054+ if not (authorized && validOrigin)
1055+ then yield {pr: pr, ok: False {}}
1056+ else do
1057+ messages <- *messages
1058+ ackedIds <- *ackedIds
1059+ if any messages (\m -> m.messageId == messageId && m.toKey == toKey)
1060+ || any ackedIds (\a -> a.messageId == messageId && a.toKey == toKey)
1061+ then yield {pr: pr, ok: True {}}
1062+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1063+ then yield {pr: pr, ok: False {}}
1064+ else do
1065+ seqCounter <- *seqCounter
1066+ let seqTs = nextSeqTs t seqCounter
1067+ set *seqCounter = seqTs
1068+ set *messages = union messages
1069+ [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1070+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
1071+ yield {pr: pr, ok: True {}}
10651072 forkBroadcastOffline t result.pr
10661073 yield (respond {ok: result.ok})
10671074