83d4d5b · skrepka · git.iko.soy

git.iko.soy/skrepka

on master help sign in

Simplify server: add filter/pluck/countWhere/elem helpers, extract gateAuth for handlers

83d4d5biko6 months ago

server.knot+301−366

compare with master

unifiedsplit

@@ -439,19 +439,23 @@ route Api = AuthApi | ClientApi | FederationApi | FederationVerifyApi | Federati

439-- Generic helpers

440-- ============================================

441

442+filter = \xs pred -> do x <- xs; where pred x; yield x

443+pluck = \xs f -> do x <- xs; yield (f x)

444+countWhere = \xs pred -> count (filter xs pred)

442any : [a] -> (a -> Bool) -> Bool

445any : [a] -> (a -> Bool) -> Bool

443−any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0

446+any = \xs pred -> countWhere xs pred > 0

444none : [a] -> (a -> Bool) -> Bool

447none : [a] -> (a -> Bool) -> Bool

445none = \xs pred -> not (any xs pred)

448none = \xs pred -> not (any xs pred)

446all : [a] -> (a -> Bool) -> Bool

449all : [a] -> (a -> Bool) -> Bool

447all = \xs pred -> none xs (\x -> not (pred x))

450all = \xs pred -> none xs (\x -> not (pred x))

451+elem = \xs x -> any xs (\y -> y == x)

448when = \cond action -> if cond then action else yield {}

452when = \cond action -> if cond then action else yield {}

449clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x

453clamp = \lo hi x -> if x < lo then lo else if x > hi then hi else x

450min = \a b -> if a < b then a else b

454min = \a b -> if a < b then a else b

451max = \a b -> if a > b then a else b

455max = \a b -> if a > b then a else b

452

456

453-- Dedup: keeps first occurrence (O(n^2)).

457-- Dedup: keeps first occurrence (O(n^2)).

454−unique = \items -> fold (\acc x -> if any acc (\a -> a == x) then acc else union acc [x]) [] items

458+unique = \items -> fold (\acc x -> if elem acc x then acc else union acc [x]) [] items

455

459

456-- Replace a ref's contents with the entries that satisfy `keep`. Caller controls atomicity.

460-- Replace a ref's contents with the entries that satisfy `keep`. Caller controls atomicity.

457pruneRef = \ref keep -> do

461pruneRef = \ref keep -> do

@@ -596,9 +600,25 @@ withMaintAuth = \authorization onUnauth onAuth -> do

596 authorized <- verifyMaintenanceAuth authorization

600 authorized <- verifyMaintenanceAuth authorization

597 if authorized then onAuth else onUnauth

601 if authorized then onAuth else onUnauth

598

602

603+-- Common gate for client handlers: drain check, global rate, session auth.

604+-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".

605+gateAuth = \auth ip t respond mkErr onAuth -> do

606+ dp <- *drainPhase

607+ if dp >= 2 then yield (respond (mkErr "draining"))

608+ else do

609+ g <- checkGlobalRate t

610+ if g then yield (respond (mkErr "rate_limited"))

611+ else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth

612+

613+-- Variant without the drain check (used by lookup/status handlers).

614+gateAuthNoDrain = \auth ip t respond mkErr onAuth -> do

615+ g <- checkGlobalRate t

616+ if g then yield (respond (mkErr "rate_limited"))

617+ else withSessionAuth auth t ip (yield (respond (mkErr "unauthorized"))) onAuth

618+

599authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

619authenticateSession : [Session] -> Text -> Timestamp -> AuthResult

600authenticateSession = \sessions token t ->

620authenticateSession = \sessions token t ->

601− case headOf (do s <- sessions; where s.expiresAt > t && s.token == token; yield s) of

621+ case headOf (filter sessions (\s -> s.expiresAt > t && s.token == token)) of

602 Nothing {} -> Unauthorized {}

622 Nothing {} -> Unauthorized {}

603 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}

623 Just {value: s} -> Authorized {pubkey: s.pubkey, ip: s.ip}

604

624

@@ -612,12 +632,6 @@ withSessionAuth = \authorization t clientIp onUnauth onAuth -> do

612 if matchIp clientIp ip then onAuth pubkey else onUnauth

632 if matchIp clientIp ip then onAuth pubkey else onUnauth

613 _ -> onUnauth

633 _ -> onUnauth

614

634

615−isBlockedPeer : Text -> [Text] -> Bool

616−isBlockedPeer = \server blocked -> any blocked (\b -> b == server)

617−

618−isAllowedByAllowlist : Text -> [Text] -> Bool

619−isAllowedByAllowlist = \server allowed -> count allowed == 0 || any allowed (\a -> a == server)

620−

621splitOnColon : Text -> {before: Text, after: Text}

635splitOnColon : Text -> {before: Text, after: Text}

622splitOnColon = \s ->

636splitOnColon = \s ->

623 if s == "" then {before: "", after: ""}

637 if s == "" then {before: "", after: ""}

@@ -668,8 +682,8 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp

668 blocked <- *blockedPeers

682 blocked <- *blockedPeers

669 allowed <- *allowedPeers

683 allowed <- *allowedPeers

670 let preErr =

684 let preErr =

671− if isBlockedPeer fromServer blocked then "blocked"

685+ if elem blocked fromServer then "blocked"

672− else if not (isAllowedByAllowlist fromServer allowed) then "not_allowed"

686+ else if count allowed > 0 && not (elem allowed fromServer) then "not_allowed"

673 else if not precondition then "invalid_request"

687 else if not precondition then "invalid_request"

674 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance) then "clock_skew"

688 else if requestTimestamp > 0 && (requestTimestamp > t + clockSkewTolerance || requestTimestamp < t - clockSkewTolerance) then "clock_skew"

675 else ""

689 else ""

@@ -689,14 +703,13 @@ withFederationAuth = \authorization t respond fromServer scope requestTimestamp

689-- Returns {limited, entries}. Increments count when the key fits within the window.

703-- Returns {limited, entries}. Increments count when the key fits within the window.

690-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.

704-- Simplified: drop any stale/matching entry, then either reject or insert a fresh entry.

691checkRate = \entries key maxRate windowMs maxEntries t ->

705checkRate = \entries key maxRate windowMs maxEntries t ->

692− let cur = headOf (do r <- entries; where r.key == key && t - r.windowStart < windowMs; yield r) in

706+ let cur = headOf (filter entries (\r -> r.key == key && t - r.windowStart < windowMs)) in

693 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in

707 let nextCount = case cur of Just {value: r} -> r.count + 1; _ -> 1 in

694 if nextCount > maxRate

708 if nextCount > maxRate

695 then {limited: True {}, entries: entries}

709 then {limited: True {}, entries: entries}

696 else

710 else

697− let others = do x <- entries; where x.key != key; yield x in

698 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in

711 let windowStart = case cur of Just {value: r} -> r.windowStart; _ -> t in

699− {limited: False {}, entries: capList (union others [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}

712+ {limited: False {}, entries: capList (union (filter entries (\x -> x.key != key)) [{key: key, windowStart: windowStart, count: nextCount}]) maxEntries (\r -> r.windowStart)}

700

713

701checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do

714checkWindowRate = \windowRef countRef windowMs maxRate t -> atomic do

702 ws <- *windowRef

715 ws <- *windowRef

@@ -744,7 +757,7 @@ retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failur

744

757

745shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool

758shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool

746shouldRetryServer = \server t failures ->

759shouldRetryServer = \server t failures ->

747− case headOf (do f <- failures; where f.server == server; yield f) of

760+ case headOf (filter failures (\f -> f.server == server)) of

748 Nothing {} -> True {}

761 Nothing {} -> True {}

749 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures

762 Just {value: f} -> t - f.failedAt >= retryBackoff f.failures

750

763

@@ -752,11 +765,10 @@ recordFailure = \server -> do

752 t <- now

765 t <- now

753 atomic do

766 atomic do

754 failures <- *forwardFailures

767 failures <- *forwardFailures

755− let others = do f <- failures; where f.server != server; yield f

756− let new = case headOf (do f <- failures; where f.server == server; yield f) of

768+ let new = case headOf (filter failures (\f -> f.server == server)) of

757 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}

769 Just {value: f} -> {server: server, failedAt: t, failures: min (f.failures + 1) 10, firstFailedAt: f.firstFailedAt}

758 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}

770 Nothing {} -> {server: server, failedAt: t, failures: 1, firstFailedAt: t}

759− set *forwardFailures = union others [new]

771+ set *forwardFailures = union (filter failures (\f -> f.server != server)) [new]

760

772

761clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))

773clearFailure = \server -> atomic (pruneRef forwardFailures (\f -> f.server != server))

762

774

@@ -782,7 +794,7 @@ dedupMessageIds : [MessageIdHex] -> [MessageIdHex]

782dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)

794dedupMessageIds = \ids -> takeRelation maxMessageIds (unique ids)

783

795

784presenceForKey = \presence pubkey t ->

796presenceForKey = \presence pubkey t ->

785− do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p

797+ filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)

786

798

787-- Pick local if present, else the remote with the latest expiresAt; "" when none.

799-- Pick local if present, else the remote with the latest expiresAt; "" when none.

788bestServerForKey = \entries ->

800bestServerForKey = \entries ->

@@ -796,7 +808,7 @@ hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey

796

808

797collectPeers : [a] -> (a -> Text) -> [Text]

809collectPeers : [a] -> (a -> Text) -> [Text]

798collectPeers = \items getServer ->

810collectPeers = \items getServer ->

799− unique (do x <- items; let s = getServer x; where not (isLocalServer s); yield s)

811+ unique (filter (pluck items getServer) (\s -> not (isLocalServer s)))

800

812

801knownPeerServersFromPresence : [PresenceEntry] -> [Text]

813knownPeerServersFromPresence : [PresenceEntry] -> [Text]

802knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)

814knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)

@@ -805,7 +817,7 @@ knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.serve

805mergePresence = \existing incoming ->

817mergePresence = \existing incoming ->

806 fold (\acc e ->

818 fold (\acc e ->

807 if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt) then acc

819 if any acc (\a -> a.pubkey == e.pubkey && a.server == e.server && a.expiresAt >= e.expiresAt) then acc

808− else union (do x <- acc; where x.pubkey != e.pubkey || x.server != e.server; yield x) [e]) existing incoming

820+ else union (filter acc (\x -> x.pubkey != e.pubkey || x.server != e.server)) [e]) existing incoming

809

821

810-- ============================================

822-- ============================================

811-- Pruning

823-- Pruning

@@ -827,9 +839,9 @@ pruneExpired = \t -> atomic do

827 pruneRef pendingFederationCancels (\x -> t - x.createdAt < messageMaxAge)

839 pruneRef pendingFederationCancels (\x -> t - x.createdAt < messageMaxAge)

828 -- presence: remember locally-expired keys for offline gossip

840 -- presence: remember locally-expired keys for offline gossip

829 pres <- *presence

841 pres <- *presence

830− let keptPres = do p <- pres; where p.expiresAt > t; yield p

842+ let keptPres = filter pres (\p -> p.expiresAt > t)

831 set *presence = keptPres

843 set *presence = keptPres

832− yield (do p <- pres; where p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server); yield p)

844+ yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && none keptPres (\k -> k.pubkey == p.pubkey && isLocalServer k.server)))

833

845

834-- ============================================

846-- ============================================

835-- Federation gossip

847-- Federation gossip

@@ -852,8 +864,7 @@ broadcastOfflineGossip = \t expiredLocal -> do

852 presence <- *presence

864 presence <- *presence

853 let peers = knownPeerServersFromPresence presence

865 let peers = knownPeerServersFromPresence presence

854 when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do

866 when (federationEnabled && count peers > 0 && count expiredLocal > 0) (do

855− let events = do p <- expiredLocal; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}

867+ sendBatchedGossip peers (pluck expiredLocal (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops)

856− sendBatchedGossip peers events gossipOriginHops)

857

868

858-- ============================================

869-- ============================================

859-- Forward retries and acks

870-- Forward retries and acks

@@ -864,9 +875,9 @@ retryPendingForwards = do

864 fwds <- *forwards

875 fwds <- *forwards

865 failures <- *forwardFailures

876 failures <- *forwardFailures

866 blocked <- *blockedPeers

877 blocked <- *blockedPeers

867− let servers = do s <- collectPeers fwds (\f -> f.toServer); where shouldRetryServer s t failures && not (isBlockedPeer s blocked); yield s

878+ let servers = filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not (elem blocked s))

868 forEach servers (\s -> s) (\server -> do

879 forEach servers (\s -> s) (\server -> do

869− let serverFwds = takeRelation maxBatchSize (do f <- fwds; where f.toServer == server; yield f)

880+ let serverFwds = takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))

870 forEach serverFwds (\f -> f.messageId) (\f -> do

881 forEach serverFwds (\f -> f.messageId) (\f -> do

871 t2 <- now

882 t2 <- now

872 auth <- generateFederationAuth ("forward:" ++ f.messageId) t2

883 auth <- generateFederationAuth ("forward:" ++ f.messageId) t2

@@ -888,7 +899,7 @@ retryPendingForwards = do

888

899

889sendAcksAndClean = \pendingAcks ->

900sendAcksAndClean = \pendingAcks ->

890 forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do

901 forEach (collectPeers pendingAcks (\a -> a.originServer)) (\s -> s) (\server -> do

891− let mids = do a <- pendingAcks; where a.originServer == server; yield a.messageId

902+ let mids = pluck (filter pendingAcks (\a -> a.originServer == server)) (\a -> a.messageId)

892 t <- now

903 t <- now

893 auth <- generateFederationAuth "ack" t

904 auth <- generateFederationAuth "ack" t

894 result <- fetch (federationUrl server) (FederationAck {

905 result <- fetch (federationUrl server) (FederationAck {

@@ -901,20 +912,16 @@ sendAcksAndClean = \pendingAcks ->

901 clearFailure server

912 clearFailure server

902 atomic do

913 atomic do

903 cur <- *ackedIds

914 cur <- *ackedIds

904− set *ackedIds = do a <- cur; where not (a.originServer == server && any mids (\m -> m == a.messageId)); yield a)

915+ set *ackedIds = filter cur (\a -> not (a.originServer == server && elem mids a.messageId)))

905

916

906-- Forward queued messages to a server when a remote key comes online.

917-- Forward queued messages to a server when a remote key comes online.

907forwardQueuedToServer = \pubkey server t -> atomic do

918forwardQueuedToServer = \pubkey server t -> atomic do

908 msgs <- *messages

919 msgs <- *messages

909 fwds <- *forwards

920 fwds <- *forwards

910− let candidates = do

911− m <- msgs

912− where m.toKey == pubkey

913− where none fwds (\f -> f.messageId == m.messageId && f.toServer == server)

921+ let mine = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.messageId == m.messageId && f.toServer == server))

914− let origin = if m.originServer == "" then serverHostN else m.originServer

922+ let candidates = pluck mine (\m ->

915− yield {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: origin, retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey}

923+ {messageId: m.messageId, toServer: server, createdAt: t, toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, originServer: (if m.originServer == "" then serverHostN else m.originServer), retries: 0, expiresAt: m.expiresAt, priority: m.priority, senderKey: m.senderKey})

916− let capped = takeRelation maxForwardsPerServer candidates

917− set *forwards = capList (union fwds capped) maxForwardsEntries (\f -> f.createdAt)

924+ set *forwards = capList (union fwds (takeRelation maxForwardsPerServer candidates)) maxForwardsEntries (\f -> f.createdAt)

918

925

919-- ============================================

926-- ============================================

920-- Background loops

927-- Background loops

@@ -927,7 +934,7 @@ pruneAndRetry = \t -> do

927 when federationEnabled (fork retryPendingForwards)

934 when federationEnabled (fork retryPendingForwards)

928 when federationEnabled (do

935 when federationEnabled (do

929 aids <- *ackedIds

936 aids <- *ackedIds

930− let pendingAcks = do a <- aids; where not (isLocalServer a.originServer); yield a

937+ let pendingAcks = filter aids (\a -> not (isLocalServer a.originServer))

931 fork (sendAcksAndClean pendingAcks))

938 fork (sendAcksAndClean pendingAcks))

932 -- Refresh local presence for active sessions every ~5 minutes

939 -- Refresh local presence for active sessions every ~5 minutes

933 lastRefresh <- *lastPresenceRefresh

940 lastRefresh <- *lastPresenceRefresh

@@ -940,18 +947,15 @@ refreshLocalPresence = \t -> do

940 refreshState <- atomic do

947 refreshState <- atomic do

941 sessions <- *sessions

948 sessions <- *sessions

942 presence <- *presence

949 presence <- *presence

943− let activeKeys = unique (do s <- sessions; yield s.pubkey)

950+ let activeKeys = unique (pluck sessions (\s -> s.pubkey))

944− let updated = do

951+ let updated = pluck presence (\p ->

945− p <- presence

946− yield (if isLocalServer p.server && any activeKeys (\k -> k == p.pubkey) && p.expiresAt - t < sessionExpiry / 2

952+ if isLocalServer p.server && elem activeKeys p.pubkey && p.expiresAt - t < sessionExpiry / 2

947− then {pubkey: p.pubkey, server: p.server, expiresAt: t + sessionExpiry, timestamp: t}

953+ then {pubkey: p.pubkey, server: p.server, expiresAt: t + sessionExpiry, timestamp: t}

948− else p)

954+ else p)

949 set *presence = updated

955 set *presence = updated

950− let refreshed = do p <- updated; where isLocalServer p.server && p.expiresAt == t + sessionExpiry; yield p

956+ yield {refreshed: filter updated (\p -> isLocalServer p.server && p.expiresAt == t + sessionExpiry), peers: knownPeerServersFromPresence updated}

951− let peers = knownPeerServersFromPresence updated

952− yield {refreshed: refreshed, peers: peers}

953 when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do

957 when (count refreshState.refreshed > 0 && count refreshState.peers > 0) (do

954− let events = do p <- refreshState.refreshed; yield {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]}

958+ let events = pluck refreshState.refreshed (\p -> {eventType: GossipOnline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: clampTtl (sessionExpiry / 1000), seenBy: [serverHostN]})

955 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))

959 fork (sendBatchedGossip refreshState.peers events gossipOriginHops))

956

960

957-- ============================================

961-- ============================================

@@ -972,9 +976,8 @@ handleChallenge = \pubkey clientIp respond -> do

972 challenge <- genToken

976 challenge <- genToken

973 result <- atomic do

977 result <- atomic do

974 ch <- *challenges

978 ch <- *challenges

975− let active = do c <- ch; where c.expiresAt > t; yield c

979+ let active = filter ch (\c -> c.expiresAt > t)

976− let activeForKey = count (do c <- active; where c.pubkey == pubkey; yield c)

977− let cappedActive = if activeForKey >= maxChallengesPerKey

980+ let cappedActive = if countWhere active (\c -> c.pubkey == pubkey) >= maxChallengesPerKey

978 then evictOldest active (\c -> if c.pubkey == pubkey then c.expiresAt else 999999999999)

981 then evictOldest active (\c -> if c.pubkey == pubkey then c.expiresAt else 999999999999)

979 else active

982 else active

980 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}

983 if count cappedActive >= maxChallenges then yield {ok: False {}, error: "capacity"}

@@ -1006,26 +1009,23 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond ->

1006 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))

1009 if not (verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature))

1007 then yield {valid: False {}, peers: []}

1010 then yield {valid: False {}, peers: []}

