git.iko.soy/skrepka
on master help sign in
Extract appendMessage and refreshLocalPresence helpers to remove duplication
4ffa00biko
server.knot+32−23
compare with master
unifiedsplit
@@ -452,10 +452,11 @@ clearFailure = \server -> atomic (do
452 *forwardFailures = filter (\f -> f.server != server) forwardFailures)
452 *forwardFailures = filter (\f -> f.server != server) forwardFailures)
453
453
454-- ============================================
454-- ============================================
455−-- Common utility queries
455+-- Common helpers
456-- ============================================
456-- ============================================
457−-- Pure helpers shared by send/poll/federation paths: monotonic sequence
457+-- Helpers shared by send/poll/federation paths: monotonic sequence
458−-- generation and presence lookup.
458+-- generation, presence queries, and atomic mailbox/presence writes
459+-- (callers supply the surrounding `atomic` block).
459
460
460-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
461-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
461-- Guarantees /poll cursors order events even when many arrive in the same ms.
462-- Guarantees /poll cursors order events even when many arrive in the same ms.
@@ -481,6 +482,28 @@ collectPeers = \items getServer -> filter isRemote (map getServer items)
481knownPeerServersFromPresence : [PresenceEntry] -> [Text]
482knownPeerServersFromPresence : [PresenceEntry] -> [Text]
482knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
483knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
483
484
485+-- Append one ciphertext to the local mailbox under a fresh monotonic
486+-- sequence and bump pollWakeup so blocked /poll holders unblock. Caller
487+-- wraps in `atomic`.
488+appendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, rw *seqCounter} {}
489+appendMessage = \toKey encryptedBlob t -> do
490+ msgs <- *messages
491+ seq <- *seqCounter
492+ let seqTs = nextSeqTs t seq
493+ *seqCounter = seqTs
494+ *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
495+ *pollWakeup = seqTs
496+
497+-- Replace any local presence rows for `pubkey` with one fresh row at
498+-- `expiry`, and return the post-update peer broadcast set. Caller wraps in
499+-- `atomic`.
500+refreshLocalPresence : [PresenceEntry] -> PubkeyHex -> Timestamp -> IO {rw *presence} [Text]
501+refreshLocalPresence = \pres pubkey expiry -> do
502+ let kept = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
503+ let newPres = union kept [{pubkey, server: serverHost, expiresAt: expiry}]
504+ *presence = newPres
505+ yield (knownPeerServersFromPresence newPres)
506+
484-- ============================================
507-- ============================================
485-- Pruning
508-- Pruning
486-- ============================================
509-- ============================================
@@ -638,10 +661,8 @@ handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp -> do
638 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
661 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp}
639 *sessions = capList (union pruned [newSession]) maxSessions
662 *sessions = capList (union pruned [newSession]) maxSessions
640 pres <- *presence
663 pres <- *presence
641− let prunedPres = filter (\p -> not (p.pubkey == pubkey && isLocalServer p.server)) pres
664+ peers <- refreshLocalPresence pres pubkey expiry
642− let newPres = union prunedPres [{pubkey, server: serverHost, expiresAt: expiry}]
643− *presence = newPres
644− yield (Just {value: knownPeerServersFromPresence newPres}))
665+ yield (Just {value: peers}))
645 case result of
666 case result of
646 Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
667 Nothing {} -> yield (Err {error: {status: 401, message: "invalid"}})
647 Just {value: peers} -> do
668 Just {value: peers} -> do
@@ -672,10 +693,8 @@ handlePoll = \cursor authorization clientIp -> do
672 let stale = not any (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2) pres
693 let stale = not any (\p -> mine p && p.expiresAt - t >= sessionExpiry / 2) pres
673 refreshPeers <- if stale
694 refreshPeers <- if stale
674 then do
695 then do
675− let kept = filter (\p -> not mine p) pres
696+ peers <- refreshLocalPresence pres pubkey (t + sessionExpiry)
676− let newPres = union kept [{pubkey, server: serverHost, expiresAt: t + sessionExpiry}]
677− *presence = newPres
678− yield (Just {value: knownPeerServersFromPresence newPres})
697+ yield (Just {value: peers})
679 else yield (Nothing {})
698 else yield (Nothing {})
680 yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
699 yield (Just {value: {pubkey, clampedCursor, startWakeup, refreshPeers}}))
681 case setup of
700 case setup of
@@ -769,12 +788,7 @@ handleSendMessages = \messages authorization clientIp -> do
769-- via `resolveSendTarget`.
788-- via `resolveSendTarget`.
770commitSendMessage : PubkeyHex -> BlobHex -> [Text] -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter} {}
789commitSendMessage : PubkeyHex -> BlobHex -> [Text] -> Int<Ms> -> IO {rw *forwards, rw *messages, rw *pollWakeup, rw *seqCounter} {}
771commitSendMessage = \to encryptedBlob remoteServers t -> do
790commitSendMessage = \to encryptedBlob remoteServers t -> do
772− msgs <- *messages
791+ appendMessage to encryptedBlob t
773− seq <- *seqCounter
774− let seqTs = nextSeqTs t seq
775− *seqCounter = seqTs
776− *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
777− *pollWakeup = seqTs
778 when federationEnabled (do
792 when federationEnabled (do
779 fwds <- *forwards
793 fwds <- *forwards
780 let newFwds = map (\s -> {toServer: s, toKey: to, encryptedBlob, retries: 0}) (filter (\s -> not any (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s) fwds) remoteServers)
794 let newFwds = map (\s -> {toServer: s, toKey: to, encryptedBlob, retries: 0}) (filter (\s -> not any (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s) fwds) remoteServers)
@@ -860,15 +874,10 @@ handleRecvGossip = \events fromServer -> do
860storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter} (Result HttpError {})
874storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> IO {rw *messages, rw *pollWakeup, r *presence, rw *seqCounter} (Result HttpError {})
861storeFederated = \toKey encryptedBlob t -> do
875storeFederated = \toKey encryptedBlob t -> do
862 pres <- *presence
876 pres <- *presence
863− msgs <- *messages
864 if not (hasLocalPresence pres toKey t)
877 if not (hasLocalPresence pres toKey t)
865 then yield (Err {error: {status: 404, message: "no_presence"}})
878 then yield (Err {error: {status: 404, message: "no_presence"}})
866 else do
879 else do
867− seq <- *seqCounter
868− let seqTs = nextSeqTs t seq
869− *seqCounter = seqTs
870− *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
880+ appendMessage toKey encryptedBlob t
871− *pollWakeup = seqTs
872 yield (Ok {value: {}})
881 yield (Ok {value: {}})
873
882
874-- POST /federation/forward: receive a single federated message destined for
883-- POST /federation/forward: receive a single federated message destined for