@@ -1,15 +1,24 @@
1-- Skrepka Relay Server
1-- Skrepka Relay Server
2-- Single-server implementation of the Skrepka protocol (v0.1)
2-- Single-server implementation of the Skrepka protocol (v0.1)
3-- Binary data (keys, signatures, blobs) uses hex encoding (bytesToHex/bytesFromHex)
3-- Binary data (keys, signatures, blobs) uses hex encoding (bytesToHex/bytesFromHex)
6+-- Clients authenticate via Ed25519 challenge-response, then poll for messages via long-polling.
7+-- Messages are E2E encrypted blobs routed by recipient pubkey.
8+-- Federation: servers gossip presence info and forward messages for keys homed elsewhere.
9+-- Presence tracks which server a pubkey was last seen on (online/offline via gossip).
10+-- All state is in-memory; background tasks handle pruning and retry of failed federation ops.
5-- ============================================
12-- ============================================
6-- Configuration
13-- Configuration
7-- ============================================
14-- ============================================
16+-- Server identity and federation credentials
9serverHost = "localhost:8080"
17serverHost = "localhost:8080"
10federationSecret = "" -- must be set to enable federation
18federationSecret = "" -- must be set to enable federation
11maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
19maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
12federationScheme = "https://" -- scheme for federation HTTP calls
20federationScheme = "https://" -- scheme for federation HTTP calls
21+-- Message and data size limits
13messageMaxAge : Int<Ms>
22messageMaxAge : Int<Ms>
14messageMaxAge = 2592000000 -- 30 days
23messageMaxAge = 2592000000 -- 30 days
15maxPresenceTtl : Int<S>
24maxPresenceTtl : Int<S>
@@ -22,6 +31,7 @@ maxGossipEvents = 100
22maxPollEvents = 50
31maxPollEvents = 50
23maxPresenceEntries = 10000
32maxPresenceEntries = 10000
24maxMessagesPerRecipient = 500
33maxMessagesPerRecipient = 500
34+-- Global capacity limits for in-memory collections
25clockSkewTolerance : Int<Ms>
35clockSkewTolerance : Int<Ms>
26clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
36clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
27maxForwardsEntries = 10000
37maxForwardsEntries = 10000
@@ -31,40 +41,44 @@ maxTokenLen = 256
31maxChallenges = 10000
41maxChallenges = 10000
32maxSessions = 10000
42maxSessions = 10000
33maxAckedIds = 10000
43maxAckedIds = 10000
34−maxAckOverflow = 200
44+maxAckOverflow = 200 -- overflow buffer for acks that couldn't be sent yet
35maxDeliveredIds = 10000
45maxDeliveredIds = 10000
46+-- Per-key limits and rate limiting
36maxChallengesPerKey = 5
47maxChallengesPerKey = 5
37maxSessionsPerKey = 10
48maxSessionsPerKey = 10
38challengeExpiry : Int<Ms>
49challengeExpiry : Int<Ms>
39challengeExpiry = 60000 -- 1 minute
50challengeExpiry = 60000 -- 1 minute
40−maxChallengeRatePerKey = 5
51+maxChallengeRatePerKey = 5 -- max challenge requests per key per window
41maxChallengeRateEntries = 1000
52maxChallengeRateEntries = 1000
42challengeRateWindowMs : Int<Ms>
53challengeRateWindowMs : Int<Ms>
43challengeRateWindowMs = 1000
54challengeRateWindowMs = 1000
44−maxSendRatePerKey = 20
55+maxSendRatePerKey = 20 -- max messages sent per key per window
45sendRateWindowMs : Int<Ms>
56sendRateWindowMs : Int<Ms>
46sendRateWindowMs = 1000
57sendRateWindowMs = 1000
47maxSendRateEntries = 1000
58maxSendRateEntries = 1000
59+-- Session and gossip timing
48sessionExpiry : Int<Ms>
60sessionExpiry : Int<Ms>
49sessionExpiry = 3600000 -- 1 hour
61sessionExpiry = 3600000 -- 1 hour
50onlineGossipTtl : Int<S>
62onlineGossipTtl : Int<S>
51−onlineGossipTtl = 3600 -- 1 hour
63+onlineGossipTtl = 3600 -- 1 hour; TTL announced when a key authenticates
52offlineGossipMinTtl : Int<S>
64offlineGossipMinTtl : Int<S>
53offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
65offlineGossipMinTtl = 60 -- minimum TTL for offline gossip events
54maxForwardRetriesPerServer = 100
66maxForwardRetriesPerServer = 100
55−maxGossipHops = 5
67+maxGossipHops = 5 -- max hops a gossip event can propagate across servers
56−gossipOriginHops = 2
68+gossipOriginHops = 2 -- hops set when this server originates a gossip event
69+-- Exponential backoff for failed federation forwards
57retryBackoffBase : Int<Ms>
70retryBackoffBase : Int<Ms>
58retryBackoffBase = 30000 -- 30 seconds
71retryBackoffBase = 30000 -- 30 seconds
59retryBackoffMax : Int<Ms>
72retryBackoffMax : Int<Ms>
60retryBackoffMax = 480000 -- 8 minutes
73retryBackoffMax = 480000 -- 8 minutes
61−maxForwardFailures = 1000
74+maxForwardFailures = 1000 -- max tracked server failure records
76+-- Background task intervals
63backgroundPruneInterval : Int<Ms>
77backgroundPruneInterval : Int<Ms>
64−backgroundPruneInterval = 30000 -- 30 seconds
78+backgroundPruneInterval = 30000 -- 30 seconds; also retries failed forwards/acks
65pollTimeoutMs : Int<Ms>
79pollTimeoutMs : Int<Ms>
66−pollTimeoutMs = 25000 -- long-poll heartbeat interval
80+pollTimeoutMs = 25000 -- long-poll heartbeat interval (how long a poll blocks before returning empty)
67−maxPollsPerKey = 3
81+maxPollsPerKey = 3 -- max concurrent long-poll connections per pubkey
68maxActivePolls = 5000
82maxActivePolls = 5000
69presenceRefreshInterval : Int<Ms>
83presenceRefreshInterval : Int<Ms>
70presenceRefreshInterval = 300000 -- 5 minutes
84presenceRefreshInterval = 300000 -- 5 minutes
@@ -78,8 +92,10 @@ federationUrl = \server -> federationScheme ++ server
79-- ============================================
93-- ============================================
95+-- Event delivered to a polling client: an encrypted message blob
81type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}
96type PollEvent = {id: MessageIdHex, eventType: Text, payload: {encryptedBlob: BlobHex, timestamp: Timestamp, toKey: PubkeyHex}}
98+-- Gossip events propagate presence changes between federated servers
83data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
99data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
85impl ToJSON GossipEventType where
101impl ToJSON GossipEventType where
@@ -95,10 +111,15 @@ impl FromJSON GossipEventType where
95 else if s == "offline" then GossipOffline {}
111 else if s == "offline" then GossipOffline {}
96 else GossipUnknown {}
112 else GossipUnknown {}
114+-- Wire format for gossip events sent between servers
98type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>}
115type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex, server: ServerName, timestamp: Timestamp, ttl: Int<S>}
117+-- A gossip event that passed validation (timestamp not in future, TTL > 0, not expired)
100type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
118type ValidGossipEntry = {pubkey: PubkeyHex, server: ServerName, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}
120+-- Status returned to the sender after SendMessage
121+-- Delivered: recipient is online locally; Federated: forwarded to remote server;
122+-- Queued: stored but recipient location unknown; Unauthorized/Rejected: not accepted
102data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {}
123data MessageStatus = MsDelivered {} | MsFederated {} | MsQueued {} | MsUnauthorized {} | MsRejected {}
104impl ToJSON MessageStatus where
125impl ToJSON MessageStatus where
@@ -109,8 +130,10 @@ impl ToJSON MessageStatus where
109 MsUnauthorized {} -> "unauthorized"
130 MsUnauthorized {} -> "unauthorized"
110 MsRejected {} -> "rejected")
131 MsRejected {} -> "rejected")
133+-- Internal delivery status (same as MessageStatus but without error variants)
112data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
134data DeliveryStatus = DsDelivered {} | DsFederated {} | DsQueued {}
136+-- Validated hex-encoded types with maximum length constraints
114isValidBoundedHex : Int -> Text -> Bool
137isValidBoundedHex : Int -> Text -> Bool
115isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
138isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
116type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
139type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
@@ -120,6 +143,8 @@ isValidMessageId : Text -> Bool
120isValidMessageId = isValidBoundedHex maxMessageIdLen
143isValidMessageId = isValidBoundedHex maxMessageIdLen
121type MessageIdHex = Text where isValidMessageId
144type MessageIdHex = Text where isValidMessageId
122type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
145type OptionalMessageId = Text where \s -> s == "" || isValidMessageId s
146+-- ServerName: validated server hostname (rejects private IPs, localhost, etc.)
147+-- AnyServerName: normalized server name without the security checks (used in presence lookups)
123isValidServerName : Text -> Bool
148isValidServerName : Text -> Bool
124isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
149isValidServerName = \s -> not (tooLong s maxServerNameLen) && s == normalizeServerName s && not (isBadServerName s)
125type ServerName = Text where isValidServerName
150type ServerName = Text where isValidServerName
@@ -127,16 +152,26 @@ type AnyServerName = Text where \s -> not (tooLong s maxServerNameLen) && s == n
128type Timestamp = Int<Ms>
153type Timestamp = Int<Ms>
155+-- Pending challenge for Ed25519 challenge-response auth
130type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
156type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp}
157+-- Authenticated session; token is a random hex string
131type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
158type Session = {token: Text, pubkey: PubkeyHex, expiresAt: Timestamp}
159+-- Stored message; originServer is "" for locally-originated messages, set to sender server for federated forwards
132type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
160type Message = {messageId: MessageIdHex, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: Text, receivedAt: Timestamp}
161+-- Presence: tracks that a pubkey is online at a given server, with expiry
133type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
162type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp, timestamp: Timestamp}
163+-- Queued federation forward: message waiting to be delivered to a remote server
134type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
164type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
165+-- Tracks message IDs that were acked by a polling client, pending federation ack back to origin
135type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
166type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
167+-- Terminal state: message delivered and ack confirmed (used for dedup after ack overflow)
136type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
168type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
169+-- Tracks consecutive failures per federation peer for exponential backoff
137type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
170type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int}
171+-- Sliding-window rate limit state per pubkey
138type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
172type SendRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
139type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
173type ChallengeRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
174+-- Tracks active long-poll connections to enforce per-key concurrency limits
140type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
175type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
141data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
176data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
142data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
177data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName]} | SendDuplicate {}
@@ -145,6 +180,8 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
145-- ============================================
180-- ============================================
146-- Persisted State
181-- Persisted State
147-- ============================================
182-- ============================================
183+-- All state is in-memory (no disk persistence). Prefixed with * (mutable refs).
184+-- Atomic blocks provide transactional consistency for reads + writes.
149*challenges : [Challenge]
186*challenges : [Challenge]
@@ -154,10 +191,12 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
155*presence : [PresenceEntry]
192*presence : [PresenceEntry]
194+-- Monotonic counter ensuring unique receivedAt ordering even within the same millisecond
157*seqCounter : Timestamp
195*seqCounter : Timestamp
159*forwards : [ForwardEntry]
197*forwards : [ForwardEntry]
199+-- Messages acked by local recipients, pending federation ack to origin server
161*ackedIds : [AckedEntry]
200*ackedIds : [AckedEntry]
163*lastPrunedAt : Timestamp
202*lastPrunedAt : Timestamp
@@ -170,8 +209,12 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
171*challengeRates : [ChallengeRateEntry]
210*challengeRates : [ChallengeRateEntry]
212+-- Heartbeat signal: incremented periodically to wake all blocking long-polls
173*pollWakeup : Timestamp
213*pollWakeup : Timestamp
215+-- Per-bucket message signals (16 buckets, keyed by first hex digit of pubkey).
216+-- Updated when a message is stored for a recipient; polls for that pubkey's
217+-- bucket watch for changes to avoid waking unrelated polls.
175*messageSignal0 : Timestamp
218*messageSignal0 : Timestamp
176*messageSignal1 : Timestamp
219*messageSignal1 : Timestamp
177*messageSignal2 : Timestamp
220*messageSignal2 : Timestamp
@@ -191,14 +234,20 @@ data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
192*activePolls : [ActivePoll]
235*activePolls : [ActivePoll]
237+-- Overflow buffer: acks evicted from ackedIds before they could be sent to origin
194*ackOverflow : [AckedEntry]
238*ackOverflow : [AckedEntry]
240+-- Final dedup set: messages confirmed delivered (ack sent or overflow drained)
196*deliveredIds : [DeliveredEntry]
241*deliveredIds : [DeliveredEntry]
198-- ============================================
243-- ============================================
200-- ============================================
245-- ============================================
247+-- Authentication: Ed25519 challenge-response flow
248+-- 1. Client POSTs pubkey to /auth/challenge, gets a random challenge string
249+-- 2. Client signs the challenge with their private key and POSTs to /auth/verify
250+-- 3. Server verifies signature, returns a session token
202route AuthApi where
251route AuthApi where
204 POST {pubkey: PubkeyHex} /challenge
253 POST {pubkey: PubkeyHex} /challenge
@@ -208,30 +257,39 @@ route AuthApi where
208 -> {token: Text, expiresAt: Timestamp}
257 -> {token: Text, expiresAt: Timestamp}
209 = VerifyAuth
258 = VerifyAuth
260+-- Client-facing API (requires session token in Authorization header)
211route ClientApi where
261route ClientApi where
262+ -- Long-poll for new messages; cursor tracks last-seen receivedAt timestamp
212 POST {cursor: Timestamp} /poll headers {authorization: Text}
263 POST {cursor: Timestamp} /poll headers {authorization: Text}
213 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
264 -> {events: [PollEvent], cursor: Timestamp, authorized: Bool}
215 /messages
266 /messages
267+ -- Send an E2E encrypted message to a recipient pubkey
216 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}
268 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, messageId: OptionalMessageId} / headers {authorization: Text}
217 -> {status: MessageStatus, messageId: OptionalMessageId}
269 -> {status: MessageStatus, messageId: OptionalMessageId}
218 = SendMessage
270 = SendMessage
271+ -- Look up which server a pubkey is present on (for federation routing)
219 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}
272 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text}
220 -> {server: AnyServerName, authorized: Bool}
273 -> {server: AnyServerName, authorized: Bool}
221 = LookupPresence
274 = LookupPresence
276+-- Server-to-server federation API (requires federation secret)
223route FederationApi where
277route FederationApi where
224 /federation
278 /federation
279+ -- Receive presence gossip events from a peer server
225 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
280 POST {events: [GossipEvent], hopsRemaining: Int, fromServer: ServerName} /gossip headers {authorization: Text}
226 -> {ok: Bool}
281 -> {ok: Bool}
227 = RecvGossip
282 = RecvGossip
283+ -- Receive a forwarded message from a peer server on behalf of a sender
228 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
284 POST {toKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp, originServer: ServerName, messageId: MessageIdHex, fromServer: ServerName} /forward headers {authorization: Text}
229 -> {ok: Bool}
285 -> {ok: Bool}
230 = ForwardMessage
286 = ForwardMessage
287+ -- Acknowledge delivery of messages back to the origin server
231 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
288 POST {messageIds: [MessageIdHex], originServer: ServerName, fromServer: ServerName} /ack headers {authorization: Text}
232 -> {ok: Bool}
289 -> {ok: Bool}
233 = FederationAck
290 = FederationAck
292+-- Triggers pruning of expired state and retry of pending federation ops
235route MaintenanceApi where
293route MaintenanceApi where
236 POST {} /maintenance/prune headers {authorization: Text}
294 POST {} /maintenance/prune headers {authorization: Text}
237 -> {ok: Bool}
295 -> {ok: Bool}
@@ -240,6 +298,7 @@ route MaintenanceApi where
240route HealthApi where
298route HealthApi where
241 GET /health -> {ok: Bool} = Health
299 GET /health -> {ok: Bool} = Health
301+-- Returns counts of in-memory collections (requires federation or maintenance secret)
243route StatsApi where
302route StatsApi where
244 GET /stats headers {authorization: Text}
303 GET /stats headers {authorization: Text}
245 -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ok: Bool}
304 -> {sessions: Int, messages: Int, presence: Int, forwards: Int, challenges: Int, ackedIds: Int, ok: Bool}
@@ -274,12 +333,14 @@ minBy = \items getKey -> fold (\acc x -> case acc of
274 Nothing {} -> Just {value: x}
333 Nothing {} -> Just {value: x}
275 Just {value: v} -> if getKey x < getKey v then Just {value: x} else acc) Nothing {} items
334 Just {value: v} -> if getKey x < getKey v then Just {value: x} else acc) Nothing {} items
336+-- Hex encoding utilities for token generation
277hexDigit : Int -> Text
337hexDigit : Int -> Text
278hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
338hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
280byteToHex : Int -> Text
340byteToHex : Int -> Text
281byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
341byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
343+-- Encode a 24-bit integer as 6 hex characters (3 bytes)
283intTo3Hex : Int -> Text
344intTo3Hex : Int -> Text
284intTo3Hex = \n ->
345intTo3Hex = \n ->
285 let b0 = n / 65536 in
346 let b0 = n / 65536 in
@@ -288,11 +349,13 @@ intTo3Hex = \n ->
288 let b2 = r0 - b1 * 256 in
349 let b2 = r0 - b1 * 256 in
289 byteToHex b0 ++ byteToHex b1 ++ byteToHex b2
350 byteToHex b0 ++ byteToHex b1 ++ byteToHex b2
352+-- Generate a random hex token (n chunks of 3 bytes each = 6n hex chars)
291genTokenN = \n -> if n <= 0 then yield "" else do
353genTokenN = \n -> if n <= 0 then yield "" else do
292 chunk <- randomInt 16777216
354 chunk <- randomInt 16777216
293 rest <- genTokenN (n - 1)
355 rest <- genTokenN (n - 1)
294 yield (intTo3Hex chunk ++ rest)
356 yield (intTo3Hex chunk ++ rest)
358+-- Default token: 8 chunks = 48 hex chars = 24 bytes of randomness
296genToken = genTokenN 8
359genToken = genTokenN 8
298tooLong : Text -> Int -> Bool
361tooLong : Text -> Int -> Bool
@@ -323,6 +386,7 @@ isAllowedServerChar = \c -> contains c "0123456789abcdefghijklmnopqrstuvwxyz-.:"
323containsBadUrlChar : Text -> Bool
386containsBadUrlChar : Text -> Bool
324containsBadUrlChar = \s -> any (chars s) (\c -> not (isAllowedServerChar c))
387containsBadUrlChar = \s -> any (chars s) (\c -> not (isAllowedServerChar c))
389+-- Carrier-grade NAT range: 100.64.0.0/10 (RFC 6598)
326isCgnatRange : Text -> Bool
390isCgnatRange : Text -> Bool
327isCgnatRange = \s ->
391isCgnatRange = \s ->
328 take 4 s == "100."
392 take 4 s == "100."
@@ -431,6 +495,9 @@ hasDotBeforePort = \s ->
431 else if take 1 s == "." then True {}
495 else if take 1 s == "." then True {}
432 else hasDotBeforePort (drop 1 s)
496 else hasDotBeforePort (drop 1 s)
498+-- DNS rebinding / SSRF protection: reject server names that resolve to private/reserved addresses.
499+-- Blocks localhost, private RFC1918 ranges, link-local, loopback, CGNAT, documentation ranges,
500+-- IPv6 shorthand, octal/hex IP tricks, bare IPs without dots, and .arpa domains.
434isBadServerName : Text -> Bool
501isBadServerName : Text -> Bool
435isBadServerName = \sn ->
502isBadServerName = \sn ->
@@ -462,18 +529,21 @@ isBadServerName = \sn ->
462 || not (hasDotBeforePort sn)
529 || not (hasDotBeforePort sn)
463 || containsSuffix ".arpa" sn
530 || containsSuffix ".arpa" sn
532+-- Look up a non-expired session matching the given token (constant-time comparison)
465authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
533authenticateSession : [Session] -> Text -> Timestamp -> AuthResult
466authenticateSession = \sessions token t ->
534authenticateSession = \sessions token t ->
467 case headOf (do s <- sessions; where s.expiresAt > t && safeCompare s.token token; yield s) of
535 case headOf (do s <- sessions; where s.expiresAt > t && safeCompare s.token token; yield s) of
468 Nothing {} -> Unauthorized {}
536 Nothing {} -> Unauthorized {}
469 Just {value: s} -> Authorized {pubkey: s.pubkey}
537 Just {value: s} -> Authorized {pubkey: s.pubkey}
539+-- Extract bearer token from "Authorization: Bearer <token>" header
471extractToken = \authHeader ->
540extractToken = \authHeader ->
472 let token = drop 7 authHeader in
541 let token = drop 7 authHeader in
473 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
542 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
474 then Just {value: token}
543 then Just {value: token}
475 else Nothing {}
544 else Nothing {}
546+-- Constant-time secret comparison (prevents timing attacks on federation/maintenance secrets)
477verifySecret : Text -> Text -> Bool
547verifySecret : Text -> Text -> Bool
478verifySecret = \token secret ->
548verifySecret = \token secret ->
479 secret != "" && safeCompare token secret
549 secret != "" && safeCompare token secret
@@ -505,9 +575,11 @@ deliveryToMessageStatus = \ds -> case ds of
505isQueued : DeliveryStatus -> Bool
575isQueued : DeliveryStatus -> Bool
506isQueued = \ds -> case ds of DsQueued {} -> True {}; _ -> False {}
576isQueued = \ds -> case ds of DsQueued {} -> True {}; _ -> False {}
578+-- Ensure strictly monotonic timestamps for message ordering
508nextSeqTs : Timestamp -> Timestamp -> Timestamp
579nextSeqTs : Timestamp -> Timestamp -> Timestamp
509nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
580nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
582+-- Deduplicate message IDs, preserving order, capped at maxMessageIds
511dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
583dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
512dedupMessageIds = \messageIds ->
584dedupMessageIds = \messageIds ->
513 fold (\acc mid ->
585 fold (\acc mid ->
@@ -515,10 +587,14 @@ dedupMessageIds = \messageIds ->
515 else if any acc (\a -> a == mid) then acc
587 else if any acc (\a -> a == mid) then acc
516 else union acc [mid]) [] messageIds
588 else union acc [mid]) [] messageIds
590+-- Remove the N oldest acked entries (by ackedAt timestamp)
518evictOldestAckedIds = \items n ->
591evictOldestAckedIds = \items n ->
519 if n <= 0 then items
592 if n <= 0 then items
520 else diff items (takeRelation n (sortBy (\a -> a.ackedAt) items))
593 else diff items (takeRelation n (sortBy (\a -> a.ackedAt) items))
595+-- Constant-time string comparison to prevent timing side-channels.
596+-- Always iterates maxTokenLen characters regardless of input length;
597+-- rejects empty strings to avoid vacuous equality.
522safeCompareLoop : Text -> Text -> Int -> Int -> Int
598safeCompareLoop : Text -> Text -> Int -> Int -> Int
523safeCompareLoop = \s1 s2 n result ->
599safeCompareLoop = \s1 s2 n result ->
524 if n <= 0 && s1 == "" && s2 == ""
600 if n <= 0 && s1 == "" && s2 == ""
@@ -529,6 +605,7 @@ safeCompare : Text -> Text -> Bool
529safeCompare = \a b ->
605safeCompare = \a b ->
530 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
606 safeCompareLoop a b maxTokenLen 0 + (if a == "" then 1 else 0) + (if b == "" then 1 else 0) == 0
608+-- Remove the single item with the smallest key value (LRU-style eviction)
532evictOneOldest = \items getKey ->
609evictOneOldest = \items getKey ->
533 let result = fold (\acc x ->
610 let result = fold (\acc x ->
534 case acc.candidate of
611 case acc.candidate of
@@ -542,6 +619,8 @@ evictOneOldest = \items getKey ->
542 Nothing {} -> items
619 Nothing {} -> items
543 Just {} -> result.items
620 Just {} -> result.items
622+-- Enforce per-key capacity: if a key has >= maxPerKey items, evict the oldest for that key.
623+-- Also prunes expired items.
545capPerKey = \items matchKey maxPerKey getExpiry t ->
624capPerKey = \items matchKey maxPerKey getExpiry t ->
546 let active = do x <- items; where getExpiry x > t; yield x in
625 let active = do x <- items; where getExpiry x > t; yield x in
547 let forKey = do x <- active; where matchKey x; yield x in
626 let forKey = do x <- active; where matchKey x; yield x in
@@ -550,11 +629,13 @@ capPerKey = \items matchKey maxPerKey getExpiry t ->
550 union others (evictOneOldest forKey getExpiry)
629 union others (evictOneOldest forKey getExpiry)
551 else active
630 else active
632+-- Enforce global capacity: if at maxCount, evict the single oldest item
553capGlobal = \items maxCount getKey ->
633capGlobal = \items maxCount getKey ->
554 if count items >= maxCount
634 if count items >= maxCount
555 then evictOneOldest items getKey
635 then evictOneOldest items getKey
556 else items
636 else items
638+-- Two-tier capacity enforcement: per-key first, then global
558capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
639capPerKeyAndGlobal = \items matchKey maxPerKey getExpiry t maxGlobal ->
559 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
640 capGlobal (capPerKey items matchKey maxPerKey getExpiry t) maxGlobal getExpiry
@@ -566,6 +647,7 @@ removeOneForKey = \items pubkey ->
566 ) {removed: False {}, items: []} items
647 ) {removed: False {}, items: []} items
567 in result.items
648 in result.items
650+-- Extract unique remote (non-local) server names from a list of items
569collectPeerServers : [a] -> (a -> Text) -> [Text]
651collectPeerServers : [a] -> (a -> Text) -> [Text]
570collectPeerServers = \items getServer ->
652collectPeerServers = \items getServer ->
571 fold (\acc item ->
653 fold (\acc item ->
@@ -577,6 +659,7 @@ collectPeerServers = \items getServer ->
577samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
659samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server
578pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
660pubkeyServer = \e -> {pubkey: e.pubkey, server: e.server}
662+-- Upsert items by key, keeping the fresher entry on conflict (last-write-wins by freshness)
580mergeByKey = \sameKey getFreshness initial items ->
663mergeByKey = \sameKey getFreshness initial items ->
581 fold (\acc e ->
664 fold (\acc e ->
582 if any acc (\a -> sameKey a e && getFreshness a >= getFreshness e)
665 if any acc (\a -> sameKey a e && getFreshness a >= getFreshness e)
@@ -585,11 +668,14 @@ mergeByKey = \sameKey getFreshness initial items ->
585 (do a <- acc; where not (sameKey a e); yield a)
668 (do a <- acc; where not (sameKey a e); yield a)
586 [e]) initial items
669 [e]) initial items
671+-- Merge presence entries by (pubkey, server), keeping the one with the latest expiresAt
588mergeByPubkeyServer = mergeByKey samePubkeyServer (\x -> x.expiresAt)
672mergeByPubkeyServer = mergeByKey samePubkeyServer (\x -> x.expiresAt)
674+-- Keep entries from setA that are not superseded by a newer-timestamped entry in setB
590resolveByTimestamp = \setA setB ->
675resolveByTimestamp = \setA setB ->
591 do e <- setA; where none setB (\o -> samePubkeyServer o e && o.timestamp > e.timestamp); yield e
676 do e <- setA; where none setB (\o -> samePubkeyServer o e && o.timestamp > e.timestamp); yield e
678+-- Remove presence entries that have a corresponding offline event with a newer timestamp
593removeOffline = \presenceSet trustedOffline t ->
679removeOffline = \presenceSet trustedOffline t ->
594 diff presenceSet (do
680 diff presenceSet (do
595 p <- presenceSet
681 p <- presenceSet
@@ -597,6 +683,8 @@ removeOffline = \presenceSet trustedOffline t ->
597 where samePubkeyServer p o && min o.timestamp t > p.timestamp
683 where samePubkeyServer p o && min o.timestamp t > p.timestamp
686+-- Cap presence table after gossip merge. If over capacity, only allow updates to
687+-- existing entries and reclaim slots from expired ones for new entries.
600capGossipPresence = \args ->
688capGossipPresence = \args ->
601 if count args.afterOffline <= maxPresenceEntries
689 if count args.afterOffline <= maxPresenceEntries
602 then {finalPresence: args.afterOffline, appliedOnline: args.appliedOnline, withinCapacity: True {}}
690 then {finalPresence: args.afterOffline, appliedOnline: args.appliedOnline, withinCapacity: True {}}
@@ -620,6 +708,8 @@ countMessagesFor = \messages toKey t ->
620hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
708hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
621hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
709hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
711+-- Validate incoming gossip events: reject local-server events, future timestamps,
712+-- and already-expired entries. Clamp TTLs and split into online/offline buckets.
623validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
713validGossipEvents : [GossipEvent] -> Timestamp -> {online: [ValidGossipEntry], offline: [ValidGossipEntry]}
624validGossipEvents = \events t ->
714validGossipEvents = \events t ->
625 let validate = \e ->
715 let validate = \e ->
@@ -636,6 +726,7 @@ validGossipEvents = \events t ->
636 GossipOffline {} -> {online: acc.online, offline: union acc.offline (validate e)}
726 GossipOffline {} -> {online: acc.online, offline: union acc.offline (validate e)}
637 _ -> acc) {online: [], offline: []} events
727 _ -> acc) {online: [], offline: []} events
729+-- Execute an action for each server in parallel (fork per server)
639sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}
730sendToEach : [ServerName] -> (ServerName -> (a {})) -> a {}
640sendToEach = \servers action ->
731sendToEach = \servers action ->
641 case headOf servers of
732 case headOf servers of
@@ -645,6 +736,8 @@ sendToEach = \servers action ->
645 fork (action server)
736 fork (action server)
646 sendToEach rest action
737 sendToEach rest action
739+-- Remove all expired entries from every collection. Returns expired local presence
740+-- entries (for broadcasting offline gossip) and the set of peer servers known before pruning.
648pruneExpired = \t -> do
741pruneExpired = \t -> do
649 challenges <- *challenges
742 challenges <- *challenges
650 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
743 set *challenges = do x <- challenges; where x.expiresAt > t; yield x
@@ -675,6 +768,7 @@ pruneExpired = \t -> do
675 set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > messageCutoff; yield d
768 set *deliveredIds = do d <- deliveredIds; where d.deliveredAt > messageCutoff; yield d
676 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
769 yield {expiredLocal: expiredLocalPresence presenceBefore presenceAfter t, peerServers: peerServersBefore}
771+-- Opportunistic pruning: runs at most once per second, triggered by request handlers
678maybePrune = \t -> do
772maybePrune = \t -> do
679 lastPruned <- *lastPrunedAt
773 lastPruned <- *lastPrunedAt
680 if t - lastPruned < 1000
774 if t - lastPruned < 1000
@@ -759,6 +853,7 @@ removeForward = \mid srv -> atomic do
759 forwards <- *forwards
853 forwards <- *forwards
760 set *forwards = forwardsExcluding forwards srv [mid]
854 set *forwards = forwardsExcluding forwards srv [mid]
856+-- Exponential backoff: 30s -> 60s -> 120s -> 240s -> 480s (capped)
762retryBackoff : Int -> Int<Ms>
857retryBackoff : Int -> Int<Ms>
763retryBackoff = \failures ->
858retryBackoff = \failures ->
764 if failures <= 1 then retryBackoffBase
859 if failures <= 1 then retryBackoffBase
@@ -808,6 +903,7 @@ retryForwardsForServer = \entries remaining ->
808 clearFailure f.toServer
903 clearFailure f.toServer
809 retryForwardsForServer rest (remaining - 1)
904 retryForwardsForServer rest (remaining - 1)
906+-- Retry all pending forwards, respecting per-server backoff. Skips servers that failed recently.
811retryPendingForwards : [ForwardEntry] -> IO {network} {}
907retryPendingForwards : [ForwardEntry] -> IO {network} {}
812retryPendingForwards = \pending -> do
908retryPendingForwards = \pending -> do
813 let servers = collectPeerServers pending (\x -> x.toServer)
909 let servers = collectPeerServers pending (\x -> x.toServer)
@@ -818,6 +914,7 @@ retryPendingForwards = \pending -> do
818 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
914 retryForwardsForServer (forwardsForServer pending server) maxForwardRetriesPerServer)
917+-- When a remote key comes online, create forward entries for queued messages and send them
821forwardQueuedMessages = \targets t -> do
918forwardQueuedMessages = \targets t -> do
822 newForwards <- atomic do
919 newForwards <- atomic do
823 messages <- *messages
920 messages <- *messages
@@ -851,12 +948,16 @@ sendBatchedAcksToServer = \serverAcks server ->
851 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
948 let updatedAckedIds = ackedIdsExcludingBatch ackedIds server batch
852 set *ackedIds = updatedAckedIds)
949 set *ackedIds = updatedAckedIds)
951+-- Send delivery acknowledgments back to origin servers and remove acked entries on success
854sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
952sendAcksAndClean : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> IO {network} {}
855sendAcksAndClean = \pendingAcks ->
953sendAcksAndClean = \pendingAcks ->
856 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in
954 let servers = collectPeerServers pendingAcks (\a -> a.originServer) in
857 sendToEach servers (\server ->
955 sendToEach servers (\server ->
858 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
956 sendBatchedAcksToServer (acksForServer pendingAcks server) server)
958+-- Retry sending acks that were evicted from ackedIds due to capacity limits.
959+-- On failure, re-insert into ackedIds or overflow buffer; overflow entries eventually
960+-- move to deliveredIds (terminal dedup state) if the buffer is also full.
860retryEvictedAcks = \evictedEntries pendingAcks ->
961retryEvictedAcks = \evictedEntries pendingAcks ->
861 let evictedToRetry = retriableEvictedAcks evictedEntries pendingAcks
962 let evictedToRetry = retriableEvictedAcks evictedEntries pendingAcks
@@ -897,6 +998,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
897 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
998 (println ("ackMessages: overflow at capacity (" ++ show maxAckOverflow ++ "), " ++ show dropped ++ " ack entries moved to delivered set for " ++ server)))
898 Ok _ -> yield {})
999 Ok _ -> yield {})
1001+-- Retry acks sitting in the overflow buffer (last chance before they become deliveredIds)
900retryOverflowAcks = \overflow ->
1002retryOverflowAcks = \overflow ->
901 let servers = collectPeerServers overflow (\a -> a.originServer) in
1003 let servers = collectPeerServers overflow (\a -> a.originServer) in
902 sendToEach servers (\server ->
1004 sendToEach servers (\server ->
@@ -908,6 +1010,9 @@ retryOverflowAcks = \overflow ->
908 ackOverflow <- *ackOverflow
1010 ackOverflow <- *ackOverflow
909 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
1011 set *ackOverflow = do a <- ackOverflow; where not (a.originServer == server && hasMessage a.messageId a.toKey batch); yield a))
1013+-- Full maintenance cycle: prune expired state, retry pending forwards and acks,
1014+-- retry overflow acks, broadcast offline gossip, and refresh presence for local keys
1015+-- nearing expiry. Called by both the background task and the /maintenance/prune endpoint.
911pruneAndRetry = \t -> do
1016pruneAndRetry = \t -> do
912 result <- atomic do
1017 result <- atomic do
913 set *lastPrunedAt = t
1018 set *lastPrunedAt = t
@@ -941,9 +1046,13 @@ hexCharIndex = \c ->
941 else if c == "8" then 8 else if c == "9" then 9 else if c == "a" then 10 else if c == "b" then 11
1046 else if c == "8" then 8 else if c == "9" then 9 else if c == "a" then 10 else if c == "b" then 11
942 else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15
1047 else if c == "c" then 12 else if c == "d" then 13 else if c == "e" then 14 else 15
1049+-- Map pubkey to one of 16 signal buckets based on first hex digit.
1050+-- This partitions the signal space so storing a message for one pubkey
1051+-- only wakes polls in the same bucket, not all active polls.
944pubkeyBucket : PubkeyHex -> Int
1052pubkeyBucket : PubkeyHex -> Int
945pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)
1053pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)
1055+-- Update the signal timestamp for a pubkey's bucket, waking any blocked polls in that bucket
947signalMessageForKey = \pubkey ts ->
1056signalMessageForKey = \pubkey ts ->
948 let b = pubkeyBucket pubkey in
1057 let b = pubkeyBucket pubkey in
949 if b == 0 then set *messageSignal0 = ts
1058 if b == 0 then set *messageSignal0 = ts
@@ -982,6 +1091,9 @@ readMessageSignal = \pubkey ->
982 else if b == 14 then do s <- *messageSignal14; yield s
1091 else if b == 14 then do s <- *messageSignal14; yield s
983 else do s <- *messageSignal15; yield s
1092 else do s <- *messageSignal15; yield s
1094+-- Acknowledge messages up to the given cursor: remove from messages, cancel pending forwards,
1095+-- and create ack entries for federated messages (to notify origin servers).
1096+-- Returns pending acks and any evicted entries that exceeded ackedIds capacity.
985ackPolledMessages = \pubkey cursor t -> do
1097ackPolledMessages = \pubkey cursor t -> do
986 messages <- *messages
1098 messages <- *messages
987 let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
1099 let toAck = do m <- messages; where m.toKey == pubkey && m.receivedAt <= cursor; yield m
@@ -1012,6 +1124,12 @@ pollNewEvents = \pubkey clampedCursor t -> do
1012 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
1124 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
1013 yield {events: evts, cursor: lastReceivedAt}
1125 yield {events: evts, cursor: lastReceivedAt}
1127+-- Execute a single long-poll cycle:
1128+-- 1. Authenticate and ack messages up to cursor (atomic)
1129+-- 2. Acquire a poll slot (limit concurrent polls per key)
1130+-- 3. Check for existing messages; if none, block on signal/heartbeat (retry-based wait)
1131+-- 4. Re-authenticate after waking (session may have expired during wait)
1132+-- 5. Return new events and release the poll slot
1015pollOnce = \token cursor -> do
1133pollOnce = \token cursor -> do
1016 t <- now
1134 t <- now
1017 setup <- atomic do
1135 setup <- atomic do
@@ -1079,41 +1197,51 @@ pollOnce = \token cursor -> do
1079-- ============================================
1197-- ============================================
1080-- Derived Relations
1198-- Derived Relations
1081-- ============================================
1199-- ============================================
1200+-- Pure query functions over in-memory state. No side effects.
1202+-- All non-expired presence entries for a given pubkey
1083presenceForKey = \presence pubkey t ->
1203presenceForKey = \presence pubkey t ->
1084 do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
1204 do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield p
1206+-- Pubkeys that are online on this server
1086localOnlineKeys = \presence t ->
1207localOnlineKeys = \presence t ->
1087 do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
1208 do p <- presence; where isLocalServer p.server && p.expiresAt > t; yield p.pubkey
1210+-- Remove the local presence entry for a pubkey (used before re-inserting on auth)
1089presenceWithoutLocalKey = \presence pubkey ->
1211presenceWithoutLocalKey = \presence pubkey ->
1090 do p <- presence; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
1212 do p <- presence; where p.pubkey != pubkey || not (isLocalServer p.server); yield p
1214+-- Local presence entries that expired during this prune cycle (for offline gossip)
1092expiredLocalPresence = \before after t ->
1215expiredLocalPresence = \before after t ->
1093 do p <- before; where p.expiresAt <= t && isLocalServer p.server && none after (\q -> q.pubkey == p.pubkey && isLocalServer q.server); yield p
1216 do p <- before; where p.expiresAt <= t && isLocalServer p.server && none after (\q -> q.pubkey == p.pubkey && isLocalServer q.server); yield p
1218+-- Resolve the best server for a pubkey: prefer local, otherwise pick the remote with latest expiry
1095bestServerForKey = \entries ->
1219bestServerForKey = \entries ->
1096 if any entries (\p -> isLocalServer p.server) then serverHostN
1220 if any entries (\p -> isLocalServer p.server) then serverHostN
1097 else case maxBy (do p <- entries; where not (isLocalServer p.server); yield p) (\p -> p.expiresAt) of
1221 else case maxBy (do p <- entries; where not (isLocalServer p.server); yield p) (\p -> p.expiresAt) of
1098 Just {value: p} -> p.server
1222 Just {value: p} -> p.server
1099 Nothing {} -> ""
1223 Nothing {} -> ""
1225+-- Messages for a pubkey newer than cursor and within messageMaxAge
1101pollableMessages = \messages pubkey cursor t ->
1226pollableMessages = \messages pubkey cursor t ->
1102 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
1227 do m <- messages; where m.toKey == pubkey && m.receivedAt >= t - messageMaxAge && m.receivedAt > cursor; yield m
1229+-- Build ack entries for messages that came from a remote origin server (need federation ack)
1104federatedAckEntries = \ackedMessages t ->
1230federatedAckEntries = \ackedMessages t ->
1105 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
1231 do m <- ackedMessages; where m.originServer != ""; yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
1233+-- Remove forwards for specific message IDs destined to a given server
1107forwardsExcluding = \forwards server messageIds ->
1234forwardsExcluding = \forwards server messageIds ->
1108 do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
1235 do f <- forwards; where not (f.toServer == server && any messageIds (\mid -> f.messageId == mid)); yield f
1237+-- Find locally-originated messages that haven't been forwarded yet to each target's server
1110forwardCandidates = \targets messages forwards t ->
1238forwardCandidates = \targets messages forwards t ->
1111 fold (\acc target ->
1239 fold (\acc target ->
1112 let serverFwdIds = do f <- forwards; where f.toServer == target.server; yield f.messageId in
1240 let serverFwdIds = do f <- forwards; where f.toServer == target.server; yield f.messageId in
1113 union acc (do
1241 union acc (do
1114 m <- messages
1242 m <- messages
1115 where m.toKey == target.pubkey
1243 where m.toKey == target.pubkey
1116− where m.originServer == ""
1244+ where m.originServer == "" -- only forward locally-originated messages
1117 where none serverFwdIds (\fid -> fid == m.messageId)
1245 where none serverFwdIds (\fid -> fid == m.messageId)
1118 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
1246 yield {messageId: m.messageId, toServer: target.server, createdAt: t,
1119 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp})) [] targets
1247 toKey: m.toKey, encryptedBlob: m.encryptedBlob, timestamp: m.timestamp})) [] targets
@@ -1137,32 +1265,41 @@ acksForServer = \acks server -> do a <- acks; where a.originServer == server; yi
1138forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f
1266forwardsForServer = \forwards server -> do f <- forwards; where f.toServer == server; yield f
1268+-- Only trust offline events that claim to be from the sending server itself
1140trustedOfflineEvents = \events fromServer ->
1269trustedOfflineEvents = \events fromServer ->
1141 do e <- events; where e.server == fromServer; yield e
1270 do e <- events; where e.server == fromServer; yield e
1272+-- Peer servers to re-gossip to, excluding the server that sent us the gossip
1143gossipTargets = \peerServers fromServer ->
1273gossipTargets = \peerServers fromServer ->
1144 do s <- peerServers; where s != fromServer; yield s
1274 do s <- peerServers; where s != fromServer; yield s
1276+-- Remote keys that just came online (triggers forwarding of queued messages)
1146newlyOnlineRemoteKeys = \events appliedKeys ->
1277newlyOnlineRemoteKeys = \events appliedKeys ->
1147 do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}
1278 do e <- events; where any appliedKeys (\a -> samePubkeyServer a e); where not (isLocalServer e.server); yield {pubkey: e.pubkey, server: e.server}
1280+-- Trust gossip entries if they're from the sender or about a server we already know about
1149trustedGossipEntries = \deduped presence fromServer t ->
1281trustedGossipEntries = \deduped presence fromServer t ->
1150 do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
1282 do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
1284+-- Keys whose presence was actually updated (new or extended expiry)
1152appliedOnlineKeys = \entries presence ->
1285appliedOnlineKeys = \entries presence ->
1153 do e <- entries; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e)
1286 do e <- entries; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e)
1288+-- Gossip entries that update existing presence (vs. creating new entries)
1155existingPresenceUpdates = \entries presence ->
1289existingPresenceUpdates = \entries presence ->
1156 do e <- entries; where any presence (\p -> samePubkeyServer p e); yield e
1290 do e <- entries; where any presence (\p -> samePubkeyServer p e); yield e
1292+-- Gossip entries for keys not yet in the presence table
1158newPresenceEntries = \entries presence ->
1293newPresenceEntries = \entries presence ->
1159 do e <- entries; where none presence (\p -> samePubkeyServer p e); yield e
1294 do e <- entries; where none presence (\p -> samePubkeyServer p e); yield e
1296+-- Generate offline gossip events for locally-expired presence entries
1161offlineGossipEvents = \expiredLocal -> do
1297offlineGossipEvents = \expiredLocal -> do
1162 p <- expiredLocal
1298 p <- expiredLocal
1163 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
1299 let ttl = clamp offlineGossipMinTtl maxPresenceTtl ((p.expiresAt - p.timestamp) / 1000)
1164 yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl}
1300 yield {eventType: GossipOffline {}, pubkey: p.pubkey, server: serverHostN, timestamp: p.expiresAt, ttl: ttl}
1302+-- Filter targets to only those not locally online (avoid forwarding to keys that can poll directly)
1166remoteOnlyTargets = \targets localOnline ->
1303remoteOnlyTargets = \targets localOnline ->
1167 do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
1304 do target <- targets; where none localOnline (\pk -> pk == target.pubkey); yield target
@@ -1172,6 +1309,8 @@ remoteOnlyTargets = \targets localOnline ->
1173serve = \req -> case req of
1310serve = \req -> case req of
1312+ -- POST /auth/challenge: Generate a random challenge for Ed25519 authentication.
1313+ -- Rate-limited per key; globally capped. Challenge expires in 1 minute.
1175 ReqChallenge {pubkey, respond} -> do
1314 ReqChallenge {pubkey, respond} -> do
1176 challenge <- genToken
1315 challenge <- genToken
1177 t <- now
1316 t <- now
@@ -1208,6 +1347,8 @@ serve = \req -> case req of
1208 forkBroadcastOffline t result.pr
1347 forkBroadcastOffline t result.pr
1209 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1348 yield (respond {challenge: if result.accepted then challenge else "", ok: result.accepted, expiresAt: if result.accepted then t + challengeExpiry else 0})
1350+ -- POST /auth/verify: Verify Ed25519 signature over challenge string.
1351+ -- On success: creates a session, registers local presence, and gossips online status.
1211 VerifyAuth {pubkey, challenge, signature, respond} -> do
1352 VerifyAuth {pubkey, challenge, signature, respond} -> do
1212 t <- now
1353 t <- now
1213 let expiry = t + sessionExpiry
1354 let expiry = t + sessionExpiry
@@ -1249,6 +1390,9 @@ serve = \req -> case req of
1249 forkBroadcastOffline t validation.pr
1390 forkBroadcastOffline t validation.pr
1250 yield (respond {token: "", expiresAt: 0})
1391 yield (respond {token: "", expiresAt: 0})
1393+ -- POST /poll: Long-poll for new messages. Cursor-based: messages at or before cursor
1394+ -- are acked (removed from mailbox). Blocks until new messages arrive or heartbeat fires.
1395+ -- Acks for federated messages are sent to origin servers in the background.
1252 Poll {cursor, authorization, respond} -> do
1396 Poll {cursor, authorization, respond} -> do
1253 result <- case extractToken authorization of
1397 result <- case extractToken authorization of
1254 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
1398 Nothing {} -> yield {authorized: False {}, events: [], cursor: 0, pendingAcks: [], evictedEntries: []}
@@ -1260,6 +1404,9 @@ serve = \req -> case req of
1260 retryEvictedAcks result.evictedEntries result.pendingAcks))
1404 retryEvictedAcks result.evictedEntries result.pendingAcks))
1261 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
1405 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})
1407+ -- POST /messages: Send an encrypted message to a recipient.
1408+ -- Deduplicates by messageId across messages, ackedIds, and deliveredIds.
1409+ -- Rate-limited per sender key. If recipient is known on a remote server, forwards via federation.
1263 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
1410 SendMessage {to, encryptedBlob, timestamp, messageId, authorization, respond} -> do
1264 t <- now
1411 t <- now
1265 effectiveMsgId <- if messageId != "" then yield messageId else genToken
1412 effectiveMsgId <- if messageId != "" then yield messageId else genToken
@@ -1329,6 +1476,8 @@ serve = \req -> case req of
1329 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
1476 SendDone {delivery} -> respond {status: deliveryToMessageStatus delivery, messageId: effectiveMsgId})
1479+ -- GET /lookup/:pubkey: Look up the best known server for a pubkey.
1480+ -- Returns local server if key is online locally, otherwise the remote with latest expiry.
1332 LookupPresence {pubkey, authorization, respond} -> do
1481 LookupPresence {pubkey, authorization, respond} -> do
1333 t <- now
1482 t <- now
1334 result <- atomic do
1483 result <- atomic do
@@ -1342,6 +1491,10 @@ serve = \req -> case req of
1342 forkBroadcastOffline t result.pr
1491 forkBroadcastOffline t result.pr
1343 yield (respond {server: result.server, authorized: result.authorized})
1492 yield (respond {server: result.server, authorized: result.authorized})
1494+ -- POST /federation/gossip: Receive presence gossip from a peer server.
1495+ -- Validates events, merges into presence table (last-write-wins), and re-gossips
1496+ -- to other known peers with decremented hop count. Triggers message forwarding
1497+ -- when a remote key comes online.
1345 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
1498 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
1346 t <- now
1499 t <- now
1347 let cappedHops = clamp 0 maxGossipHops hopsRemaining
1500 let cappedHops = clamp 0 maxGossipHops hopsRemaining
@@ -1370,6 +1523,9 @@ serve = \req -> case req of
1370 (forwardQueuedMessages newlyOnlineRemote t))
1523 (forwardQueuedMessages newlyOnlineRemote t))
1371 yield (respond {ok: True {}}))
1524 yield (respond {ok: True {}}))
1526+ -- POST /federation/forward: Receive a message forwarded from a peer server.
1527+ -- Only accepts messages where fromServer == originServer (no cascade forwarding).
1528+ -- Deduplicates and stores; rejects if recipient mailbox is full.
1373 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do
1529 ForwardMessage {toKey, encryptedBlob, timestamp, originServer, messageId, fromServer, authorization, respond} -> do
1374 t <- now
1530 t <- now
1375 withFederationAuth authorization t respond (not (isLocalServer originServer) && fromServer == originServer) (do
1531 withFederationAuth authorization t respond (not (isLocalServer originServer) && fromServer == originServer) (do
@@ -1401,6 +1557,9 @@ serve = \req -> case req of
1401 ForwardRejected {} -> False {}
1557 ForwardRejected {} -> False {}
1402 _ -> True {}}))
1558 _ -> True {}}))
1560+ -- POST /federation/ack: Peer server confirms delivery of messages we originated.
1561+ -- Removes corresponding forward entries. Also cleans up the original message if the
1562+ -- recipient has no local presence and no other pending forwards for that message.
1404 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1563 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1405 t <- now
1564 t <- now
1406 withFederationAuth authorization t respond (isLocalServer originServer && not (isLocalServer fromServer)) (do
1565 withFederationAuth authorization t respond (isLocalServer originServer && not (isLocalServer fromServer)) (do
@@ -1423,6 +1582,8 @@ serve = \req -> case req of
1423 forkBroadcastOffline t pruneResult
1582 forkBroadcastOffline t pruneResult
1424 yield (respond {ok: True {}}))
1583 yield (respond {ok: True {}}))
1585+ -- POST /maintenance/prune: Trigger a full maintenance cycle (prune + retry all pending ops).
1586+ -- Accepts either federation or maintenance secret.
1426 Prune {authorization, respond} -> do
1587 Prune {authorization, respond} -> do
1427 t <- now
1588 t <- now
1428 let authorized = verifyMaintenanceAuth authorization
1589 let authorized = verifyMaintenanceAuth authorization
@@ -1450,6 +1611,8 @@ serve = \req -> case req of
1450-- Entry Point
1611-- Entry Point
1451-- ============================================
1612-- ============================================
1614+-- Background loop: periodically prunes expired state and retries pending federation ops.
1615+-- Jitter prevents thundering herd when multiple servers restart simultaneously.
1453backgroundPrune = do
1616backgroundPrune = do
1454 jitter <- randomInt 5000
1617 jitter <- randomInt 5000
1455 sleep (backgroundPruneInterval + jitter)
1618 sleep (backgroundPruneInterval + jitter)
@@ -1457,6 +1620,8 @@ backgroundPrune = do
1457 fork (pruneAndRetry t)
1620 fork (pruneAndRetry t)
1458 backgroundPrune
1621 backgroundPrune
1623+-- Background loop: periodically bumps pollWakeup to unblock long-polls that have been
1624+-- waiting too long (heartbeat). Ensures polls return within pollTimeoutMs even if no messages arrive.
1460pollHeartbeat = do
1625pollHeartbeat = do
1461 jitter <- randomInt 2000
1626 jitter <- randomInt 2000
1462 sleep (pollTimeoutMs + jitter)
1627 sleep (pollTimeoutMs + jitter)