1008 else do

1011 else do

1009− set *challenges = do c <- ch; where c.pubkey != pubkey || c.challenge != challenge; yield c

1012+ set *challenges = filter ch (\c -> c.pubkey != pubkey || c.challenge != challenge)

1010 dp <- *drainPhase

1013 dp <- *drainPhase

1011 if dp >= 1 then yield {valid: False {}, peers: []}

1014 if dp >= 1 then yield {valid: False {}, peers: []}

1012 else do

1015 else do

1013 ses <- *sessions

1016 ses <- *sessions

1014− let kept = if revokeOthers then (do s <- ses; where s.pubkey != pubkey; yield s) else ses

1017+ let kept = if revokeOthers then filter ses (\s -> s.pubkey != pubkey) else ses

1018+ let forKey = filter kept (\s -> s.pubkey == pubkey)

1015− let perKeyOver = count (do s <- kept; where s.pubkey == pubkey; yield s) >= maxSessionsPerKey

1019+ let afterPerKey = if count forKey >= maxSessionsPerKey

1016− let afterPerKey = if perKeyOver

1017− then let forKey = do s <- kept; where s.pubkey == pubkey; yield s in

1018− let others = do s <- kept; where s.pubkey != pubkey; yield s in

1019− union others (evictOldest forKey (\s -> s.expiresAt))

1020+ then union (filter kept (\s -> s.pubkey != pubkey)) (evictOldest forKey (\s -> s.expiresAt))

1020 else kept

1021 else kept

1021 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

1022 let newSession = {token: token, pubkey: pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}

1022 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)

1023 set *sessions = capList (union afterPerKey [newSession]) maxSessions (\s -> s.expiresAt)

1023 pres <- *presence

1024 pres <- *presence

1024− let withoutLocal = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p

1025+ let withoutLocal = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))

1025 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)

1026 let newPres = capList (union withoutLocal [{pubkey: pubkey, server: serverHostN, expiresAt: expiry, timestamp: t}]) maxPresenceEntries (\p -> p.expiresAt)

1026 set *presence = newPres

1027 set *presence = newPres

1027− let peers = knownPeerServersFromPresence newPres

1028− yield {valid: True {}, peers: peers}

1028+ yield {valid: True {}, peers: knownPeerServersFromPresence newPres}

1029 if not result.valid then yield (respond {token: "", expiresAt: 0, error: "invalid"})

1030 else do

1031 when federationEnabled

@@ -1044,7 +1044,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do

1044 let found = any ses (\s -> s.pubkey == caller && s.token == targetToken)

1045 if not found then yield False

1046 else do

1047− set *sessions = do s <- ses; where s.pubkey != caller || s.token != targetToken; yield s

1047+ set *sessions = filter ses (\s -> s.pubkey != caller || s.token != targetToken)

1048 yield True

1049 yield (respond {ok: result, authorized: True {}}))

1050

@@ -1060,14 +1060,14 @@ handleLogout = \authorization clientIp respond -> do

1060 Authorized {pubkey, ip} ->

1061 if not (matchIp clientIp ip) then yield {ok: False {}, pubkey: "", peers: []}

1062 else do

1063− set *sessions = do s <- ses; where s.token != token; yield s

1063+ set *sessions = filter ses (\s -> s.token != token)

1064 -- If no remaining sessions for this key, drop local presence

1065 after <- *sessions

1066 let stillHas = any after (\s -> s.pubkey == pubkey)

1067 peers <- if stillHas then yield []

1068 else do

1069 pres <- *presence

1070− set *presence = do p <- pres; where p.pubkey != pubkey || not (isLocalServer p.server); yield p

1070+ set *presence = filter pres (\p -> p.pubkey != pubkey || not (isLocalServer p.server))

1071 yield (knownPeerServersFromPresence pres)

1072 yield {ok: True {}, pubkey: pubkey, peers: peers}

1073 when (removed.ok && removed.pubkey != "" && count removed.peers > 0 && federationEnabled)

@@ -1111,7 +1111,7 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -

1111 Just {value: s} -> do

1112 acquired <- atomic do

1113 aps <- *activePolls

1114− let perKey = count (do p <- aps; where p.pubkey == s.pubkey; yield p)

1114+ let perKey = countWhere aps (\p -> p.pubkey == s.pubkey)

1115 if perKey >= maxPollsPerKey then yield False

1116 else do

1117 set *activePolls = capList (union aps [{pubkey: s.pubkey, startedAt: t}]) maxActivePolls (\p -> p.startedAt)

@@ -1127,18 +1127,18 @@ pollOnce = \token cursor cursorWasReset watchKeys clientIp serverGen respond t -

1127 then atomic do

1128 t2 <- now

1129 pres <- *presence

1130− yield (do k <- watchKeys; yield {pubkey: k, online: count (presenceForKey pres k t2) > 0})

1130+ yield (pluck watchKeys (\k -> {pubkey: k, online: count (presenceForKey pres k t2) > 0}))

1131 else yield []

1132 drains <- atomic do

1133 se <- *senderExpirations

1134− let myExp = do x <- se; where x.senderKey == s.pubkey; yield {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt}

1134+ let myExp = pluck (filter se (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, expiredAt: x.expiredAt})

1135− set *senderExpirations = do x <- se; where x.senderKey != s.pubkey; yield x

1135+ set *senderExpirations = filter se (\x -> x.senderKey != s.pubkey)

1136 sd <- *senderDeliveries

1137− let myDel = do x <- sd; where x.senderKey == s.pubkey; yield {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt}

1137+ let myDel = pluck (filter sd (\x -> x.senderKey == s.pubkey)) (\x -> {messageId: x.messageId, toKey: x.toKey, deliveredAt: x.deliveredAt})

1138− set *senderDeliveries = do x <- sd; where x.senderKey != s.pubkey; yield x

1138+ set *senderDeliveries = filter sd (\x -> x.senderKey != s.pubkey)

1139 cn <- *cancelNotifications

1140− let myCancels = do x <- cn; where x.toKey == s.pubkey; yield {messageId: x.messageId, cancelledAt: x.cancelledAt}

1140+ let myCancels = pluck (filter cn (\x -> x.toKey == s.pubkey)) (\x -> {messageId: x.messageId, cancelledAt: x.cancelledAt})

1141− set *cancelNotifications = do x <- cn; where x.toKey != s.pubkey; yield x

1141+ set *cancelNotifications = filter cn (\x -> x.toKey != s.pubkey)

1142 yield {expirations: myExp, deliveries: myDel, cancellations: myCancels}

1143 let isHeartbeat = pollResult.authorized && count pollResult.events == 0

1144 yield (respond {events: pollResult.events, cursor: pollResult.cursor, authorized: pollResult.authorized, rateLimited: False {}, draining: False {}, generation: serverGen, presenceUpdates: presenceUpdates, heartbeat: isHeartbeat, retryAfterMs: 0, cursorReset: cursorWasReset, hasMore: pollResult.hasMore, cursorStale: False {}, messageExpirations: drains.expirations, messageDeliveries: drains.deliveries, droppedDeliveryReceipts: 0, droppedExpirationNotifs: 0, loadPct: 0, messageCancellations: drains.cancellations})

@@ -1160,27 +1160,26 @@ doPoll = \s t -> do

1160 yield {authorized: True {}, events: r.events, cursor: r.cursor, hasMore: r.hasMore}

1161

1162collectPollEvents = \messages pubkey cursor t ->

1163− let myMsgs = do m <- messages; where m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor; yield m in

1163+ let myMsgs = filter messages (\m -> m.toKey == pubkey && m.expiresAt > t && m.receivedAt > cursor) in

1164− let sorted = sortBy (\m -> m.receivedAt) myMsgs in

1165− let limited = takeRelation maxPollEvents sorted in

1164+ let limited = takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs) in

1166− let evts = do m <- limited; yield {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}} in

1165+ let evts = pluck limited (\m -> {id: m.messageId, eventType: "message", payload: {toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp, expiresAt: m.expiresAt}}) in

1167− let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) cursor limited in

1166+ {events: evts, cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}

1168− {events: evts, cursor: lastReceivedAt, hasMore: count myMsgs > count limited}

1169

1167

1170ackPolledMessages = \pubkey cursor t -> do

1168ackPolledMessages = \pubkey cursor t -> do

1171 msgs <- *messages

1169 msgs <- *messages

1172− let toAck = do m <- msgs; where m.toKey == pubkey && m.receivedAt <= cursor; yield m

1170+ let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor

1171+ let toAck = filter msgs matches

1173 if count toAck == 0 then yield {pendingAcks: []}

1172 if count toAck == 0 then yield {pendingAcks: []}

1174 else do

1173 else do

1175− set *messages = do m <- msgs; where not (m.toKey == pubkey && m.receivedAt <= cursor); yield m

1174+ set *messages = filter msgs (\m -> not (matches m))

1176 fwds <- *forwards

1175 fwds <- *forwards

1177− let mids = do m <- toAck; yield m.messageId

1176+ let mids = pluck toAck (\m -> m.messageId)

1178− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any mids (\mid -> mid == f.messageId)); yield f

1177+ set *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem mids f.messageId))

1179− let ackedFederated = do m <- toAck; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1178+ let ackedFederated = pluck (filter toAck (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})

1180 aids <- *ackedIds

1179 aids <- *ackedIds

1181 set *ackedIds = capList (union aids ackedFederated) maxAckedIds (\a -> a.ackedAt)

1180 set *ackedIds = capList (union aids ackedFederated) maxAckedIds (\a -> a.ackedAt)

1182 -- Sender delivery receipts for locally-originated messages

1181 -- Sender delivery receipts for locally-originated messages

1183− let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}

1182+ let localDelivered = pluck (filter toAck (\m -> m.senderKey != "" && m.originServer == "")) (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})

1184 sd <- *senderDeliveries

1183 sd <- *senderDeliveries

