@@ -46,9 +46,8 @@ federationScheme = "https://"
4646-- The Ed25519 federation keypair lives in `*federationPrivateKeyHex` /
4747-- `*federationPublicKeyHex` refs and is generated on first startup; see `main`.
4848
49−-- Storage retention for messages (and clamp on client-supplied wall-clock
50−-- timestamps + dedup-id retention). Stored messages live this long; sends
51−-- whose `timestamp` is older than this are rejected (30 days).
49+-- Storage retention for messages and dedup-id records. Mailbox rows live this
50+-- long after `receivedAt`; expired rows are pruned by `pruneExpired` (30 days).
5251messageMaxAge : Int<Ms>
5352messageMaxAge = 2592000000
5453-- Upper bound on TTL a client may request for a presence entry (1 hour).
@@ -240,10 +239,7 @@ sToMs = \s -> stripUnit s * 1000
240239
241240-- One event in a /poll response. The payload is the opaque cipher blob plus
242241-- delivery metadata for the recipient; the server cannot read the blob.
243−type PollEvent = {
244− id: MessageIdHex,
245− payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}
246−}
242+type PollEvent = {id: MessageIdHex, payload: {encryptedBlob: BlobHex, toKey: PubkeyHex}}
247243
248244-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
249245-- the parser's fallback for forward-compatibility with future event types.
@@ -337,13 +333,10 @@ type Session = {
337333-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
338334-- Identity is `messageIdOf encryptedBlob` (BLAKE3 of the ciphertext); we don't
339335-- persist it because it's deterministically derivable. Storage TTL is fixed
340−-- at `messageMaxAge`, so expiry is derived from `receivedAt`.
341−type Message = {
342− toKey: PubkeyHex,
343− encryptedBlob: BlobHex,
344− timestamp: Timestamp,
345− receivedAt: Timestamp
346−}
336+-- at `messageMaxAge`, so expiry is derived from `receivedAt`. Any send time
337+-- the sender wants to convey lives inside `encryptedBlob` (the server cannot
338+-- read it).
339+type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}
347340-- Last-known location of a pubkey, either local (`server == serverHostN`) or
348341-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
349342type PresenceEntry = {
@@ -360,7 +353,6 @@ type ForwardEntry = {
360353 createdAt: Timestamp,
361354 toKey: PubkeyHex,
362355 encryptedBlob: BlobHex,
363− timestamp: Timestamp,
364356 retries: Int
365357}
366358-- Replay-protection record: a message id we already delivered locally.
@@ -374,7 +366,7 @@ type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
374366type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
375367
376368-- One entry of a /messages/batch request body.
377−type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
369+type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex}
378370-- Per-entry result returned by /messages/batch.
379371type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
380372
@@ -452,19 +444,19 @@ route AuthApi where
452444
453445route ClientApi where
454446 POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int} = Poll
455− POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
447+ POST {to: PubkeyHex, encryptedBlob: BlobHex} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
456448 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
457449 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
458450 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch
459− POST {offset: Int} /messages/list headers {authorization: Text, clientIp: Text} -> {messages: [{messageId: MessageIdHex, timestamp: Timestamp, receivedAt: Timestamp}], authorized: Bool, count: Int, error: Text} = ListMessages
451+ POST {offset: Int} /messages/list headers {authorization: Text, clientIp: Text} -> {messages: [{messageId: MessageIdHex, receivedAt: Timestamp}], authorized: Bool, count: Int, error: Text} = ListMessages
460452 POST {messageIds: [MessageIdHex]} /messages/delete headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, deletedCount: Int, error: Text} = DeleteMessages
461453 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text} -> {server: AnyServerName, authorized: Bool, error: Text} = LookupPresence
462454 POST {pubkeys: [PubkeyHex]} /lookup/batch headers {authorization: Text, clientIp: Text} -> {results: [{pubkey: PubkeyHex, server: AnyServerName}], authorized: Bool, error: Text} = BatchLookupPresence
463455
464456route FederationApi where
465457 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/gossip headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = RecvGossip
466− POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
467− POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: MessageIdHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
458+ POST {toKey: PubkeyHex, encryptedBlob: BlobHex, messageId: MessageIdHex, fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = ForwardMessage
459+ POST {forwards: [{toKey: PubkeyHex, encryptedBlob: BlobHex, messageId: MessageIdHex}], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/forward-batch headers {authorization: Text} -> {ok: Bool, error: Text, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} = FederationBatchForward
468460
469461route FederationPubkeyApi where
470462 GET /federation/pubkey -> {pubkey: Text, ok: Bool} = FederationPubkey
@@ -597,7 +589,7 @@ pct : Int -> Int -> Int
597589pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
598590
599591-- Common ack/delete tail: pull matching messages, drop them and their forwards.
600−removeForKey : (forall r. {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
592+removeForKey : (forall r. {toKey: PubkeyHex, encryptedBlob: BlobHex | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
601593removeForKey = \pred -> do
602594 msgs <- *messages
603595 fwds <- *forwards
@@ -998,19 +990,14 @@ clearFailure = \server -> atomic (do
998990-- ============================================
999991-- Common utility queries
1000992-- ============================================
1001−-- Pure helpers shared by send/poll/federation paths: TTL/timestamp clamping,
1002−-- monotonic sequence generation, presence lookup, duplicate detection across
993+-- Pure helpers shared by send/poll/federation paths: TTL clamping, monotonic
994+-- sequence generation, presence lookup, duplicate detection across
1003995-- (messages, deliveredIds), and last-write-wins presence merge.
1004996
1005997-- Clamp a client-supplied TTL to (0, maxPresenceTtl].
1006998clampTtl : Int<Ms> -> Int<Ms>
1007999clampTtl = clamp 1 maxPresenceTtl
10081000
1009−-- Clamp a client-supplied message timestamp into [now - messageMaxAge, now].
1010−-- Stops senders from claiming arbitrary future or far-past timestamps.
1011−clampTimestamp : Timestamp -> Timestamp -> Timestamp
1012−clampTimestamp = \timestamp t -> clamp (t - messageMaxAge) t timestamp
1013−
10141001-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
10151002-- Guarantees /poll cursors order events even when many arrive in the same ms.
10161003nextSeqTs : Timestamp -> Timestamp -> Timestamp
@@ -1155,7 +1142,6 @@ retryPendingForwards = do
11551142 result <- fetch (federationUrl server) (ForwardMessage {
11561143 f.toKey,
11571144 f.encryptedBlob,
1158− f.timestamp,
11591145 messageId,
11601146 fromServer: serverHostN,
11611147 requestTimestamp: t2,
@@ -1174,7 +1160,7 @@ retryPendingForwards = do
11741160 atomic (do
11751161 cur <- *forwards
11761162 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1177− then {x.toServer, x.createdAt, x.toKey, x.encryptedBlob, x.timestamp, retries: x.retries + 1}
1163+ then {x.toServer, x.createdAt, x.toKey, x.encryptedBlob, retries: x.retries + 1}
11781164 else x))))
11791165
11801166-- Forward queued messages to a server when a remote key comes online.
@@ -1188,7 +1174,6 @@ forwardQueuedToServer = \pubkey server t -> atomic (do
11881174 createdAt: t,
11891175 m.toKey,
11901176 m.encryptedBlob,
1191− m.timestamp,
11921177 retries: 0
11931178 })
11941179 *forwards = union fwds newFwds)
@@ -1350,7 +1335,7 @@ handleRevoke = \targetToken authorization clientIp respond -> do
13501335
13511336-- POST /auth/logout: drop the caller's own session. If this was the last
13521337-- session for the pubkey, also clear local presence and gossip offline.
1353−handleLogout : Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {r *sessions, clock} a
1338+handleLogout : Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {rw *presence, rw *sessions, clock} a
13541339handleLogout = \authorization clientIp respond -> do
13551340 t <- now
13561341 let myToken = case extractToken authorization of
@@ -1390,7 +1375,7 @@ handleLogout = \authorization clientIp respond -> do
13901375-- POST /poll: top-level dispatch. Validates drain/global gates, detects a
13911376-- generation mismatch (server restart) to force a cursor reset, then hands
13921377-- off to `pollOnce` for the actual long-poll body.
1393−handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {rw *activePolls, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *pollWakeup, r *serverGeneration, r *sessions, clock} Response
1378+handlePoll : Int -> Int -> Text -> Text -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> IO {r *drainPhase, r *pollWakeup, r *serverGeneration, r *sessions, rw *activePolls, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} Response
13941379handlePoll = \cursor generation authorization clientIp respond -> do
13951380 t <- now
13961381 serverGen <- *serverGeneration
@@ -1407,7 +1392,7 @@ handlePoll = \cursor generation authorization clientIp respond -> do
14071392-- Body of a single poll. Treats the cursor advance as an implicit ack
14081393-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
14091394-- collect events (or block until wakeup), then assembles the response.
1410−pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {rw *activePolls, r *pollWakeup, r *sessions, clock} Response
1395+pollOnce : Text -> Int -> Bool -> Text -> Int -> ({authorized: Bool, cursor: Timestamp, cursorReset: Bool, cursorStale: Bool, draining: Bool, events: [PollEvent], generation: Int, hasMore: Bool, heartbeat: Bool, loadPct: Int, rateLimited: Bool, retryAfterMs: Int} -> Response) -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *activePolls, rw *forwards, rw *messages, clock} Response
14111396pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
14121397 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
14131398 let clampedCursor = clamp 0 t cursor
@@ -1446,7 +1431,7 @@ pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t ->
14461431
14471432-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
14481433-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1449−doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}], hasMore: Bool}
1434+doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{id: MessageIdHex, payload: {encryptedBlob: BlobHex, toKey: PubkeyHex}}], hasMore: Bool}
14501435doPoll = \s t -> do
14511436 let collectAt = \tnow -> atomic (do
14521437 msgs <- *messages
@@ -1462,7 +1447,7 @@ doPoll = \s t -> do
14621447-- Pull this pubkey's unexpired messages newer than `cursor`, sort by
14631448-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.
14641449collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
1465−collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: messageIdOf m.encryptedBlob, payload: {m.toKey, m.encryptedBlob, m.timestamp}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))
1450+collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {id: messageIdOf m.encryptedBlob, payload: {m.toKey, m.encryptedBlob}}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))
14661451
14671452-- Treat the supplied cursor as an ack: drop all messages with
14681453-- receivedAt <= cursor, plus their pending forwards.
@@ -1489,10 +1474,10 @@ ackPolledMessages = \pubkey cursor -> do
14891474-- Federated forwards are kicked off in a background fork after the atomic.
14901475
14911476-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1492−-- limit/stale timestamp; otherwise calls `sendMessageCore` inside an atomic
1493−-- block and forks the federation forward (if any) once the local commit lands.
1494−handleSendMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> ({error: Text, loadPct: Int, messageId: MessageIdHex, retryAfterMs: Int, status: MessageStatus} -> a) -> IO {r *deliveredIds, r *drainPhase, r *presence, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *pairRates, rw *pollWakeup, rw *recvRates, rw *sendRates, rw *seqCounter, clock} a
1495−handleSendMessage = \to encryptedBlob timestamp authorization clientIp respond -> do
1477+-- limit; otherwise calls `sendMessageCore` inside an atomic block and forks
1478+-- the federation forward (if any) once the local commit lands.
1479+handleSendMessage : PubkeyHex -> BlobHex -> Text -> Text -> ({error: Text, loadPct: Int, messageId: MessageIdHex, retryAfterMs: Int, status: MessageStatus} -> a) -> IO {r *deliveredIds, r *drainPhase, r *presence, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, rw *pairRates, rw *pollWakeup, rw *recvRates, rw *sendRates, rw *seqCounter, clock} a
1480+handleSendMessage = \to encryptedBlob authorization clientIp respond -> do
14961481 t <- now
14971482 dp <- *drainPhase
14981483 g <- checkGlobalRate t
@@ -1502,13 +1487,10 @@ handleSendMessage = \to encryptedBlob timestamp authorization clientIp respond -
15021487 then err (MsDraining {}) "draining"
15031488 else if g
15041489 then err (MsRateLimited {}) "rate_limited"
1505− else if t - timestamp > messageMaxAge
1506− then err (MsRejected {}) "stale_timestamp"
15071490 else do
1508− let clampedTs = clampTimestamp timestamp t
15091491 result <- atomic (do
1510− withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob clampedTs t))
1511− when federationEnabled (fork (forkForwardsTo to encryptedBlob clampedTs result.remoteServers))
1492+ withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob t))
1493+ when federationEnabled (fork (forkForwardsTo to encryptedBlob result.remoteServers))
15121494 let ok = isSendSuccess result.status
15131495 yield (respond {
15141496 result.status,
@@ -1524,8 +1506,8 @@ handleSendMessage = \to encryptedBlob timestamp authorization clientIp respond -
15241506-- `seqCounter`, appends the message, and pre-stages forwards for any remote
15251507-- presence rows that already exist for `to`. Returns the chosen status and
15261508-- the list of remote servers the caller should kick forwards to.
1527−sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
1528−sendMessageCore = \senderKey to encryptedBlob clampedTs t -> do
1509+sendMessageCore : PubkeyHex -> PubkeyHex -> BlobHex -> Int<Ms> -> IO {r *deliveredIds, rw *forwards, rw *messages, rw *pairRates, rw *pollWakeup, r *presence, rw *recvRates, rw *sendRates, rw *seqCounter} {remoteServers: [Text], status: MessageStatus}
1510+sendMessageCore = \senderKey to encryptedBlob t -> do
15291511 msgs <- *messages
15301512 dids <- *deliveredIds
15311513 sRates <- *sendRates
@@ -1552,7 +1534,7 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs t -> do
15521534 seq <- *seqCounter
15531535 let seqTs = nextSeqTs t seq
15541536 *seqCounter = seqTs
1555− *messages = union msgs [{toKey: to, encryptedBlob, timestamp: clampedTs, receivedAt: seqTs}]
1537+ *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
15561538 *pollWakeup = seqTs
15571539 let toPresence = presenceForKey pres to t
15581540 let remoteServers = collectPeers toPresence (\p -> p.server)
@@ -1564,7 +1546,6 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs t -> do
15641546 createdAt: t,
15651547 toKey: to,
15661548 encryptedBlob,
1567− timestamp: clampedTs,
15681549 retries: 0
15691550 })
15701551 *forwards = union fwds newFwds
@@ -1583,8 +1564,8 @@ sendMessageCore = \senderKey to encryptedBlob clampedTs t -> do
15831564-- `*messages` row is dropped too: the origin's job is done and the message
15841565-- now lives on the receiving peer's mailbox. On failure the forwards row
15851566-- stays so `retryPendingForwards` can re-attempt with backoff.
1586−forkForwardsTo : PubkeyHex -> BlobHex -> Int<Ms> -> [Text] -> IO {} {}
1587−forkForwardsTo = \to encryptedBlob clampedTs servers -> forEach servers (\server -> do
1567+forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {} {}
1568+forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
15881569 let effMsgId = messageIdOf encryptedBlob
15891570 fwds <- *forwards
15901571 let same = \x -> x.encryptedBlob == encryptedBlob && x.toServer == server
@@ -1596,7 +1577,6 @@ forkForwardsTo = \to encryptedBlob clampedTs servers -> forEach servers (\server
15961577 result <- fetch (federationUrl server) (ForwardMessage {
15971578 toKey: to,
15981579 encryptedBlob,
1599− timestamp: clampedTs,
16001580 messageId: effMsgId,
16011581 fromServer: serverHostN,
16021582 requestTimestamp: t,
@@ -1703,7 +1683,7 @@ handleAckMessages = \messageIds authorization clientIp respond -> do
17031683
17041684-- POST /messages/list: paginated peek at the caller's mailbox without acking.
17051685-- Returns metadata only (no ciphertext) plus the total count.
1706−handleListMessages : Int -> Text -> Text -> ({authorized: Bool, count: Int, error: Text, messages: [{messageId: MessageIdHex, receivedAt: Int<Ms>, timestamp: Int<Ms>}]} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1686+handleListMessages : Int -> Text -> Text -> ({authorized: Bool, count: Int, error: Text, messages: [{messageId: MessageIdHex, receivedAt: Int<Ms>}]} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
17071687handleListMessages = \offset authorization clientIp respond -> do
17081688 t <- now
17091689 let mkErr = \e -> {
@@ -1726,7 +1706,6 @@ handleListMessages = \offset authorization clientIp respond -> do
17261706 let total = count myMsgs
17271707 let page = pluck (takeRelation maxPollEvents (dropRelation (clamp 0 total offset) myMsgs)) (\m -> {
17281708 messageId: messageIdOf m.encryptedBlob,
1729− m.timestamp,
17301709 m.receivedAt
17311710 })
17321711 yield (respond {messages: page, authorized: True {}, count: total, error: ""}))
@@ -1756,18 +1735,15 @@ handleDeleteMessages = \messageIds authorization clientIp respond -> do
17561735-- POST /messages/batch: multi-recipient send in one call. Each entry runs
17571736-- through `sendMessageCore` separately so per-entry rate limits/errors don't
17581737-- abort the rest; federated forwards are forked per entry.
1759−handleSendBatch : [{encryptedBlob: BlobHex, timestamp: Int<Ms>, to: PubkeyHex | a}] -> Text -> Text -> ({authorized: Bool, results: [BatchResult], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
1738+handleSendBatch : [{encryptedBlob: BlobHex, to: PubkeyHex | a}] -> Text -> Text -> ({authorized: Bool, results: [BatchResult], retryAfterMs: Int} -> b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} b
17601739handleSendBatch = \messages authorization clientIp respond -> do
17611740 t <- now
17621741 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
17631742 gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
17641743 results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
17651744 let effMsgId = messageIdOf m.encryptedBlob
1766− let clampedTs = clampTimestamp m.timestamp t
1767− r <- if t - m.timestamp > messageMaxAge
1768− then yield {status: MsRejected {}, remoteServers: []}
1769− else atomic (do sendMessageCore senderKey m.to m.encryptedBlob clampedTs t)
1770− when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob clampedTs r.remoteServers))
1745+ r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1746+ when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
17711747 yield {messageId: effMsgId, r.status})
17721748 yield (respond {results, authorized: True {}, retryAfterMs: 0}))
17731749
@@ -1834,8 +1810,8 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp authorizati
18341810-- on stale gossip about R could land on B with no actively-polling client and
18351811-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
18361812-- next presence target.
1837−storeFederated : PubkeyHex -> BlobHex -> Int<Ms> -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1838−storeFederated = \toKey encryptedBlob timestamp checkRecvRate t -> do
1813+storeFederated : PubkeyHex -> BlobHex -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1814+storeFederated = \toKey encryptedBlob checkRecvRate t -> do
18391815 pres <- *presence
18401816 msgs <- *messages
18411817 dids <- *deliveredIds
@@ -1857,9 +1833,7 @@ storeFederated = \toKey encryptedBlob timestamp checkRecvRate t -> do
18571833 seq <- *seqCounter
18581834 let seqTs = nextSeqTs t seq
18591835 *seqCounter = seqTs
1860− *messages = union msgs [
1861− {toKey, encryptedBlob, timestamp: clampTimestamp timestamp t, receivedAt: seqTs}
1862− ]
1836+ *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
18631837 *pollWakeup = seqTs
18641838 yield ""
18651839
@@ -1870,21 +1844,18 @@ storeFederated = \toKey encryptedBlob timestamp checkRecvRate t -> do
18701844-- the blob ourselves so a hostile peer cannot bypass dedup by varying the
18711845-- id. Honest senders derive the same value, so the "forward:" ++ messageId
18721846-- auth scope still verifies.
1873−handleForwardMessage : PubkeyHex -> BlobHex -> Int<Ms> -> Text -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
1874−handleForwardMessage = \toKey encryptedBlob timestamp _ fromServer requestTimestamp authorization respond -> do
1847+handleForwardMessage : PubkeyHex -> BlobHex -> Text -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> a) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter, clock, network} a
1848+handleForwardMessage = \toKey encryptedBlob _ fromServer requestTimestamp authorization respond -> do
18751849 let messageId = messageIdOf encryptedBlob
18761850 t <- now
18771851 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote fromServer) (do
1878− if t - timestamp > messageMaxAge
1879− then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
1880− else do
1881− result <- atomic storeFederated toKey encryptedBlob timestamp (True {}) t
1882− yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
1852+ result <- atomic storeFederated toKey encryptedBlob (True {}) t
1853+ yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
18831854
18841855-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
18851856-- Skips per-entry recv-rate charging (the per-server federation rate limit
18861857-- already gates the call as a whole).
1887−handleFederationBatchForward : [{encryptedBlob: BlobHex, messageId: MessageIdHex, timestamp: Int<Ms>, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
1858+handleFederationBatchForward : [{encryptedBlob: BlobHex, messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, results: [{messageId: MessageIdHex, ok: Bool}], retryAfterMs: Int} -> b) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, clock, network} b
18881859handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
18891860 t <- now
18901861 let batchRespond = \r -> respond {r.ok, r.error, results: [], r.retryAfterMs}
@@ -1892,12 +1863,8 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp authorizati
18921863 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> do
18931864 -- Recompute id from blob; ignore e.messageId (untrusted, peer-chosen).
18941865 let messageId = messageIdOf e.encryptedBlob
1895− ok <- if t - e.timestamp > messageMaxAge
1896− then yield (False {})
1897− else do
1898− result <- atomic storeFederated e.toKey e.encryptedBlob e.timestamp (False {}) t
1899− yield (result == "")
1900− yield {messageId, ok})
1866+ result <- atomic storeFederated e.toKey e.encryptedBlob (False {}) t
1867+ yield {messageId, ok: result == ""})
19011868 yield (respond {ok: True {}, error: "", results, retryAfterMs: 0}))
19021869
19031870-- ============================================
@@ -2595,7 +2562,7 @@ serve = \req -> case req of
25952562 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
25962563 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
25972564 Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond
2598− SendMessage {to, encryptedBlob, timestamp, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob timestamp authorization clientIp respond
2565+ SendMessage {to, encryptedBlob, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob authorization clientIp respond
25992566 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
26002567 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
26012568 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
@@ -2604,7 +2571,7 @@ serve = \req -> case req of
26042571 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
26052572 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
26062573 RecvGossip {events, hopsRemaining, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events hopsRemaining fromServer requestTimestamp authorization respond
2607− ForwardMessage {toKey, encryptedBlob, timestamp, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob timestamp messageId fromServer requestTimestamp authorization respond
2574+ ForwardMessage {toKey, encryptedBlob, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob messageId fromServer requestTimestamp authorization respond
26082575 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
26092576 FederationPubkey {respond} -> handleFederationPubkey respond
26102577 Prune {authorization, respond} -> handlePrune authorization respond