@@ -130,7 +130,7 @@ route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi
130130-- ============================================
131131
132132any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0
133−none = \xs pred -> count (do x <- xs; where pred x; yield x) == 0
133+none = \xs pred -> not (any xs pred)
134134isJust = \m -> case m of
135135 Nothing {} -> False {}
136136 _ -> True {}
@@ -463,14 +463,16 @@ collectPeerServers = \items getServer ->
463463
464464samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
465465
466−mergeByPubkeyServer = \initial items ->
466+mergeByKey = \sameKey initial items ->
467467 fold (\acc e ->
468− if any acc (\a -> samePubkeyServer a e && a.expiresAt >= e.expiresAt)
468+ if any acc (\a -> sameKey a e && a.expiresAt >= e.expiresAt)
469469 then acc
470470 else union
471− (do a <- acc; where not (samePubkeyServer a e); yield a)
471+ (do a <- acc; where not (sameKey a e); yield a)
472472 [e]) initial items
473473
474+mergeByPubkeyServer = mergeByKey samePubkeyServer
475+
474476removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->
475477 diff presenceSet (do
476478 p <- presenceSet
@@ -582,11 +584,9 @@ sendBatchedGossip = \peerServers allEvents ->
582584 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)
583585
584586broadcastOfflineGossip = \t peerServers expiredLocal -> do
585− let expiredEntries = fold (\acc p ->
586− if any acc (\a -> a.pubkey == p.pubkey && a.expiresAt >= p.expiresAt) then acc
587− else union
588− (do a <- acc; where not (a.pubkey == p.pubkey); yield a)
589− [{pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in clampTtl (if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else rawTtl)), expiresAt: p.expiresAt}]) [] expiredLocal
587+ let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do
588+ p <- expiredLocal
589+ yield {pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in clampTtl (if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else rawTtl)), expiresAt: p.expiresAt})
590590 if federationEnabled && count expiredEntries > 0 && count peerServers > 0
591591 then do
592592 let gossipEvents = do
@@ -863,23 +863,21 @@ serve = \req -> case req of
863863 where p.expiresAt > t
864864 yield p.server
865865 else []
866+ forwards <- *forwards
867+ let forwardsAtCapacity = count forwards >= maxForwardsEntries
866868 let status = if not authorized then "unauthorized"
867869 else if not shouldInsert then "rejected"
868870 else if localOnline then "delivered"
869− else if count remoteServers > 0 then "federated"
871+ else if count remoteServers > 0 && federationEnabled && not forwardsAtCapacity then "federated"
870872 else "queued"
871− forwards <- *forwards
872− let forwardsAtCapacity = count forwards >= maxForwardsEntries
873873 let forwardSlots = maxForwardsEntries - count forwards
874874 let forwardServers = takeN remoteServers forwardSlots
875875 set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled
876876 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
877877 else forwards
878− yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
879− 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")
878+ yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}
881879 fork (do
882− if count result.remoteServers > 0 && actionable && federationEnabled
880+ if count result.remoteServers > 0 && result.shouldInsert && federationEnabled
883881 then sendToEach result.remoteServers (\server -> do
884882 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {
885883 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
@@ -890,11 +888,11 @@ serve = \req -> case req of
890888 Err _ -> yield {}
891889 )
892890 else yield {}
893− if result.forwardsAtCapacity && count result.remoteServers > 0 && actionable
891+ if result.forwardsAtCapacity && count result.remoteServers > 0 && result.shouldInsert
894892 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
895893 else yield {}
896894 broadcastOfflineGossip t result.peerServers result.expiredLocal)
897− yield (respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "rejected" then "" else msgId)})
895+ yield (respond {status: result.status, messageId: if result.shouldInsert then msgId else ""})
898896
899897 AckMessages {messageIds, authorization, respond} -> do
900898 t <- now
@@ -1159,11 +1157,10 @@ serve = \req -> case req of
11591157 t <- now
11601158 let authorized = verifyFederationAuth authorization
11611159 let dedupedIds = dedupMessageIds messageIds
1162− let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)
1160+ let validAck = authorized && originServer == serverHostN && not (fromServer == serverHostN)
11631161 pruneResult <- atomic do
11641162 pr <- maybePrune t
11651163 forwards <- *forwards
1166− let validAck = validAckFields
11671164 let forwardsToRemove = do
11681165 f <- forwards
11691166 mid <- dedupedIds
@@ -1173,9 +1170,9 @@ serve = \req -> case req of
11731170 set *forwards = if validAck
11741171 then diff forwards forwardsToRemove
11751172 else forwards
1176− yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}
1173+ yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal}
11771174 forkBroadcastOffline t pruneResult
1178− yield (respond {ok: pruneResult.validAck})
1175+ yield (respond {ok: validAck})
11791176
11801177 Prune {authorization, respond} -> do
11811178 t <- now