1185 set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)

1184 set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)

1186 yield {pendingAcks: ackedFederated}

1185 yield {pendingAcks: ackedFederated}

@@ -1230,7 +1229,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg

1230 recvRates <- *recvRates

1229 recvRates <- *recvRates

1231 let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t

1230 let rr = checkRate recvRates to maxRecvRatePerKey recvRateWindowMs 5000 t

1232 set *recvRates = rr.entries

1231 set *recvRates = rr.entries

1233− let recipientCount = count (do m <- msgs; where m.toKey == to; yield m)

1232+ let recipientCount = countWhere msgs (\m -> m.toKey == to)

1234 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}

1233 if senderKey == to then yield {status: MsRejected {}, remoteServers: []}

1235 else if isDup then yield {status: MsDuplicate {}, remoteServers: []}

1234 else if isDup then yield {status: MsDuplicate {}, remoteServers: []}

1236 else if sr.limited || pr.limited || rr.limited then yield {status: MsRateLimited {}, remoteServers: []}

1235 else if sr.limited || pr.limited || rr.limited then yield {status: MsRateLimited {}, remoteServers: []}

@@ -1250,7 +1249,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs effMsgId expiresInMs msg

1250 forwardsEnqueued <- if federationEnabled && count remoteServers > 0

1249 forwardsEnqueued <- if federationEnabled && count remoteServers > 0

1251 then do

1250 then do

1252 fwds <- *forwards

1251 fwds <- *forwards

1253− let newFwds = do s <- remoteServers; where none fwds (\f -> f.messageId == effMsgId && f.toServer == s); yield {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey}

1252+ let newFwds = pluck (filter remoteServers (\s -> none fwds (\f -> f.messageId == effMsgId && f.toServer == s))) (\s -> {messageId: effMsgId, toServer: s, createdAt: t, toKey: to, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: serverHostN, retries: 0, expiresAt: expiresAt, priority: msgPriority, senderKey: senderKey})

1254 set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)

1253 set *forwards = capList (union fwds newFwds) maxForwardsEntries (\f -> f.createdAt)

1255 yield (count newFwds > 0)

1254 yield (count newFwds > 0)

1256 else yield False

1255 else yield False

@@ -1263,7 +1262,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1263 forEach servers (\s -> s) (\server -> do

1262 forEach servers (\s -> s) (\server -> do

1264 info <- atomic do

1263 info <- atomic do

1265 fwds <- *forwards

1264 fwds <- *forwards

1266− case headOf (do f <- fwds; where f.messageId == effMsgId && f.toServer == server; yield f) of

1265+ case headOf (filter fwds (\f -> f.messageId == effMsgId && f.toServer == server)) of

1267 Nothing {} -> yield Nothing {}

1266 Nothing {} -> yield Nothing {}

1268 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})

1267 Just {value: f} -> yield (Just {value: {expiresAt: f.expiresAt, priority: f.priority}})

1269 case info of

1268 case info of

@@ -1284,7 +1283,7 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1284 clearFailure server

1283 clearFailure server

1285 atomic do

1284 atomic do

1286 fwds <- *forwards

1285 fwds <- *forwards

1287− set *forwards = do f <- fwds; where not (f.messageId == effMsgId && f.toServer == server); yield f

1286+ set *forwards = filter fwds (\f -> not (f.messageId == effMsgId && f.toServer == server))

1288 else yield {})

1287 else yield {})

1289

1288

1290-- ============================================

1289-- ============================================

@@ -1293,54 +1292,45 @@ forkForwardsTo = \effMsgId to encryptedBlob clampedTs servers ->

1293

1292

1294handleLookupPresence = \pubkey authorization clientIp respond -> do

1293handleLookupPresence = \pubkey authorization clientIp respond -> do

1295 t <- now

1294 t <- now

1296− g <- checkGlobalRate t

1297− if g then yield (respond {server: "", authorized: False {}, error: "rate_limited"})

1295+ let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1298− else withSessionAuth authorization t clientIp

1296+ gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do

1299− (yield (respond {server: "", authorized: False {}, error: ""}))

1300− (\senderKey -> do

1301− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1297+ lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1302− if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})

1298+ if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})

1303− else do

1299+ else do

1304− pres <- *presence

1300+ pres <- *presence

1305− let found = bestServerForKey (presenceForKey pres pubkey t)

1301+ let found = bestServerForKey (presenceForKey pres pubkey t)

1306− let masked = if found == "" then "" else if isLocalServer found then found else "remote"

1302+ let masked = if found == "" then "" else if isLocalServer found then found else "remote"

1307− yield (respond {server: masked, authorized: True {}, error: ""}))

1303+ yield (respond {server: masked, authorized: True {}, error: ""}))

1308

1304

1309handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1305handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do

1310 t <- now

1306 t <- now

1311− g <- checkGlobalRate t

1312− if g then yield (respond {results: [], authorized: False {}, error: "rate_limited"})

1307+ let mkErr = \e -> {results: [], authorized: False {}, error: if e == "rate_limited" then "rate_limited" else ""}

1313− else withSessionAuth authorization t clientIp

1308+ gateAuthNoDrain authorization clientIp t respond mkErr (\senderKey -> do

1314− (yield (respond {results: [], authorized: False {}, error: ""}))

1315− (\senderKey -> do

1316− lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1309+ lim <- checkPubkeyRateGate lookupRates senderKey maxLookupRatePerKey lookupRateWindowMs t

1317− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1310+ if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1318− else do

1311+ else do

1312+ pres <- *presence

1313+ let results = do

1319− let capped = takeRelation 50 pubkeys

1314+ k <- takeRelation 50 pubkeys

1320− pres <- *presence

1321− let results = do

1322− k <- capped

1323− let found = bestServerForKey (presenceForKey pres k t)

1315+ let found = bestServerForKey (presenceForKey pres k t)

1324− let masked = if found == "" then "" else if isLocalServer found then found else "remote"

1316+ let masked = if found == "" then "" else if isLocalServer found then found else "remote"

1325− yield {pubkey: k, server: masked}

1317+ yield {pubkey: k, server: masked}

1326− yield (respond {results: results, authorized: True {}, error: ""}))

1318+ yield (respond {results: results, authorized: True {}, error: ""}))

1327

1319

1328handleMessageStatus = \to messageId authorization clientIp respond -> do

1320handleMessageStatus = \to messageId authorization clientIp respond -> do

1329 t <- now

1321 t <- now

1330− g <- checkGlobalRate t

1331− if g then yield (respond {status: "", authorized: False {}})

1322+ let mkErr = \_ -> {status: "", authorized: False {}}

1332− else withSessionAuth authorization t clientIp

1323+ gateAuthNoDrain authorization clientIp t respond mkErr (\_ -> do

1333− (yield (respond {status: "", authorized: False {}}))

1334− (\_ -> do

1335− msgs <- *messages

1324+ msgs <- *messages

1336− fwds <- *forwards

1325+ fwds <- *forwards

1337− aids <- *ackedIds

1326+ aids <- *ackedIds

1338− dids <- *deliveredIds

1327+ dids <- *deliveredIds

1339− let s = if any msgs (\m -> m.messageId == messageId && m.toKey == to) then "pending"

1328+ let matches = \x -> x.messageId == messageId && x.toKey == to

1329+ let s = if any msgs matches then "pending"

1340− else if any fwds (\f -> f.messageId == messageId && f.toKey == to) then "forwarding"

1330+ else if any fwds matches then "forwarding"

1341− else if hasMessage messageId to aids || hasMessage messageId to dids then "delivered"

1331+ else if any aids matches || any dids matches then "delivered"

1342− else "unknown"

1332+ else "unknown"

1343− yield (respond {status: s, authorized: True {}}))

1333+ yield (respond {status: s, authorized: True {}}))

1344

1334

1345-- ============================================

1335-- ============================================

1346-- Handler: ack / list / delete / cancel

1336-- Handler: ack / list / delete / cancel

@@ -1348,125 +1338,99 @@ handleMessageStatus = \to messageId authorization clientIp respond -> do

1348

1338

1349handleAckMessages = \messageIds authorization clientIp respond -> do

1339handleAckMessages = \messageIds authorization clientIp respond -> do

1350 t <- now

1340 t <- now

1351− dp <- *drainPhase

1352− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "draining"})

1341+ let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}

1353− else do

1354− g <- checkGlobalRate t

1355− if g then yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "rate_limited"})

1356− else withSessionAuth authorization t clientIp

1342+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1357− (yield (respond {ok: False {}, authorized: False {}, ackedCount: 0, error: "unauthorized"}))

1358− (\pubkey -> do

1359− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1343+ lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1360− if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})

1344+ if lim then yield (respond {ok: False {}, authorized: True {}, ackedCount: 0, error: "rate_limited"})

1345+ else do

1346+ let dedupedIds = dedupMessageIds messageIds

1347+ let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1348+ result <- atomic do

1349+ msgs <- *messages

1350+ let toAck = filter msgs matches

1351+ if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}

1361 else do

1352 else do

1362− let dedupedIds = dedupMessageIds messageIds

1353+ set *messages = filter msgs (\m -> not (matches m))

1363− result <- atomic do

1364− msgs <- *messages

1365− let toAck = do m <- msgs; where m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid); yield m

1366− if count toAck == 0 then yield {ackedCount: 0, pendingAcks: []}

1367− else do

1368− set *messages = do m <- msgs; where not (m.toKey == pubkey && any dedupedIds (\mid -> m.messageId == mid)); yield m

1369− fwds <- *forwards

1354+ fwds <- *forwards

1370− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any dedupedIds (\mid -> f.messageId == mid)); yield f

1355+ set *forwards = filter fwds (\f -> not (matches f))

1371− let ackedFed = do m <- toAck; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1356+ let ackedFed = pluck (filter toAck (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})

1372− aids <- *ackedIds

1357+ aids <- *ackedIds

