@@ -440,7 +440,6 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati
440440-- Generic helpers
441441-- ============================================
442442
443−id = \x -> x
444443filter = \xs pred -> do x <- xs; where pred x; yield x
445444removeWhere = \xs pred -> filter xs (\x -> not (pred x))
446445pluck = \xs f -> do x <- xs; yield (f x)
@@ -533,19 +532,17 @@ isSendSuccess = \s -> case s of
533532 MsDuplicate {} -> True {}
534533 _ -> False {}
535534
536−-- Sequence an IO action over each unique item (dedup by getKey).
537−forEach = \items getKey action -> case headOf items of
535+-- Sequence an IO action over each item, in order.
536+forEach = \items action -> case headOf items of
538537 Nothing {} -> yield {}
539− Just {value: x} -> do
540− action x
541− forEach (filter items (\y -> getKey y != getKey x)) getKey action
538+ Just {value: x} -> do action x; forEach (drop 1 items) action
542539
543540-- Like forEach, but collects each action's result into a list.
544−mapEach = \items getKey action -> case headOf items of
541+mapEach = \items action -> case headOf items of
545542 Nothing {} -> yield []
546543 Just {value: x} -> do
547544 r <- action x
548− rest <- mapEach (filter items (\y -> getKey y != getKey x)) getKey action
545+ rest <- mapEach (drop 1 items) action
549546 yield (union [r] rest)
550547
551548-- Hex tokens: one random hex char per recursive call.
@@ -890,7 +887,7 @@ cancelOne = \mid to ownerPred t -> atomic do
890887 when owns (do _ <- applyCancel msgs matches to mid t; yield {})
891888 yield owns
892889
893−sendBatchedGossip = \peers events hops -> forEach peers id (\server -> do
890+sendBatchedGossip = \peers events hops -> forEach peers (\server -> do
894891 t <- now
895892 failures <- *forwardFailures
896893 when (shouldRetryServer server t failures) (do
@@ -917,8 +914,8 @@ retryPendingForwards = do
917914 fwds <- *forwards
918915 failures <- *forwardFailures
919916 blocked <- *blockedPeers
920− forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))) id (\server ->
921− forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> f.messageId) (\f -> do
917+ forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))) (\server ->
918+ forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> do
922919 t2 <- now
923920 auth <- generateFederationAuth ("forward:" ++ f.messageId) t2
924921 result <- fetch (federationUrl server) (ForwardMessage {
@@ -940,7 +937,7 @@ retryPendingForwards = do
940937 else x)))
941938
942939sendAcksAndClean = \pendingAcks ->
943− forEach (collectPeers pendingAcks (\a -> a.originServer)) id (\server -> do
940+ forEach (collectPeers pendingAcks (\a -> a.originServer)) (\server -> do
944941 let mids = mapWhere pendingAcks (\a -> a.originServer == server) (\a -> a.messageId)
945942 t <- now
946943 auth <- generateFederationAuth "ack" t
@@ -979,7 +976,6 @@ pruneAndRetry = \t -> do
979976 when (t - lastRefresh >= presenceRefreshInterval) (do
980977 atomic do set *lastPresenceRefresh = t
981978 refreshLocalPresence t))
982− yield {}
983979
984980refreshLocalPresence = \t -> do
985981 refreshState <- atomic do
@@ -1213,7 +1209,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg
12131209 yield {status: if any toPresence (\p -> isLocalServer p.server) then MsDelivered {} else if forwardsEnqueued then MsFederated {} else MsQueued {}, remoteServers: remoteServers}
12141210
12151211forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->
1216− forEach servers id (\server -> do
1212+ forEach servers (\server -> do
12171213 fwds <- *forwards
12181214 let same = \x -> x.messageId == effMsgId && x.toServer == server
12191215 case findFirst fwds same of
@@ -1329,7 +1325,7 @@ handleCancelMessage = \messageId to authorization clientIp respond -> do
13291325 yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))
13301326
13311327sendCancelToEach = \messageId to servers ->
1332− forEach servers id (\server -> do
1328+ forEach servers (\server -> do
13331329 t <- now
13341330 auth <- generateFederationAuth ("cancel:" ++ messageId) t
13351331 result <- fetch (federationUrl server) (FederationCancel {
@@ -1344,7 +1340,7 @@ handleCancelBatch = \cancels authorization clientIp respond -> do
13441340 let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}
13451341 gateClientWrite authorization clientIp t respond mkErr
13461342 {results: [], authorized: True {}, error: "rate_limited"} (\senderKey -> do
1347− results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1343+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> do
13481344 r <- cancelOne e.messageId e.to (\m -> m.originServer == "" && m.senderKey == senderKey) t
13491345 yield {messageId: e.messageId, cancelled: r})
13501346 yield (respond {results: results, authorized: True {}, error: ""}))
@@ -1353,7 +1349,7 @@ handleSendBatch = \messages authorization clientIp respond -> do
13531349 t <- now
13541350 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
13551351 gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
1356− results <- mapEach (takeRelation maxBatchSize messages) id (\m -> do
1352+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
13571353 effMsgId <- resolveMessageId m.messageId
13581354 let clampedTs = clampTimestamp m.timestamp t
13591355 r <- if t - m.timestamp > messageMaxAge
@@ -1385,7 +1381,7 @@ handleVerifyToken = \token scope clientIp respond -> do
13851381handleFederationPubkey = \respond -> do
13861382 yield (respond {pubkey: serverFederationPublicKeyHex, ok: serverFederationPublicKeyHex != ""})
13871383
1388−handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlationId authorization respond -> do
1384+handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorization respond -> do
13891385 t <- now
13901386 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (clamp 0 maxGossipHops hopsRemaining >= 1 && not (isLocalServer fromServer) && count events <= maxGossipEvents) (do
13911387 newlyOnline <- atomic do
@@ -1402,7 +1398,7 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
14021398 let offline = mapWhere validEvents isOffline (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})
14031399 set *presence = capList (mergePresence (filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))) online) maxPresenceEntries (\p -> p.expiresAt)
14041400 yield (mapWhere online (\e -> none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)) (\e -> {pubkey: e.pubkey, server: e.server}))
1405− fork (forEach newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
1401+ fork (forEach newlyOnline (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t))
14061402 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
14071403
14081404-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
@@ -1429,7 +1425,7 @@ storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt
14291425 set *pollWakeup = seqTs
14301426 yield ""
14311427
1432−handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
1428+handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp expiresAt priority authorization respond -> do
14331429 t <- now
14341430 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
14351431 if t - timestamp > messageMaxAge
@@ -1438,7 +1434,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
14381434 result <- atomic (storeFederated toKey encryptedBlob timestamp originServer messageId expiresAt priority (True {}) t)
14391435 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
14401436
1441−handleFederationAck = \messageIds originServer fromServer requestTimestamp correlationId authorization respond -> do
1437+handleFederationAck = \messageIds originServer fromServer requestTimestamp authorization respond -> do
14421438 t <- now
14431439 let ackRespond = \r -> respond {ok: r.ok, error: r.error, ackedCount: 0, results: [], retryAfterMs: r.retryAfterMs}
14441440 withFederationAuth authorization t ackRespond fromServer "ack" requestTimestamp (isLocalServer originServer && not (isLocalServer fromServer)) (do
@@ -1461,27 +1457,27 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre
14611457 yield {ackedCount: count removed, results: pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})}
14621458 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))
14631459
1464−handleFederationCancel = \messageId toKey fromServer requestTimestamp correlationId authorization respond -> do
1460+handleFederationCancel = \messageId toKey fromServer requestTimestamp authorization respond -> do
14651461 t <- now
14661462 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}
14671463 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (do
14681464 cancelled <- cancelOne messageId toKey (\m -> m.originServer == fromServer) t
14691465 yield (respond {ok: True {}, error: "", cancelled: cancelled, retryAfterMs: 0}))
14701466
1471−handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId authorization respond -> do
1467+handleFederationBatchCancel = \cancels fromServer requestTimestamp authorization respond -> do
14721468 t <- now
14731469 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
14741470 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do
1475− results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do
1471+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> do
14761472 cancelled <- cancelOne e.messageId e.toKey (\m -> m.originServer == fromServer) t
14771473 yield {messageId: e.messageId, cancelled: cancelled})
14781474 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14791475
1480−handleFederationBatchForward = \forwards fromServer requestTimestamp correlationId authorization respond -> do
1476+handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
14811477 t <- now
14821478 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}
14831479 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1484− results <- mapEach (takeRelation maxBatchSize forwards) (\e -> e.messageId) (\e -> do
1480+ results <- mapEach (takeRelation maxBatchSize forwards) (\e -> do
14851481 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
14861482 else do
14871483 result <- atomic (storeFederated e.toKey e.encryptedBlob e.timestamp e.originServer e.messageId e.expiresAt e.priority (False {}) t)
@@ -1489,7 +1485,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
14891485 yield {messageId: e.messageId, ok: ok})
14901486 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
14911487
1492−handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp correlationId authorization respond -> do
1488+handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp authorization respond -> do
14931489 t <- now
14941490 let receiptRespond = \r -> respond {ok: r.ok, error: r.error, receivedCount: 0, retryAfterMs: r.retryAfterMs}
14951491 withFederationAuth authorization t receiptRespond fromServer "delivery-receipt" requestTimestamp (not (isLocalServer fromServer) && count messageIds <= maxMessageIds) (do
@@ -1648,8 +1644,7 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization
16481644 let same = \s -> s.ip == ip
16491645 set *sessions = removeWhere ses same
16501646 yield (mapWhere ses same (\s -> s.pubkey))
1651− let uniqueKeys = unique keys
1652− yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count uniqueKeys}))
1647+ yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count (unique keys)}))
16531648
16541649handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization
16551650 (yield (respond {sessions: [], ok: False {}, total: 0}))
@@ -1816,13 +1811,13 @@ serve = \req -> case req of
18161811 CancelBatch {cancels, authorization, clientIp, respond} -> handleCancelBatch cancels authorization clientIp respond
18171812 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
18181813 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
1819− RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp correlationId authorization respond
1820− ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, correlationId, expiresAt, priority, protocolVersion, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond
1821− FederationAck {messageIds, originServer, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp correlationId authorization respond
1822− FederationCancel {messageId, toKey, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp correlationId authorization respond
1823− FederationBatchCancel {cancels, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp correlationId authorization respond
1824− FederationBatchForward {forwards, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp correlationId authorization respond
1825− FederationDeliveryReceipt {messageIds, toKeys, fromServer, requestTimestamp, correlationId, protocolVersion, authorization, respond} -> handleFederationDeliveryReceipt messageIds toKeys fromServer requestTimestamp correlationId authorization respond
1814+ RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
1815+ ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, requestTimestamp, expiresAt, priority, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp expiresAt priority authorization respond
1816+ FederationAck {messageIds, originServer, fromServer, requestTimestamp, authorization, respond} -> handleFederationAck messageIds originServer fromServer requestTimestamp authorization respond
1817+ FederationCancel {messageId, toKey, fromServer, requestTimestamp, authorization, respond} -> handleFederationCancel messageId toKey fromServer requestTimestamp authorization respond
1818+ FederationBatchCancel {cancels, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchCancel cancels fromServer requestTimestamp authorization respond
1819+ FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
1820+ FederationDeliveryReceipt {messageIds, toKeys, fromServer, requestTimestamp, authorization, respond} -> handleFederationDeliveryReceipt messageIds toKeys fromServer requestTimestamp authorization respond
18261821 VerifyToken {token, scope, clientIp, respond} -> handleVerifyToken token scope clientIp respond
18271822 FederationPubkey {respond} -> handleFederationPubkey respond
18281823 Prune {authorization, respond} -> handlePrune authorization respond