1373− set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)

1358+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)

1374− let localDelivered = do m <- toAck; where m.senderKey != "" && m.originServer == ""; yield {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t}

1359+ let localDelivered = pluck (filter toAck (\m -> m.senderKey != "" && m.originServer == "")) (\m -> {senderKey: m.senderKey, messageId: m.messageId, toKey: m.toKey, deliveredAt: t})

1375− sd <- *senderDeliveries

1360+ sd <- *senderDeliveries

1376− set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)

1361+ set *senderDeliveries = capList (union sd localDelivered) 10000 (\d -> d.deliveredAt)

1377− yield {ackedCount: count toAck, pendingAcks: ackedFed}

1362+ yield {ackedCount: count toAck, pendingAcks: ackedFed}

1378− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))

1363+ fork (when federationEnabled (sendAcksAndClean result.pendingAcks))

1379− yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))

1364+ yield (respond {ok: True {}, authorized: True {}, ackedCount: result.ackedCount, error: ""}))

1380

1365

1381handleListMessages = \offset authorization clientIp respond -> do

1366handleListMessages = \offset authorization clientIp respond -> do

1382 t <- now

1367 t <- now

1383− dp <- *drainPhase

1384− if dp >= 2 then yield (respond {messages: [], authorized: False {}, count: 0, error: ""})

1368+ let mkErr = \e -> {messages: [], authorized: False {}, count: 0, error: if e == "rate_limited" then "rate_limited" else ""}

1385− else do

1386− g <- checkGlobalRate t

1387− if g then yield (respond {messages: [], authorized: False {}, count: 0, error: "rate_limited"})

1388− else withSessionAuth authorization t clientIp

1369+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1389− (yield (respond {messages: [], authorized: False {}, count: 0, error: ""}))

1390− (\pubkey -> do

1391− lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t

1370+ lim <- checkPubkeyRateGate lookupRates pubkey maxLookupRatePerKey lookupRateWindowMs t

1392− if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1371+ if lim then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})

1393− else do

1372+ else do

1394− msgs <- *messages

1373+ msgs <- *messages

1395− let myMsgs = sortBy (\m -> m.receivedAt) (do m <- msgs; where m.toKey == pubkey && m.expiresAt > t; yield m)

1374+ let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && m.expiresAt > t))

1396− let total = count myMsgs

1375+ let total = count myMsgs

1397− let effectiveOffset = clamp 0 total offset

1376+ let effectiveOffset = clamp 0 total offset

1398− -- Index using a fold so we can drop items by position; simple O(n).

1377+ -- Drop the first `effectiveOffset` items via fold (O(n)).

1399− let after = (fold (\acc m -> if acc.i < effectiveOffset then {i: acc.i + 1, out: acc.out} else {i: acc.i, out: union acc.out [m]}) {i: 0, out: []} myMsgs).out

1378+ let after = (fold (\acc m -> if acc.i < effectiveOffset then {i: acc.i + 1, out: acc.out} else {i: acc.i, out: union acc.out [m]}) {i: 0, out: []} myMsgs).out

1400− let capped = takeRelation maxPollEvents after

1401− let result = do m <- capped; yield {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt}

1379+ let result = pluck (takeRelation maxPollEvents after) (\m -> {messageId: m.messageId, timestamp: m.timestamp, receivedAt: m.receivedAt})

1402− yield (respond {messages: result, authorized: True {}, count: total, error: ""}))

1380+ yield (respond {messages: result, authorized: True {}, count: total, error: ""}))

1403

1381

1404handleDeleteMessages = \messageIds authorization clientIp respond -> do

1382handleDeleteMessages = \messageIds authorization clientIp respond -> do

1405 t <- now

1383 t <- now

1406− dp <- *drainPhase

1407− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: ""})

1384+ let mkErr = \e -> {ok: False {}, authorized: False {}, deletedCount: 0, error: if e == "draining" then "" else e}

1408− else do

1409− g <- checkGlobalRate t

1410− if g then yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "rate_limited"})

1411− else withSessionAuth authorization t clientIp

1385+ gateAuth authorization clientIp t respond mkErr (\pubkey -> do

1412− (yield (respond {ok: False {}, authorized: False {}, deletedCount: 0, error: "unauthorized"}))

1413− (\pubkey -> do

1414− lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1386+ lim <- checkPubkeyRateGate clientWriteRates pubkey maxClientWriteRatePerKey clientWriteRateWindowMs t

1415− if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})

1387+ if lim then yield (respond {ok: False {}, authorized: True {}, deletedCount: 0, error: "rate_limited"})

1388+ else do

1389+ let dedupedIds = dedupMessageIds messageIds

1390+ let matches = \x -> x.toKey == pubkey && elem dedupedIds x.messageId

1391+ result <- atomic do

1392+ msgs <- *messages

1393+ let toDel = filter msgs matches

1394+ if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}

1416 else do

1395 else do

1417− let dedupedIds = dedupMessageIds messageIds

1396+ set *messages = filter msgs (\m -> not (matches m))

1418− result <- atomic do

1419− msgs <- *messages

1420− let toDel = do m <- msgs; where m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId); yield m

1421− if count toDel == 0 then yield {deletedCount: 0, pendingAcks: []}

1422− else do

1423− set *messages = do m <- msgs; where not (m.toKey == pubkey && any dedupedIds (\mid -> mid == m.messageId)); yield m

1424− fwds <- *forwards

1397+ fwds <- *forwards

1425− set *forwards = do f <- fwds; where not (f.toKey == pubkey && any dedupedIds (\mid -> mid == f.messageId)); yield f

1398+ set *forwards = filter fwds (\f -> not (matches f))

1426− let federatedDeleted = do m <- toDel; where m.originServer != ""; yield m

1427− let ackedFed = do m <- federatedDeleted; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1399+ let ackedFed = pluck (filter toDel (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})

1428− aids <- *ackedIds

1400+ aids <- *ackedIds

1429− set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)

1401+ set *ackedIds = capList (union aids ackedFed) maxAckedIds (\a -> a.ackedAt)

1430− yield {deletedCount: count toDel, pendingAcks: ackedFed}

1402+ yield {deletedCount: count toDel, pendingAcks: ackedFed}

1431− fork (when federationEnabled (sendAcksAndClean result.pendingAcks))

1403+ fork (when federationEnabled (sendAcksAndClean result.pendingAcks))

1432− yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))

1404+ yield (respond {ok: True {}, authorized: True {}, deletedCount: result.deletedCount, error: ""}))

1433

1405

1434handleCancelMessage = \messageId to authorization clientIp respond -> do

1406handleCancelMessage = \messageId to authorization clientIp respond -> do

1435 t <- now

1407 t <- now

1436− dp <- *drainPhase

1437− if dp >= 2 then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: ""})

1408+ let mkErr = \e -> {ok: False {}, authorized: False {}, cancelled: False {}, error: if e == "draining" then "" else e}

1438− else do

1439− g <- checkGlobalRate t

1440− if g then yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "rate_limited"})

1441− else withSessionAuth authorization t clientIp

1409+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1442− (yield (respond {ok: False {}, authorized: False {}, cancelled: False {}, error: "unauthorized"}))

1443− (\senderKey -> do

1444− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1410+ lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1445− if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})

1411+ if lim then yield (respond {ok: False {}, authorized: True {}, cancelled: False {}, error: "rate_limited"})

1446− else do

1412+ else do

1413+ let matches = \x -> x.messageId == messageId && x.toKey == to

1447− result <- atomic do

1414+ result <- atomic do

1448− msgs <- *messages

1415+ msgs <- *messages

1449− let target = headOf (do m <- msgs; where m.messageId == messageId && m.toKey == to; yield m)

1416+ case headOf (filter msgs matches) of

1450− case target of

1451− Nothing {} -> yield {cancelled: False {}, cancelServers: []}

1417+ Nothing {} -> yield {cancelled: False {}, cancelServers: []}

1452− Just {value: msg} ->

1418+ Just {value: msg} ->

1453− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey

1419+ if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1454− then yield {cancelled: False {}, cancelServers: []}

1420+ then yield {cancelled: False {}, cancelServers: []}

1455− else do

1421+ else do

1456− set *messages = do m <- msgs; where not (m.messageId == messageId && m.toKey == to); yield m

1422+ set *messages = filter msgs (\m -> not (matches m))

1457− fwds <- *forwards

1423+ fwds <- *forwards

1458− let cancelledFwds = do f <- fwds; where f.messageId == messageId && f.toKey == to; yield f

1424+ let fwdServers = collectPeers (filter fwds matches) (\f -> f.toServer)

1459− let fwdServers = collectPeers cancelledFwds (\f -> f.toServer)

1425+ set *forwards = filter fwds (\f -> not (matches f))

1460− set *forwards = do f <- fwds; where not (f.messageId == messageId && f.toKey == to); yield f

1461− pres <- *presence

1426+ pres <- *presence

1462− let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)

1427+ let presServers = collectPeers (presenceForKey pres to t) (\p -> p.server)

1463− let allServers = unique (union presServers fwdServers)

1464− cn <- *cancelNotifications

1428+ cn <- *cancelNotifications

1465− set *cancelNotifications = capList (union cn [{toKey: to, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1429+ set *cancelNotifications = capList (union cn [{toKey: to, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1466− yield {cancelled: True {}, cancelServers: allServers}

1430+ yield {cancelled: True {}, cancelServers: unique (union presServers fwdServers)}

1467− when (result.cancelled && count result.cancelServers > 0 && federationEnabled)

1431+ when (result.cancelled && count result.cancelServers > 0 && federationEnabled)

1468− (fork (sendCancelToEach messageId to result.cancelServers))

1432+ (fork (sendCancelToEach messageId to result.cancelServers))

1469− yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))

1433+ yield (respond {ok: True {}, authorized: True {}, cancelled: result.cancelled, error: ""}))

1470

1434

1471sendCancelToEach = \messageId to servers ->

1435sendCancelToEach = \messageId to servers ->

1472 forEach servers (\s -> s) (\server -> do

1436 forEach servers (\s -> s) (\server -> do

@@ -1483,57 +1447,45 @@ sendCancelToEach = \messageId to servers ->

1483

1447

1484handleCancelBatch = \cancels authorization clientIp respond -> do

1448handleCancelBatch = \cancels authorization clientIp respond -> do

1485 t <- now

1449 t <- now

1486− dp <- *drainPhase

1487− if dp >= 2 then yield (respond {results: [], authorized: False {}, error: ""})

1450+ let mkErr = \e -> {results: [], authorized: False {}, error: if e == "draining" then "" else e}

1488− else do

1489− g <- checkGlobalRate t

1490− if g then yield (respond {results: [], authorized: False {}, error: "rate_limited"})

1491− else withSessionAuth authorization t clientIp

1451+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1492− (yield (respond {results: [], authorized: False {}, error: "unauthorized"}))

1493− (\senderKey -> do

1494− lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1452+ lim <- checkPubkeyRateGate clientWriteRates senderKey maxClientWriteRatePerKey clientWriteRateWindowMs t

1495− if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1453+ if lim then yield (respond {results: [], authorized: True {}, error: "rate_limited"})

1496− else do

1454+ else do

1497− results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1455+ results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1456+ let matches = \x -> x.messageId == e.messageId && x.toKey == e.to

1498− r <- atomic do

1457+ r <- atomic do

1499− msgs <- *messages

1458+ msgs <- *messages

1500− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.to; yield m)

1459+ case headOf (filter msgs matches) of

1501− case target of

1502− Nothing {} -> yield False

1460+ Nothing {} -> yield False

1503− Just {value: msg} ->

1461+ Just {value: msg} ->

1504− if msg.originServer != "" || msg.senderKey == "" || msg.senderKey != senderKey

1462+ if msg.originServer != "" || msg.senderKey != senderKey || msg.senderKey == ""

1505− then yield False

1463+ then yield False

1506− else do

1464+ else do

1507− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.to); yield m

1465+ set *messages = filter msgs (\m -> not (matches m))

1508− fwds <- *forwards

1466+ fwds <- *forwards

1509− set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.to); yield f

1467+ set *forwards = filter fwds (\f -> not (matches f))

1510− cn <- *cancelNotifications

1468+ cn <- *cancelNotifications

1511− set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1469+ set *cancelNotifications = capList (union cn [{toKey: e.to, messageId: e.messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1512− yield True

1470+ yield True

1513− yield {messageId: e.messageId, cancelled: r})

1471+ yield {messageId: e.messageId, cancelled: r})

1514− yield (respond {results: results, authorized: True {}, error: ""}))

1472+ yield (respond {results: results, authorized: True {}, error: ""}))

1515

1473

1516handleSendBatch = \messages authorization clientIp respond -> do

1474handleSendBatch = \messages authorization clientIp respond -> do

1517 t <- now

1475 t <- now

1518− dp <- *drainPhase

1519− if dp >= 2 then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})

1476+ let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}

1520− else do

1521− g <- checkGlobalRate t

1522− if g then yield (respond {results: [], authorized: False {}, retryAfterMs: 0})

1523− else withSessionAuth authorization t clientIp

1477+ gateAuth authorization clientIp t respond mkErr (\senderKey -> do

1524− (yield (respond {results: [], authorized: False {}, retryAfterMs: 0}))

1525− (\senderKey -> do

1526− results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do

1478+ results <- mapEach (takeRelation maxBatchSize messages) (\m -> m) (\m -> do

1527− effMsgId <- if m.messageId != "" then yield m.messageId else genToken

1479+ effMsgId <- if m.messageId != "" then yield m.messageId else genToken

1528− let clampedTs = clampTimestamp m.timestamp t

1480+ let clampedTs = clampTimestamp m.timestamp t

1529− let clampedPri = clamp 0 maxMessagePriority m.priority

1481+ let clampedPri = clamp 0 maxMessagePriority m.priority

1530− r <- if t - m.timestamp > messageMaxAge

1482+ r <- if t - m.timestamp > messageMaxAge

1531− then yield {status: MsRejected {}, remoteServers: []}

1483+ then yield {status: MsRejected {}, remoteServers: []}

1532− else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t

1484+ else atomic do sendMessageCore senderKey m.to m.encryptedBlob clampedTs effMsgId m.expiresInMs clampedPri t

1533− when (federationEnabled && count r.remoteServers > 0)

1485+ when (federationEnabled && count r.remoteServers > 0)

1534− (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))

1486+ (fork (forkForwardsTo effMsgId m.to m.encryptedBlob clampedTs r.remoteServers))

1535− yield {messageId: effMsgId, status: r.status})

1487+ yield {messageId: effMsgId, status: r.status})

1536− yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))

1488+ yield (respond {results: results, authorized: True {}, retryAfterMs: 0}))

1537

1489

1538-- ============================================

1490-- ============================================

1539-- Handler: federation receive

1491-- Handler: federation receive

@@ -1551,9 +1503,9 @@ handleVerifyToken = \token scope clientIp respond -> do

1551 else do

1503 else do

1552 found <- atomic do

1504 found <- atomic do

1553 pending <- *pendingVerifications

1505 pending <- *pendingVerifications

1554− let candidates = do p <- pending; where p.scope == scope && t - p.createdAt < verificationTokenExpiry; yield p

1506+ let candidates = filter pending (\p -> p.scope == scope && t - p.createdAt < verificationTokenExpiry)

1555 let match = any candidates (\p -> p.token == token)

1507 let match = any candidates (\p -> p.token == token)

1556− when match (set *pendingVerifications = do p <- pending; where p.scope != scope || p.token != token; yield p)

1508+ when match (set *pendingVerifications = filter pending (\p -> p.scope != scope || p.token != token))

1557 yield match

1509 yield match

1558 yield (respond {ok: found})

1510 yield (respond {ok: found})

1559

1511

@@ -1572,19 +1524,13 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation

1572 let expiresAt = min e.timestamp t + clampTtl e.ttl * 1000

1524 let expiresAt = min e.timestamp t + clampTtl e.ttl * 1000

1573 where expiresAt > t

1525 where expiresAt > t

1574 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}

1526 yield {pubkey: e.pubkey, server: e.server, expiresAt: expiresAt, timestamp: e.timestamp, eventType: e.eventType}

1575− let online = do v <- validEvents; where (case v.eventType of GossipOnline {} -> True {}; _ -> False {}); yield {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp}

1527+ let isOnline = \v -> case v.eventType of GossipOnline {} -> True {}; _ -> False {}

1576− let offline = do v <- validEvents; where (case v.eventType of GossipOffline {} -> True {}; _ -> False {}); yield {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp}

1528+ let isOffline = \v -> case v.eventType of GossipOffline {} -> True {}; _ -> False {}

1577− let withoutOffline = do

1529+ let online = pluck (filter validEvents isOnline) (\v -> {pubkey: v.pubkey, server: v.server, expiresAt: v.expiresAt, timestamp: v.timestamp})

1578− p <- pres

1530+ let offline = pluck (filter validEvents isOffline) (\v -> {pubkey: v.pubkey, server: v.server, timestamp: v.timestamp})

1579− where none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp)

1531+ let withoutOffline = filter pres (\p -> none offline (\o -> o.pubkey == p.pubkey && o.server == p.server && o.timestamp > p.timestamp))

1580− yield p

1581− let merged = mergePresence withoutOffline online

1582− set *presence = capList merged maxPresenceEntries (\p -> p.expiresAt)

1532+ set *presence = capList (mergePresence withoutOffline online) maxPresenceEntries (\p -> p.expiresAt)

1583− let newlyOnline = do

1584− e <- online

1585− where not (isLocalServer e.server)

1586− where none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt)

1533+ let newlyOnline = pluck (filter online (\e -> not (isLocalServer e.server) && none pres (\p -> p.pubkey == e.pubkey && p.server == e.server && p.expiresAt >= e.expiresAt))) (\e -> {pubkey: e.pubkey, server: e.server})

1587− yield {pubkey: e.pubkey, server: e.server}

1588 yield {newlyOnline: newlyOnline}

1534 yield {newlyOnline: newlyOnline}

1589 when (count valid.newlyOnline > 0)

1535 when (count valid.newlyOnline > 0)

1590 (fork (forwardForNewlyOnline valid.newlyOnline t))

1536 (fork (forwardForNewlyOnline valid.newlyOnline t))

@@ -1614,8 +1560,7 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro

1614 if isDup then yield "duplicate"

1560 if isDup then yield "duplicate"

1615 else if count msgs >= maxMessages then yield "global_full"

1561 else if count msgs >= maxMessages then yield "global_full"

1616 else

1562 else

1617− let rcCount = count (do m <- msgs; where m.toKey == toKey; yield m) in

1618− if rcCount >= maxMessagesPerRecipient then yield "mailbox_full"

1563+ if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"

1619 else do

1564 else do

1620 rr <- *recvRates

1565 rr <- *recvRates

1621 let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t

1566 let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t

@@ -1641,22 +1586,21 @@ handleFederationAck = \messageIds originServer fromServer requestTimestamp corre

1641 let dedupedIds = dedupMessageIds messageIds

1586 let dedupedIds = dedupMessageIds messageIds

1642 result <- atomic do

1587 result <- atomic do

1643 fwds <- *forwards

1588 fwds <- *forwards

1644− let removed = do f <- fwds; where f.toServer == fromServer && any dedupedIds (\m -> m == f.messageId); yield f

1589+ let isAcked = \f -> f.toServer == fromServer && elem dedupedIds f.messageId

1645− set *forwards = do f <- fwds; where not (f.toServer == fromServer && any dedupedIds (\m -> m == f.messageId)); yield f

1590+ let removed = filter fwds isAcked

1591+ set *forwards = filter fwds (\f -> not (isAcked f))

1646 -- Reclaim origin messages with no remaining forwards

1592 -- Reclaim origin messages with no remaining forwards

1647 msgs <- *messages

1593 msgs <- *messages

1648− let removedIds = do f <- removed; yield f.messageId

1594+ let removedIds = pluck removed (\f -> f.messageId)

1649 pres <- *presence

1595 pres <- *presence

1650 remainingFwds <- *forwards

1596 remainingFwds <- *forwards

1651− let toReap = do

1597+ let toReap = filter msgs (\m ->

1652− m <- msgs

1653− where any removedIds (\mid -> mid == m.messageId)

1598+ elem removedIds m.messageId

1654− where t - m.receivedAt >= ackGracePeriod

1599+ && t - m.receivedAt >= ackGracePeriod

1655− where none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)

1600+ && none pres (\p -> p.pubkey == m.toKey && isLocalServer p.server && p.expiresAt > t)

1656− where none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey)

1601+ && none remainingFwds (\f -> f.messageId == m.messageId && f.toKey == m.toKey))

1657− yield m

1658− set *messages = do m <- msgs; where none toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey); yield m

1602+ set *messages = filter msgs (\m -> none toReap (\r -> r.messageId == m.messageId && r.toKey == m.toKey))

1659− let perResult = do mid <- dedupedIds; yield {messageId: mid, found: any removed (\f -> f.messageId == mid)}

1603+ let perResult = pluck dedupedIds (\mid -> {messageId: mid, found: any removed (\f -> f.messageId == mid)})

1660 yield {ackedCount: count removed, results: perResult}

1604 yield {ackedCount: count removed, results: perResult}

1661 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))

1605 yield (respond {ok: True {}, error: "", ackedCount: result.ackedCount, results: result.results, retryAfterMs: 0}))

1662

1606

@@ -1664,15 +1608,15 @@ handleFederationCancel = \messageId toKey fromServer requestTimestamp correlatio

1664 t <- now

1608 t <- now

1665 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}

1609 let cancelRespond = \r -> respond {ok: r.ok, error: r.error, cancelled: False {}, retryAfterMs: r.retryAfterMs}

1666 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (do

1610 withFederationAuth authorization t cancelRespond fromServer ("cancel:" ++ messageId) requestTimestamp (not (isLocalServer fromServer)) (do

1611+ let matches = \x -> x.messageId == messageId && x.toKey == toKey

1667 cancelled <- atomic do

1612 cancelled <- atomic do

1668 msgs <- *messages

1613 msgs <- *messages

1669− let target = headOf (do m <- msgs; where m.messageId == messageId && m.toKey == toKey && m.originServer == fromServer; yield m)

1614+ case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of

1670− case target of

1671 Nothing {} -> yield False

1615 Nothing {} -> yield False

1672 Just {} -> do

1616 Just {} -> do

1673− set *messages = do m <- msgs; where not (m.messageId == messageId && m.toKey == toKey); yield m

1617+ set *messages = filter msgs (\m -> not (matches m))

1674 fwds <- *forwards

1618 fwds <- *forwards

1675− set *forwards = do f <- fwds; where not (f.messageId == messageId && f.toKey == toKey); yield f

1619+ set *forwards = filter fwds (\f -> not (matches f))

1676 cn <- *cancelNotifications

1620 cn <- *cancelNotifications

1677 set *cancelNotifications = capList (union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1621 set *cancelNotifications = capList (union cn [{toKey: toKey, messageId: messageId, cancelledAt: t}]) 10000 (\c -> c.cancelledAt)

1678 yield True

1622 yield True

@@ -1683,15 +1627,15 @@ handleFederationBatchCancel = \cancels fromServer requestTimestamp correlationId

1683 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}

1627 let batchRespond = \r -> respond {ok: r.ok, error: r.error, results: [], retryAfterMs: r.retryAfterMs}

1684 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do

1628 withFederationAuth authorization t batchRespond fromServer "cancel-batch" requestTimestamp (not (isLocalServer fromServer) && count cancels <= maxBatchSize) (do

1685 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1629 results <- mapEach (takeRelation maxBatchSize cancels) (\e -> e.messageId) (\e -> do

1630+ let matches = \x -> x.messageId == e.messageId && x.toKey == e.toKey

1686 cancelled <- atomic do

1631 cancelled <- atomic do

1687 msgs <- *messages

1632 msgs <- *messages

1688− let target = headOf (do m <- msgs; where m.messageId == e.messageId && m.toKey == e.toKey && m.originServer == fromServer; yield m)

1633+ case headOf (filter msgs (\m -> matches m && m.originServer == fromServer)) of

1689− case target of

1690 Nothing {} -> yield False

1634 Nothing {} -> yield False

1691 Just {} -> do

1635 Just {} -> do

1692− set *messages = do m <- msgs; where not (m.messageId == e.messageId && m.toKey == e.toKey); yield m

1636+ set *messages = filter msgs (\m -> not (matches m))

1693 fwds <- *forwards

1637 fwds <- *forwards

1694− set *forwards = do f <- fwds; where not (f.messageId == e.messageId && f.toKey == e.toKey); yield f

1638+ set *forwards = filter fwds (\f -> not (matches f))

1695 yield True

1639 yield True

1696 yield {messageId: e.messageId, cancelled: cancelled})

1640 yield {messageId: e.messageId, cancelled: cancelled})

1697 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))

1641 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))

@@ -1716,8 +1660,7 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation

1716 if isDup then yield True

1660 if isDup then yield True

1717 else if count msgs >= maxMessages then yield False

1661 else if count msgs >= maxMessages then yield False

1718 else

1662 else

1719− let rcCount = count (do m <- msgs; where m.toKey == e.toKey; yield m) in

1720− if rcCount >= maxMessagesPerRecipient then yield False

1663+ if countWhere msgs (\m -> m.toKey == e.toKey) >= maxMessagesPerRecipient then yield False

1721 else do

1664 else do

1722 seq <- *seqCounter

1665 seq <- *seqCounter

1723 let seqTs = nextSeqTs t seq

1666 let seqTs = nextSeqTs t seq

@@ -1750,7 +1693,7 @@ handleFederationDeliveryReceipt = \messageIds toKeys fromServer requestTimestamp

1750 set *senderDeliveries = capList (union sd receipts) 10000 (\d -> d.deliveredAt)

1693 set *senderDeliveries = capList (union sd receipts) 10000 (\d -> d.deliveredAt)

1751 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))

1694 yield (respond {ok: True {}, error: "", receivedCount: count pairs, retryAfterMs: 0}))

1752

1695

1753−-- Pair up parallel lists by index. Builds index arrays and joins on i (O(n*m) but trivial code).

1696+-- Pair up parallel lists by index (O(n*m) join on i; trivial code).

1754withIndex = \xs -> (fold (\acc x -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: x}]}) {i: 0, out: []} xs).out

1697withIndex = \xs -> (fold (\acc x -> {i: acc.i + 1, out: union acc.out [{i: acc.i, v: x}]}) {i: 0, out: []} xs).out

1755zipPairs = \mids keys ->

1698zipPairs = \mids keys ->

1756 do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}

1699 do mp <- withIndex mids; kp <- withIndex keys; where mp.i == kp.i; yield {messageId: mp.v, toKey: kp.v}

@@ -1770,13 +1713,13 @@ handleDrain = \phase authorization respond -> withMaintAuth authorization

1770 if effectivePhase == 1 then yield {drainedCount: 0, peers: [], localKeys: []}

1713 if effectivePhase == 1 then yield {drainedCount: 0, peers: [], localKeys: []}

1771 else do

1714 else do

1772 pres <- *presence

1715 pres <- *presence

1773− let localEntries = do p <- pres; where isLocalServer p.server && p.expiresAt > t; yield p

1716+ let localEntries = filter pres (\p -> isLocalServer p.server && p.expiresAt > t)

1774− set *presence = do p <- pres; where not (isLocalServer p.server) || p.expiresAt <= t; yield p

1717+ set *presence = filter pres (\p -> not (isLocalServer p.server) || p.expiresAt <= t)

1775 set *pollWakeup = t

1718 set *pollWakeup = t

1776 let peers = knownPeerServersFromPresence pres

1719 let peers = knownPeerServersFromPresence pres

1777 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}

1720 yield {drainedCount: count localEntries, peers: peers, localKeys: localEntries}

1778 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)

1721 when (effectivePhase == 2 && count drainResult.localKeys > 0 && count drainResult.peers > 0 && federationEnabled)

1779− (fork (sendBatchedGossip drainResult.peers (do p <- drainResult.localKeys; yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]}) gossipOriginHops))

1722+ (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: t, ttl: offlineGossipMinTtl, seenBy: [serverHostN]})) gossipOriginHops))

1780 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))

1723 yield (respond {ok: True {}, drained: drainResult.drainedCount, phase: effectivePhase}))

1781

1724

1782handleUndrain = \authorization respond -> withMaintAuth authorization

1725handleUndrain = \authorization respond -> withMaintAuth authorization

@@ -1860,7 +1803,7 @@ handleAdminListSessions = \pubkey authorization respond -> withMaintAuth authori

1860 (yield (respond {sessions: [], count: 0, ok: False {}}))

1803 (yield (respond {sessions: [], count: 0, ok: False {}}))

1861 (do

1804 (do

1862 ses <- *sessions

1805 ses <- *sessions

1863− let mine = do s <- ses; where s.pubkey == pubkey; yield {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt}

1806+ let mine = pluck (filter ses (\s -> s.pubkey == pubkey)) (\s -> {token: "***", expiresAt: s.expiresAt, ip: s.ip, createdAt: s.createdAt})

1864 yield (respond {sessions: mine, count: count mine, ok: True {}}))

1807 yield (respond {sessions: mine, count: count mine, ok: True {}}))

1865

1808

1866handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization

1809handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization

@@ -1868,10 +1811,9 @@ handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth autho

1868 (do

1811 (do

1869 revoked <- atomic do

1812 revoked <- atomic do

1870 ses <- *sessions

1813 ses <- *sessions

1871− let mine = do s <- ses; where s.pubkey == pubkey; yield s

1872 pruneRef sessions (\s -> s.pubkey != pubkey)

1814 pruneRef sessions (\s -> s.pubkey != pubkey)

1873 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))

1815 pruneRef presence (\p -> p.pubkey != pubkey || not (isLocalServer p.server))

1874− yield (count mine)

1816+ yield (countWhere ses (\s -> s.pubkey == pubkey))

1875 yield (respond {ok: True {}, revokedCount: revoked}))

1817 yield (respond {ok: True {}, revokedCount: revoked}))

1876

1818

1877handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization

1819handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization

@@ -1880,10 +1822,10 @@ handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth author

1880 t <- now

1822 t <- now

1881 purged <- atomic do

1823 purged <- atomic do

1882 msgs <- *messages

1824 msgs <- *messages

1883− let mine = do m <- msgs; where m.toKey == pubkey; yield m

1825+ let mine = filter msgs (\m -> m.toKey == pubkey)

1884 pruneRef messages (\m -> m.toKey != pubkey)

1826 pruneRef messages (\m -> m.toKey != pubkey)

1885 pruneRef forwards (\f -> f.toKey != pubkey)

1827 pruneRef forwards (\f -> f.toKey != pubkey)

1886− let federated = do m <- mine; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0}

1828+ let federated = pluck (filter mine (\m -> m.originServer != "")) (\m -> {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer, retries: 0})

1887 aids <- *ackedIds

1829 aids <- *ackedIds

1888 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)

1830 set *ackedIds = capList (union aids federated) maxAckedIds (\a -> a.ackedAt)

1889 yield (count mine)

1831 yield (count mine)

@@ -1895,10 +1837,9 @@ handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization

1895 else do

1837 else do

1896 result <- atomic do

1838 result <- atomic do

1897 ses <- *sessions

1839 ses <- *sessions

1898− let mine = do s <- ses; where s.ip == ip; yield s

1840+ let mine = filter ses (\s -> s.ip == ip)

1899 pruneRef sessions (\s -> s.ip != ip)

1841 pruneRef sessions (\s -> s.ip != ip)

1900− let keys = unique (do s <- mine; yield s.pubkey)

1842+ yield {revoked: count mine, keys: count (unique (pluck mine (\s -> s.pubkey)))}

1901− yield {revoked: count mine, keys: count keys}

1902 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))

1843 yield (respond {ok: True {}, revokedCount: result.revoked, affectedKeys: result.keys}))

1903

1844

1904handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization

1845handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization

@@ -1906,14 +1847,13 @@ handleAdminListAllSessions = \limit authorization respond -> withMaintAuth autho

1906 (do

1847 (do

1907 t <- now

1848 t <- now

1908 ses <- *sessions

1849 ses <- *sessions

1909− let active = do s <- ses; where s.expiresAt > t; yield s

1850+ let active = filter ses (\s -> s.expiresAt > t)

1910− let grouped = do

1851+ let grouped = pluck (unique (pluck active (\s -> s.pubkey))) (\pk ->

1911− pk <- unique (do s <- active; yield s.pubkey)

1912− let mine = do s <- active; where s.pubkey == pk; yield s

1852+ let mine = filter active (\s -> s.pubkey == pk) in

1913− yield {pubkey: pk,

1853+ {pubkey: pk,

1914− count: count mine,

1854+ count: count mine,

1915− ips: unique (do s <- mine; where s.ip != ""; yield s.ip),

1855+ ips: unique (pluck (filter mine (\s -> s.ip != "")) (\s -> s.ip)),

1916− latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine}

1856+ latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine})

1917 let sorted = sortBy (\g -> 0 - g.count) grouped

1857 let sorted = sortBy (\g -> 0 - g.count) grouped

1918 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit

1858 let effLimit = if limit <= 0 then 100 else clamp 1 500 limit

1919 yield (respond {sessions: takeRelation effLimit sorted, ok: True {}, total: count grouped}))

1859 yield (respond {sessions: takeRelation effLimit sorted, ok: True {}, total: count grouped}))

@@ -1979,10 +1919,8 @@ handleStatsPeers = \authorization respond -> withMaintAuth authorization

1979 t <- now

1919 t <- now

1980 failures <- *forwardFailures

1920 failures <- *forwardFailures

1981 fwds <- *forwards

1921 fwds <- *forwards

1982− let peers = do

1983− f <- failures

1922+ let peers = pluck failures (\f ->

1984− let pending = count (do x <- fwds; where x.toServer == f.server; yield x)

1985− yield {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: pending, avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0}

1923+ {server: f.server, failures: f.failures, lastFailedAt: f.failedAt, firstFailedAt: f.firstFailedAt, backoffMs: retryBackoff f.failures, retriable: shouldRetryServer f.server t failures, pendingForwards: countWhere fwds (\x -> x.toServer == f.server), avgLatencyMs: 0, maxLatencyMs: 0, latencySamples: 0, gossipEventsPerMin: 0})

1986 yield (respond {peers: peers, ok: True {}}))

1924 yield (respond {peers: peers, ok: True {}}))

1987

1925

1988handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization

1926handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization

@@ -1994,22 +1932,21 @@ handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization

1994 ses <- *sessions

1932 ses <- *sessions

1995 fwds <- *forwards

1933 fwds <- *forwards

1996 aps <- *activePolls

1934 aps <- *activePolls

1935+ let myFwds = filter fwds (\x -> x.toKey == pubkey)

1936+ yield (respond {

1997− let m = count (do x <- msgs; where x.toKey == pubkey; yield x)

1937+ messages: countWhere msgs (\x -> x.toKey == pubkey),

1998− let p = do x <- pres; where x.pubkey == pubkey && x.expiresAt > t; yield {server: x.server, expiresAt: x.expiresAt}

1938+ presence: pluck (filter pres (\x -> x.pubkey == pubkey && x.expiresAt > t)) (\x -> {server: x.server, expiresAt: x.expiresAt}),

1999− let s = count (do x <- ses; where x.pubkey == pubkey; yield x)

1939+ sessions: countWhere ses (\x -> x.pubkey == pubkey),

2000− let myFwds = do x <- fwds; where x.toKey == pubkey; yield x

1940+ forwards: count myFwds,

2001− let f = count myFwds

1941+ forwardDetails: pluck myFwds (\x -> {server: x.toServer, retries: x.retries, createdAt: x.createdAt}),

2002− let fd = do x <- myFwds; yield {server: x.toServer, retries: x.retries, createdAt: x.createdAt}

1942+ activePolls: countWhere aps (\x -> x.pubkey == pubkey),

2003− let a = count (do x <- aps; where x.pubkey == pubkey; yield x)

1943+ ok: True {}}))

2004− yield (respond {messages: m, presence: p, sessions: s, forwards: f, forwardDetails: fd, activePolls: a, ok: True {}}))

2005

1944

2006handleTopRecipients = \limit authorization respond -> withMaintAuth authorization

1945handleTopRecipients = \limit authorization respond -> withMaintAuth authorization

2007 (yield (respond {recipients: [], ok: False {}, total: 0}))

1946 (yield (respond {recipients: [], ok: False {}, total: 0}))

2008 (do

1947 (do

2009 msgs <- *messages

1948 msgs <- *messages

2010− let counts = do

1949+ let counts = pluck (unique (pluck msgs (\m -> m.toKey))) (\pk -> {pubkey: pk, count: countWhere msgs (\m -> m.toKey == pk)})

2011− pk <- unique (do m <- msgs; yield m.toKey)

2012− yield {pubkey: pk, count: count (do m <- msgs; where m.toKey == pk; yield m)}

2013 let sorted = sortBy (\c -> 0 - c.count) counts

1950 let sorted = sortBy (\c -> 0 - c.count) counts

2014 let effLimit = if limit <= 0 then 20 else clamp 1 100 limit

1951 let effLimit = if limit <= 0 then 20 else clamp 1 100 limit

2015 yield (respond {recipients: takeRelation effLimit sorted, ok: True {}, total: count counts}))

1952 yield (respond {recipients: takeRelation effLimit sorted, ok: True {}, total: count counts}))

@@ -2020,12 +1957,10 @@ handleStatsForwards = \authorization respond -> withMaintAuth authorization

2020 t <- now

1957 t <- now

2021 fwds <- *forwards

1958 fwds <- *forwards

2022 let servers = collectPeers fwds (\f -> f.toServer)

1959 let servers = collectPeers fwds (\f -> f.toServer)

2023− let result = do

1960+ let result = pluck servers (\s ->

2024− s <- servers

2025− let svFwds = do f <- fwds; where f.toServer == s; yield f

1961+ let svFwds = filter fwds (\f -> f.toServer == s) in

2026− let oldest = case minBy svFwds (\f -> f.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt

1962+ let oldest = case minBy svFwds (\f -> f.createdAt) of Nothing {} -> 0; Just {value: x} -> t - x.createdAt in

2027− let retries = fold (\acc f -> acc + f.retries) 0 svFwds

2028− yield {server: s, count: count svFwds, oldestAge: oldest, totalRetries: retries}

1963+ {server: s, count: count svFwds, oldestAge: oldest, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds})

2029 yield (respond {servers: result, ok: True {}, total: count fwds}))

1964 yield (respond {servers: result, ok: True {}, total: count fwds}))

2030

1965

2031handleStatsRates = \authorization respond -> withMaintAuth authorization

1966handleStatsRates = \authorization respond -> withMaintAuth authorization