1-- Skrepka Relay Server (maximally simplified)
2-- Single-server implementation of the Skrepka protocol (v0.1).
3-- Every collection is a flat list; rate limiters are plain count-per-window.
4-- Token comparison is plain `==` (timing-attack vulnerable), retry backoff is linear,
5-- and many helpers are inlined. Performance suffers; clarity is the only goal.
7-- ============================================
9-- ============================================
10-- Top-level constants: bind address, federation identity, capacity ceilings,
11-- per-key/window rate limits, expiry/backoff/TTL durations, and protocol version.
12-- Values written here are defaults; every scalar (Int/Float/Text/Bool/Maybe ...)
13-- top-level constant can be overridden at startup via `--name=value` on the
14-- command line (see Knot's `knot_override_lookup`). Run with `--help` for the
17-- This server's canonical federation identity. Used as the `server` field in
18-- presence/gossip events, in signed federation challenges, and for self-checks
19-- (`isLocalServer`). Not a bind address — the listener accepts connections on
20-- every interface regardless of this value. May also differ from the URL
21-- clients use to reach the server (e.g. behind a reverse proxy or alternate
22-- DNS name); peers must resolve this name to reach us for federation.
24-- Must be unique per deployment. Every operator participating in federation
25-- has to recompile with their own public hostname here — peers reject events
26-- whose origin fails `isBadServerName` (which blocks the default "localhost"
27-- and other private/reserved names), so two default-built servers cannot
28-- federate with each other. The default is only suitable for a single-node,
29-- non-federated deployment.
31serverHost = "localhost"
32-- Network interface the HTTP listener binds to. "0.0.0.0" accepts connections
33-- on every IPv4 interface (the typical default behind a reverse proxy);
34-- "127.0.0.1" restricts the listener to localhost-only; "::" / "::1" do the
35-- IPv6 equivalents. Override with `--bindHost=...` at startup. Unrelated to
36-- `serverHost` above — this is the kernel-level bind address; that is the
37-- federation identity peers see.
40-- TCP port for incoming HTTP connections.
43-- URL scheme used when contacting peer federation servers.
44federationScheme : Text
45federationScheme = "https://"
46-- The Ed25519 federation keypair lives in `*federationPrivateKeyHex` /
47-- `*federationPublicKeyHex` refs and is generated on first startup; see `main`.
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).
51messageMaxAge : Int<Ms>
52messageMaxAge = 2592000000
53-- Length (chars) of a hex-encoded Ed25519 pubkey (32 bytes → 64 hex).
56-- Max length of a hex-encoded message body / opaque blob (40 MiB hex ≈ 20 MiB bytes).
59-- Length of a hex-encoded Ed25519 signature (64 bytes → 128 hex).
62-- Cap on the number of message IDs accepted in a single ack/lookup request.
65-- Cap on items in a batched send request.
68-- Cap on gossip events accepted or emitted in a single federation exchange.
71-- Cap on events returned by one long-poll response.
74-- Allowance for clock drift when validating client timestamps (5 s).
75clockSkewTolerance : Int<Ms>
76clockSkewTolerance = 5000
77-- Per-peer ceiling on outstanding forward attempts.
78maxForwardsPerServer : Int
79maxForwardsPerServer = 2000
80-- Max accepted length of a server hostname.
83-- Max accepted length of a hex-encoded message ID.
86-- Ceiling on outstanding auth challenges held in memory.
89-- Ceiling on concurrent active sessions.
92-- Ceiling on remembered delivered message IDs.
94maxDeliveredIds = 10000
95-- Per-pubkey ceiling on outstanding challenges (prevents one key from exhausting the pool).
96maxChallengesPerKey : Int
97maxChallengesPerKey = 5
98-- Per-pubkey ceiling on concurrent sessions.
99maxSessionsPerKey : Int
100maxSessionsPerKey = 10
101-- Validity window for an auth challenge before it must be re-issued (1 min).
102challengeExpiry : Int<Ms>
103challengeExpiry = 60000
104-- Challenge requests permitted per key inside its rate window.
105maxChallengeRatePerKey : Int
106maxChallengeRatePerKey = 5
107-- Window over which the per-key challenge rate is measured (1 s).
108challengeRateWindowMs : Int<Ms>
109challengeRateWindowMs = 1000
110-- Send-message requests permitted per key per window.
111maxSendRatePerKey : Int
112maxSendRatePerKey = 20
113-- Window for the per-key send rate limit (1 s).
114sendRateWindowMs : Int<Ms>
115sendRateWindowMs = 1000
116-- Receive/poll requests permitted per key per window.
117maxRecvRatePerKey : Int
118maxRecvRatePerKey = 10
119-- Window for the per-key receive rate limit (1 s).
120recvRateWindowMs : Int<Ms>
121recvRateWindowMs = 1000
122-- Messages permitted per (sender, recipient) pair within the pair window.
123maxPairRatePerPair : Int
124maxPairRatePerPair = 30
125-- Window for the per-pair rate limit (1 min).
126pairRateWindowMs : Int<Ms>
127pairRateWindowMs = 60000
128-- Federation requests permitted from a single peer server per window.
129maxFederationRatePerServer : Int
130maxFederationRatePerServer = 50
131-- Window for the per-peer federation rate limit (1 s).
132federationRateWindowMs : Int<Ms>
133federationRateWindowMs = 1000
134-- Presence/key lookups permitted per key per window.
135maxLookupRatePerKey : Int
136maxLookupRatePerKey = 5
137-- Window for the per-key lookup rate limit (1 s).
138lookupRateWindowMs : Int<Ms>
139lookupRateWindowMs = 1000
140-- Other client write requests permitted per key per window.
141maxClientWriteRatePerKey : Int
142maxClientWriteRatePerKey = 10
143-- Window for the per-key client-write rate limit (1 s).
144clientWriteRateWindowMs : Int<Ms>
145clientWriteRateWindowMs = 1000
146-- Requests permitted per source IP per window (covers unauthenticated traffic).
149-- Window for the per-IP rate limit (1 s).
150ipRateWindowMs : Int<Ms>
152-- Global ceiling on total request rate across all clients.
153maxGlobalRequestRate : Int
154maxGlobalRequestRate = 1000
155-- Window for the global request rate limit (1 s).
156globalRateWindowMs : Int<Ms>
157globalRateWindowMs = 1000
158-- Global ceiling on auth-endpoint requests (challenge + verify combined).
159maxAuthGlobalRate : Int
160maxAuthGlobalRate = 200
161-- Window for the global auth rate limit (1 s).
162authGlobalRateWindowMs : Int<Ms>
163authGlobalRateWindowMs = 1000
165-- How long a session token remains valid after issue (1 hour).
166sessionExpiry : Int<Ms>
167sessionExpiry = 3600000
168-- How long a received "online" presence gossip is considered fresh (90 min).
169-- Receivers anchor the TTL to their own wall clock at receive time, so this
170-- value isn't carried on the wire.
171onlineGossipTtl : Int<Ms>
172onlineGossipTtl = 5400000
173-- Times to retry a federation forward before giving up.
174maxForwardRetries : Int
175maxForwardRetries = 10
177-- Base delay for federation retry backoff (linear, not exponential — 30 s).
178retryBackoffBase : Int<Ms>
179retryBackoffBase = 30000
180-- Cap on federation retry backoff (8 min).
181retryBackoffMax : Int<Ms>
182retryBackoffMax = 480000
184-- Cadence of the background prune/cleanup task (30 s).
185backgroundPruneInterval : Int<Ms>
186backgroundPruneInterval = 30000
187-- Long-poll hold timeout before returning empty to the client (25 s).
188pollTimeoutMs : Int<Ms>
190-- Concurrent active long-polls permitted per pubkey.
193-- Cadence at which this server refreshes its own presence to peers (5 min).
194presenceRefreshInterval : Int<Ms>
195presenceRefreshInterval = 300000
197-- Window during which a delivered-but-unacked message is retained for re-delivery (2 min).
198ackGracePeriod : Int<Ms>
199ackGracePeriod = 120000
201-- Human-readable build identifier returned by the /health endpoint.
202serverBuildVersion : Text
203serverBuildVersion = "2026-04-27-simple"
205-- Master switch for federation; when False the server runs in single-node mode.
206federationEnabled : Bool
207federationEnabled = True {}
208federationUrl : Text -> Text
209federationUrl = \server -> federationScheme ++ server
210-- When True, reject requests that arrive without a client-IP header (for IP-based rate limits).
211requireClientIp : Bool
212requireClientIp = True {}
214-- Drop the Ms unit and convert milliseconds to seconds.
215msToS : Int<Ms> -> Int<S>
216msToS = \ms -> stripUnit ms / 1000
218-- Drop the S unit and convert seconds to milliseconds.
219sToMs : Int<S> -> Int<Ms>
220sToMs = \s -> stripUnit s * 1000
222-- ============================================
224-- ============================================
225-- Wire/state shapes plus refinement aliases for hex/server-name validation.
226-- Hex types are bounded length + lowercase-hex; server names are lowercased
227-- and stripped of trailing dots, with a deny-list filter (`isBadServerName`)
228-- applied where SSRF is a concern.
230-- One event in a /poll response: the opaque cipher blob addressed to the
231-- polling pubkey. The server cannot read the blob; clients derive any
232-- per-message id locally as `messageIdOf encryptedBlob`.
233type PollEvent = {encryptedBlob: BlobHex}
235-- Kind of presence event flowing through federation gossip. `GossipUnknown` is
236-- the parser's fallback for forward-compatibility with future event types.
237data GossipEventType = GossipOnline {} | GossipOffline {} | GossipUnknown {}
239impl ToJSON GossipEventType where
240 toJson et = toJson (case et of
241 GossipOnline {} -> "online"
242 GossipOffline {} -> "offline"
243 GossipUnknown {} -> "unknown")
245impl FromJSON GossipEventType where
246 parseJson t = (\s -> if s == "online" then GossipOnline {} else if s == "offline" then GossipOffline {} else GossipUnknown {}) (parseJson t : Text)
248-- One presence change exchanged between federated servers.
249type GossipEvent = {eventType: GossipEventType, pubkey: PubkeyHex}
251-- Outcome classification returned to the sender of /messages and per-entry in
252-- /messages/batch. The first four are success-ish (see `isSendSuccess`); the
264impl ToJSON MessageStatus where
265 toJson ms = toJson (case ms of
266 MsDelivered {} -> "delivered"
267 MsFederated {} -> "federated"
268 MsQueued {} -> "queued"
269 MsUnauthorized {} -> "unauthorized"
270 MsRejected {} -> "rejected"
271 MsRateLimited {} -> "rate_limited"
272 MsDuplicate {} -> "duplicate"
273 MsDraining {} -> "draining")
275-- True when `s` is non-empty lowercase hex of even length, with `maxLen` ceiling.
276isValidBoundedHex : Int -> Text -> Bool
277isValidBoundedHex = \maxLen s -> not tooLong s maxLen && isValidHex s
278-- True when `s` is lowercase hex of exactly `exactLen` characters (even by definition).
279isValidExactHex : Int -> Text -> Bool
280isValidExactHex = \exactLen s -> count (chars s) == exactLen && isValidHex s
281-- Hex-encoded Ed25519 public key (32 bytes → 64 chars exact).
282type PubkeyHex = Text where isValidExactHex pubkeyLen
283-- Hex-encoded ciphertext or opaque blob; capped at `maxBlobLen` (~20 MiB).
284type BlobHex = Text where isValidBoundedHex maxBlobLen
285-- Hex-encoded Ed25519 signature (64 bytes → 128 chars exact).
286type SignatureHex = Text where isValidExactHex signatureLen
287isValidMessageId : Text -> Bool
288isValidMessageId = isValidBoundedHex maxMessageIdLen
289-- Hex-encoded message id, length ≤ `maxMessageIdLen`.
290type MessageIdHex = Text where isValidMessageId
291-- Normalized hostname (lowercased, no trailing dot, length-bounded).
292isAnyServerName : Text -> Bool
293isAnyServerName = \s -> not tooLong s maxServerNameLen && s == normalizeServerName s
294-- A normalized hostname that also passes the SSRF deny-list (`isBadServerName`).
295isValidServerName : Text -> Bool
296isValidServerName = \s -> isAnyServerName s && not isBadServerName s
297-- Peer hostname suitable for outbound federation calls.
298type ServerName = Text where isValidServerName
299-- Normalized hostname with no SSRF filter; used when the value comes from
300-- presence rows that the client only ever sees masked to "remote".
301type AnyServerName = Text where isAnyServerName
303-- Wall-clock time in milliseconds since epoch.
304type Timestamp = Int<Ms>
306-- Outstanding nonce issued by /auth/challenge; consumed by /auth/verify.
307type Challenge = {pubkey: PubkeyHex, challenge: Text, expiresAt: Timestamp, ip: Text}
308-- An issued bearer token bound to (pubkey, ip, expiry).
312 expiresAt: Timestamp,
316-- Mailbox row: a stored message awaiting recipient pickup or federation forward.
317-- Identity is `messageIdOf encryptedBlob` (BLAKE3 of the ciphertext); we don't
318-- persist it because it's deterministically derivable. Storage TTL is fixed
319-- at `messageMaxAge`, so expiry is derived from `receivedAt`. Any send time
320-- the sender wants to convey lives inside `encryptedBlob` (the server cannot
322type Message = {toKey: PubkeyHex, encryptedBlob: BlobHex, receivedAt: Timestamp}
323-- Last-known location of a pubkey, either local (`server == serverHostN`) or
324-- learned from a peer via gossip. Multiple entries per pubkey are allowed.
325type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Timestamp}
326-- A message queued for cross-server delivery to `toServer`. `retries` is
327-- incremented by `retryPendingForwards` until `maxForwardRetries` is hit.
328-- Identity is derived from `encryptedBlob` (see `Message`).
330 toServer: ServerName,
331 createdAt: Timestamp,
333 encryptedBlob: BlobHex,
336-- Replay-protection record: a message id we already delivered locally.
337type DeliveredEntry = {messageId: MessageIdHex, toKey: PubkeyHex, deliveredAt: Timestamp}
338-- Running failure counter for a peer; keyed by `server`. Drives `retryBackoff`.
339type ForwardFailure = {server: Text, failedAt: Timestamp, failures: Int, firstFailedAt: Timestamp}
340-- One fixed-window rate counter. `key` identifies the subject (pubkey, ip,
341-- server, or composite "sender|recipient" string).
342type RateEntry = {key: Text, windowStart: Timestamp, count: Int}
343-- A live long-poll slot held by a pubkey; used to cap concurrency per key.
344type ActivePoll = {pubkey: PubkeyHex, startedAt: Timestamp}
346-- One entry of a /messages/batch request body.
347type BatchMessage = {to: PubkeyHex, encryptedBlob: BlobHex}
348-- Per-entry result returned by /messages/batch.
349type BatchResult = {messageId: MessageIdHex, status: MessageStatus}
351-- ============================================
352-- Persisted State (all flat lists, no sharding)
353-- ============================================
354-- Mutable refs (`*x`) holding the entire server state in memory. Every list
355-- scan is O(n); scaling is the first thing the production version replaces.
356-- Grouped here: auth (challenges/sessions), mailbox (messages/forwards/acks),
357-- presence, rate-limit windows, drain phase, and bookkeeping counters.
359*challenges : [Challenge]
362*presence : [PresenceEntry]
363*forwards : [ForwardEntry]
364*deliveredIds : [DeliveredEntry]
365*forwardFailures : [ForwardFailure]
366*activePolls : [ActivePoll]
368*sendRates : [RateEntry]
369*recvRates : [RateEntry]
370*pairRates : [RateEntry]
371*challengeRates : [RateEntry]
372*lookupRates : [RateEntry]
373*clientWriteRates : [RateEntry]
374*federationRates : [RateEntry]
375*ipRates : [RateEntry]
377*globalRateCount : Int
378*globalRateWindowStart : Timestamp
379*authGlobalRateCount : Int
380*authGlobalRateWindowStart : Timestamp
382*seqCounter : Timestamp
383*pollWakeup : Timestamp
384*lastPrunedAt : Timestamp
385*lastHeartbeatAt : Timestamp
386*lastPresenceRefresh : Timestamp
387*startedAt : Timestamp
388*serverGeneration : Int
391*drainCompletedAt : Timestamp
393*runtimeMaintenanceSecret : Text
395*federationPrivateKeyHex : Text
396*federationPublicKeyHex : Text
398*blockedPeers : [Text]
399*allowedPeers : [Text]
404-- ============================================
405-- Routes (response shapes preserved; removed metric fields are zero-filled)
406-- ============================================
407-- Public HTTP surface, grouped by audience:
408-- AuthApi -- challenge/verify/revoke/logout (client identity)
409-- ClientApi -- /poll, send/batch/list/ack/delete, presence lookup
410-- FederationApi -- inter-server gossip, forward, ack, delivery receipts
411-- FederationPubkey -- exposes this server's federation Ed25519 public key
412-- MaintenanceApi -- operator endpoints (drain, prune, peer allow/block, admin)
413-- HealthApi/StatsApi/MetricsApi/ReadyApi -- observability
414-- Response field shapes are kept stable for backwards compatibility even when
415-- the underlying value is no longer computed (zero-filled).
418 POST {pubkey: PubkeyHex} /auth/challenge headers {clientIp: Text} -> {challenge: Text, ok: Bool, expiresAt: Timestamp, retryAfterMs: Int, error: Text} = ReqChallenge
419 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /auth/verify headers {clientIp: Text} -> {token: Text, expiresAt: Timestamp, error: Text} = VerifyAuth
420 POST {targetToken: Text} /auth/revoke headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = RevokeSession
421 POST /auth/logout headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool} = Logout
424 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
425 POST {to: PubkeyHex, encryptedBlob: BlobHex} /messages headers {authorization: Text, clientIp: Text} -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int} = SendMessage
426 POST {to: PubkeyHex, messageId: MessageIdHex} /messages/status headers {authorization: Text, clientIp: Text} -> {status: Text, authorized: Bool} = MessageStatus
427 POST {messageIds: [MessageIdHex]} /messages/ack headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, ackedCount: Int, error: Text} = AckMessages
428 POST {messages: [BatchMessage]} /messages/batch headers {authorization: Text, clientIp: Text} -> {results: [BatchResult], authorized: Bool, retryAfterMs: Int} = SendBatch
429 POST {offset: Int} /messages/list headers {authorization: Text, clientIp: Text} -> {messages: [{messageId: MessageIdHex, receivedAt: Timestamp}], authorized: Bool, count: Int, error: Text} = ListMessages
430 POST {messageIds: [MessageIdHex]} /messages/delete headers {authorization: Text, clientIp: Text} -> {ok: Bool, authorized: Bool, deletedCount: Int, error: Text} = DeleteMessages
431 GET /lookup/{pubkey: PubkeyHex} headers {authorization: Text, clientIp: Text} -> {server: AnyServerName, authorized: Bool, error: Text} = LookupPresence
432 POST {pubkeys: [PubkeyHex]} /lookup/batch headers {authorization: Text, clientIp: Text} -> {results: [{pubkey: PubkeyHex, server: AnyServerName}], authorized: Bool, error: Text} = BatchLookupPresence
434route FederationApi where
435 POST {events: [GossipEvent], fromServer: ServerName, requestTimestamp: Timestamp, correlationId: Text} /federation/gossip headers {authorization: Text} -> {ok: Bool, error: Text, retryAfterMs: Int} = RecvGossip
436 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
437 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
439route FederationPubkeyApi where
440 GET /federation/pubkey -> {pubkey: Text, ok: Bool} = FederationPubkey
442route MaintenanceApi where
443 POST /maintenance/prune headers {authorization: Text} -> {ok: Bool} = Prune
444 POST {phase: Int} /maintenance/drain headers {authorization: Text} -> {ok: Bool, drained: Int, phase: Int} = Drain
445 POST /maintenance/undrain headers {authorization: Text} -> {ok: Bool} = Undrain
446 POST /maintenance/shutdown headers {authorization: Text} -> {ok: Bool, drained: Int} = Shutdown
447 POST {server: ServerName} /maintenance/block headers {authorization: Text} -> {ok: Bool, blocked: Int} = BlockPeer
448 POST {server: ServerName} /maintenance/unblock headers {authorization: Text} -> {ok: Bool, unblocked: Bool} = UnblockPeer
449 GET /maintenance/blocked headers {authorization: Text} -> {peers: [Text], ok: Bool} = ListBlocked
450 POST {server: ServerName} /maintenance/ping headers {authorization: Text} -> {ok: Bool, latencyMs: Int, error: Text} = PingPeer
451 GET /maintenance/sessions/{pubkey: PubkeyHex} headers {authorization: Text} -> {sessions: [{token: Text, expiresAt: Timestamp, ip: Text, createdAt: Timestamp}], count: Int, ok: Bool} = AdminListSessions
452 POST {pubkey: PubkeyHex} /maintenance/sessions/revoke-all headers {authorization: Text} -> {ok: Bool, revokedCount: Int} = AdminRevokeSessions
453 POST {pubkey: PubkeyHex} /maintenance/messages/purge headers {authorization: Text} -> {ok: Bool, purgedCount: Int} = AdminPurgeMessages
454 POST {ip: Text} /maintenance/sessions/revoke-ip headers {authorization: Text} -> {ok: Bool, revokedCount: Int, affectedKeys: Int} = AdminRevokeIp
455 POST {limit: Int} /maintenance/sessions/list headers {authorization: Text} -> {sessions: [{pubkey: PubkeyHex, count: Int, ips: [Text], latestExpiry: Timestamp}], ok: Bool, total: Int} = AdminListAllSessions
456 GET /maintenance/rate-limits headers {authorization: Text} -> {ok: Bool, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} = GetRateLimits
457 POST {globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} /maintenance/rate-limits headers {authorization: Text} -> {ok: Bool} = SetRateLimits
458 POST /maintenance/nonces/flush headers {authorization: Text} -> {ok: Bool, flushedCount: Int} = FlushNonces
459 POST {server: ServerName} /maintenance/allow headers {authorization: Text} -> {ok: Bool, allowed: Int} = AllowPeer
460 POST {server: ServerName} /maintenance/unallow headers {authorization: Text} -> {ok: Bool, removed: Bool} = UnallowPeer
461 GET /maintenance/allowed headers {authorization: Text} -> {peers: [Text], ok: Bool} = ListAllowed
462 POST /maintenance/reconcile headers {authorization: Text} -> {ok: Bool, messageDrift: Int, presenceDrift: Int, sessionDrift: Int} = Reconcile
463 POST {newSecret: Text} /maintenance/rotate-secret headers {authorization: Text} -> {ok: Bool, error: Text} = RotateSecret
464 GET /maintenance/config headers {authorization: Text} -> {ok: Bool, pollTimeoutMs: Int, ackGracePeriod: Int} = GetConfig
467 GET /health headers {authorization: Text} -> {ok: Bool, version: Text, pruneHealthy: Bool, heartbeatHealthy: Bool, capacityHealthy: Bool, capacityWarning: Bool, federationHealthy: Bool, watchdogHealthy: Bool, nonceHealthy: Bool, forwardQueueHealthy: Bool, requestPressureHealthy: Bool, draining: Bool, degradedPeers: Int} = Health
469-- Per-peer health snapshot for /stats/peers. Latency/gossip-rate fields are
470-- zero-filled in this simplified server.
474 lastFailedAt: Timestamp,
475 firstFailedAt: Timestamp,
478 pendingForwards: Int,
482 gossipEventsPerMin: Int
484-- Single presence row exposed via /stats/key (server + expiry only).
485type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
486-- Mailbox depth ranking entry for /stats/top-recipients.
487type RecipientCount = {pubkey: PubkeyHex, count: Int}
490 GET /stats headers {authorization: Text} -> {ok: Bool, version: Text, capacity: {sessions: Int, sessionsMax: Int, sessionsPct: Int, messages: Int, totalMessageCount: Int, presence: Int, forwards: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, nonceEvictions: Int, activePolls: Int, recipientCounts: Int, messageBucketMax: Int, messageBucketMin: Int}, federation: {enabled: Bool, inFlight: Int, hungSlots: Int, avgLatencyMs: Int, maxLatencyMs: Int, callCount: Int, slowVerifies: Int, verifyFailures: Int, forwardFailures: Int, oldestForwardAge: Int, oldestAckAge: Int, recentGossip: Int, blockedPeers: Int, drainCompletedAt: Timestamp, forwardServers: Int, maxForwardQueueDepth: Int, topFailingPeers: [{server: Text, failures: Int}], gossipSuccess: Int, gossipFail: Int, latencyBuckets: {lt100: Int, lt500: Int, lt2000: Int, gte2000: Int}}, rates: {send: Int, challenge: Int, federation: Int, federationPair: Int, global: Int, ip: Int, lookup: Int, recv: Int, pair: Int, gossipEvent: Int, verifyIp: Int, uniqueKeyIp: Int, pendingAckCleanups: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedAuth: Int, rejectedPerKey: Int, rejectedFederation: Int}, timing: {pruneCount: Int, pruneRunning: Timestamp, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, draining: Bool, drainPhase: Int, totalEntries: Int, activeRequests: Int, peakActiveRequests: Int, pollAvgMs: Int, pollMaxMs: Int, pollCount: Int, pollLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, sendAvgMs: Int, sendMaxMs: Int, sendCount: Int, sendLatencyBuckets: {lt50: Int, lt200: Int, lt1000: Int, gte1000: Int}, authAvgMs: Int, authMaxMs: Int, authCount: Int}} = Stats
491 GET /stats/peers headers {authorization: Text} -> {peers: [PeerStatus], ok: Bool} = StatsPeers
492 GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text} -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, forwardDetails: [{server: Text, retries: Int, createdAt: Timestamp}], activePolls: Int, ok: Bool} = StatsKey
493 POST {limit: Int} /stats/top-recipients headers {authorization: Text} -> {recipients: [RecipientCount], ok: Bool, total: Int} = TopRecipients
494 GET /stats/forwards headers {authorization: Text} -> {servers: [{server: Text, count: Int, oldestAge: Int, totalRetries: Int}], ok: Bool, total: Int} = StatsForwards
495 GET /stats/rates headers {authorization: Text} -> {ok: Bool, scalePct: Int, globalRequest: Int, ipRate: Int, sendRate: Int, federationRate: Int, authGlobal: Int, recvRate: Int, challengeRate: Int, pairRate: Int, lookupRate: Int, clientWriteRate: Int} = StatsRates
497route MetricsApi where
498 GET /metrics headers {authorization: Text} -> {metrics: Text, ok: Bool} = Metrics
501 GET /ready -> {ready: Bool} = Ready
503route Api = AuthApi | ClientApi | FederationApi | FederationPubkeyApi | MaintenanceApi | HealthApi | StatsApi | MetricsApi | ReadyApi
505-- ============================================
507-- ============================================
508-- List/predicate combinators (filter/any/none/findFirst/...), token generation,
509-- and small text/server-name utilities used throughout the rest of the file.
510-- Everything is intentionally O(n) over flat lists; performance is sacrificed
511-- for clarity per the file header.
513filter : a b -> (b -> Bool) -> a b
514filter = \xs pred -> do
518removeWhere : a b -> (b -> Bool) -> a b
519removeWhere = \xs pred -> filter xs (\x -> not pred x)
520pluck : a b -> (b -> c) -> a c
524mapWhere : a b -> (b -> Bool) -> (b -> c) -> a c
525mapWhere = \xs pred -> pluck (filter xs pred)
526countWhere : a b -> (b -> Bool) -> Int
527countWhere = \xs pred -> count (filter xs pred)
528any : [a] -> (a -> Bool) -> Bool
529any = \xs pred -> countWhere xs pred > 0
530none : [a] -> (a -> Bool) -> Bool
531none = \xs pred -> not any xs pred
532all : [a] -> (a -> Bool) -> Bool
533all = \xs pred -> none xs (\x -> not pred x)
534findFirst : [a] -> (a -> Bool) -> Maybe a
535findFirst = \xs pred -> headOf (filter xs pred)
536when : Bool -> IO {} {} -> IO {} {}
537when = \cond action -> if cond then action else yield {}
539min = \a b -> if a < b then a else b
541max = \a b -> if a > b then a else b
542clamp : a -> a -> a -> a
543clamp = \lo hi x -> max lo (min hi x)
545abs = \x -> if x < 0 then 0 - x else x
546-- Pair each element with its 0-based index.
547withIndex : [a] -> [{i: Int, v: a}]
548withIndex = fold (\acc x -> union acc [{i: count acc, v: x}]) []
550-- First element of a list, or Nothing for an empty list.
551headOf : [a] -> Maybe a
552headOf = \items -> fold (\_ x -> Just {value: x}) (Nothing {}) (takeRelation 1 items)
554-- Cap to maxN by dropping the smallest-key entries (LRU-style). Always sorts.
555capList : [a] -> Int -> (a -> b) -> [a]
556capList = \items maxN getKey -> dropRelation (max 0 (count items - maxN)) (sortBy getKey items)
558-- Integer percentage with zero-denominator guard.
559pct : Int -> Int -> Int
560pct = \cur maxN -> if maxN > 0 then cur * 100 / maxN else 0
562-- Common ack/delete tail: pull matching messages, drop them and their forwards.
563removeForKey : (forall r. {toKey: PubkeyHex, encryptedBlob: BlobHex | r} -> Bool) -> IO {rw *forwards, rw *messages} [Message]
564removeForKey = \pred -> do
567 *messages = removeWhere msgs pred
568 *forwards = removeWhere fwds pred
569 yield (filter msgs pred)
571-- Cap pred-matching entries to maxN-1 (caller adds one more).
572capPerKey : [a] -> (a -> Bool) -> Int -> (a -> b) -> [a]
573capPerKey = \items pred maxN getKey -> union (removeWhere items pred) (capList (filter items pred) (maxN - 1) getKey)
575-- Shared empty/error shape for /poll. Only a few fields ever vary.
576emptyPollResp : Timestamp -> Bool -> Bool -> Bool -> Int -> Int -> Bool -> {events: [PollEvent], cursor: Timestamp, authorized: Bool, rateLimited: Bool, draining: Bool, generation: Int, heartbeat: Bool, retryAfterMs: Int, cursorReset: Bool, hasMore: Bool, cursorStale: Bool, loadPct: Int}
577emptyPollResp = \cursor authorized rateLimited draining gen retryAfterMs cursorWasReset -> {
586 cursorReset: cursorWasReset,
588 cursorStale: False {},
592-- Shared error shape for /messages send.
593sendErrResp : MessageStatus -> Text -> MessageIdHex -> {status: MessageStatus, messageId: MessageIdHex, retryAfterMs: Int, error: Text, loadPct: Int}
594sendErrResp = \status err messageId -> {status, messageId, retryAfterMs: 0, error: err, loadPct: 0}
596-- True when a send produced (or matched) a deliverable message.
597isSendSuccess : MessageStatus -> Bool
598isSendSuccess = \s -> case s of
599 MsDelivered {} -> True {}
600 MsFederated {} -> True {}
601 MsQueued {} -> True {}
602 MsDuplicate {} -> True {}
605-- Sequence an IO action over each item, in order.
606forEach : [a] -> (a -> IO {} {}) -> IO {} {}
607forEach = \items action -> case single (takeRelation 1 items) of
608 Nothing {} -> yield {}
609 Just {value: x} -> do
611 forEach (dropRelation 1 items) action
613-- Like forEach, but collects each action's result into a list.
614mapEach : [a] -> (a -> IO {} b) -> IO {} [b]
615mapEach = \items action -> case headOf items of
616 Nothing {} -> yield []
617 Just {value: x} -> do
619 rest <- mapEach (dropRelation 1 items) action
620 yield (union [r] rest)
622-- Hex tokens: one random hex char per recursive call.
623genTokenN : Int -> IO {random} Text
624genTokenN = \n -> if n <= 0
628 rest <- genTokenN (n - 1)
629 yield (take 1 (drop d "0123456789abcdef") ++ rest)
631-- Standard 48-char hex token used for session bearers and federation nonces.
632genToken : IO {random} Text
633genToken = genTokenN 48
635-- Canonical message identity: BLAKE3 of the encrypted blob, hex-encoded.
636-- Binding the id to the payload means a hostile peer cannot multiply-store
637-- the same ciphertext under different ids on /federation/forward — same
638-- blob always collides with the existing dedup key.
639messageIdOf : BlobHex -> MessageIdHex
640messageIdOf = \blob -> bytesToHex (hash blob)
642-- True when `s` exceeds `maxLen` characters.
643tooLong : Text -> Int -> Bool
644tooLong = \s maxLen -> count (chars s) > maxLen
646-- True when `s` is non-empty, even-length, and only lowercase hex digits.
647isValidHex : Text -> Bool
648isValidHex = \s -> (\cs -> (\n -> n > 0 && n / 2 * 2 == n && all cs (\c -> contains c "0123456789abcdef")) (count cs)) (chars s)
650-- Recursively strip every trailing "." from `s` (DNS root form).
651stripTrailingDot : Text -> Text
652stripTrailingDot = \s -> if containsSuffix "." s
653then stripTrailingDot (take (count (chars s) - 1) s)
656-- Canonical form of a hostname: lowercase, no trailing dot.
657normalizeServerName : Text -> Text
658normalizeServerName = \s -> toLower (stripTrailingDot s)
660-- This server's own hostname in canonical form (used everywhere for self-checks).
662serverHostN = normalizeServerName serverHost
663-- True when `s` refers to this server.
664isLocalServer : Text -> Bool
665isLocalServer = \s -> s == serverHostN
666-- True when `s` refers to any server other than this one.
667isRemote : Text -> Bool
668isRemote = \s -> not isLocalServer s
670-- True when `s` starts with `pre`.
671hasPrefix : Text -> Text -> Bool
672hasPrefix = \pre s -> take (count (chars pre)) s == pre
674-- True when `s` ends with `suffix`.
675containsSuffix : Text -> Text -> Bool
676containsSuffix = \suffix s -> drop (max 0 (count (chars s) - count (chars suffix))) s == suffix
678-- Block private/reserved/internal hostnames (SSRF defense). Skips obscure encoding
679-- tricks (octal, hex, percent-encoding) but covers the high-value cases.
680isBadServerName : Text -> Bool
681isBadServerName = \sn -> sn == "" || sn == "localhost" || not all (chars sn) (\c -> contains c "0123456789abcdefghijklmnopqrstuvwxyz-.:") || any [
691] (\p -> hasPrefix p sn) || any [".localhost", ".local", ".internal", ".arpa", ".onion"] (\s -> containsSuffix s sn)
693-- ============================================
695-- ============================================
696-- Bearer token extraction, session lookup, maintenance secret check, and the
697-- `gateAuth`/`gateClientWrite`/`gateLookup` combinators that every client
698-- handler funnels through (drain → global rate → per-key rate → session).
699-- Federation requests use a separate path (`withFederationAuth`) using
700-- Ed25519-signed auth verified via the peer's published public key.
702-- Plain ip match (sessions with empty ip skip the check).
703matchIp : Text -> Text -> Bool
704matchIp = \clientIp sessionIp -> sessionIp == "" || clientIp == sessionIp
706-- Pull the token out of an `Authorization: Bearer <token>` header. Returns
707-- Nothing if the prefix is missing or the token is empty.
708extractToken : Text -> Maybe Text
709extractToken = \h -> if hasPrefix "Bearer " h && drop 7 h != ""
710then Just {value: drop 7 h}
713-- True when the header carries a Bearer token equal to the runtime maintenance
714-- secret. Plain `==` comparison; not constant-time (see file header).
715verifyMaintenanceAuth : Text -> IO {r *runtimeMaintenanceSecret} Bool
716verifyMaintenanceAuth = \authorization -> do
717 secret <- *runtimeMaintenanceSecret
718 yield (secret != "" && extractToken authorization == Just {value: secret})
720-- Branch on maintenance-secret auth: run `onAuth` when the bearer matches,
721-- `onUnauth` otherwise.
722withMaintAuth : Text -> IO {| effs} a -> IO {| effs} a -> IO {r *runtimeMaintenanceSecret | effs} a
723withMaintAuth = \authorization onUnauth onAuth -> do
724 authorized <- verifyMaintenanceAuth authorization
725 if authorized then onAuth else onUnauth
727-- Common gate for handlers: optional drain check, global rate, session auth.
728-- mkErr is called with one of: "draining", "rate_limited", "unauthorized".
729gateAuth : Bool -> Text -> Text -> Timestamp -> (resp -> ret) -> (Text -> resp) -> (PubkeyHex -> IO {} ret) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} ret
730gateAuth = \checkDrain auth ip t respond mkErr onAuth -> do
732 g <- checkGlobalRate t
733 let err = \e -> yield (respond (mkErr e))
734 if checkDrain && dp >= 2
736 else if g then err "rate_limited" else withSessionAuth auth t ip (err "unauthorized") onAuth
738-- gateAuth + per-key rate gate. `rateResp` is the response value when the per-key gate trips.
739gateClientWrite : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
740gateClientWrite = \auth ip t respond mkErr rateResp onAuth ->
741 gateAuth (True {}) auth ip t respond mkErr (\pk -> do
743 rates <- *clientWriteRates
744 let r = checkRate rates pk maxClientWriteRatePerKey clientWriteRateWindowMs t
745 -- TODO: make this no be a `replace`
746 *clientWriteRates = r.entries
748 if lim then yield (respond rateResp) else onAuth pk)
750-- Variant of `gateClientWrite` that skips the drain check (lookups still work
751-- while draining) and uses the per-key lookup-rate window instead.
752gateLookup : Text -> Text -> Int<Ms> -> (a -> b) -> (Text -> a) -> a -> (Text -> IO {} b) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions} b
753gateLookup = \auth ip t respond mkErr rateResp onAuth ->
754 gateAuth (False {}) auth ip t respond mkErr (\pk -> do
756 rates <- *lookupRates
757 let r = checkRate rates pk maxLookupRatePerKey lookupRateWindowMs t
758 -- TODO: make this no be a `replace`
759 *lookupRates = r.entries
761 if lim then yield (respond rateResp) else onAuth pk)
763-- Resolve a Bearer token to its session pubkey if the session is unexpired and
764-- the client IP matches. Runs `onAuth pubkey` on success, `onUnauth` otherwise.
765withSessionAuth : Text -> Int<Ms> -> Text -> IO {| e} a -> (PubkeyHex -> IO {| e} a) -> IO {r *sessions | e} a
766withSessionAuth = \authorization t clientIp onUnauth onAuth -> do
768 case extractToken authorization of
769 Nothing {} -> onUnauth
770 Just {value: token} -> case findFirst ses (\s -> s.expiresAt > t && s.token == token && matchIp clientIp s.ip) of
771 Just {value: s} -> onAuth s.pubkey
774-- Build the `Signed <hexsig>:<nonce>` header for an outbound federation call.
775-- Signs the canonical "scope|timestamp|fromServer|nonce" string with our
776-- federation private key (the peer fetches our public key via /federation/pubkey
777-- to verify). Returns "" if no keypair has been generated yet.
778generateFederationAuth : Text -> a -> IO {r *federationPrivateKeyHex, random} Text
779generateFederationAuth = \scope requestTimestamp -> do
780 hex <- *federationPrivateKeyHex
781 let privKey = if hex == "" then Nothing {} else bytesFromHex hex
783 Just {value: pk} -> do
784 nonce <- genTokenN 48
785 yield ("Signed " ++ bytesToHex (sign pk (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ serverHostN ++ "|" ++ nonce))) ++ ":" ++ nonce)
788-- Verify a federation request: Ed25519 signature over scope|ts|fromServer|nonce, key fetched from peer.
789verifyFederationAuth : Text -> Text -> Text -> a -> IO {network} {} Bool
790verifyFederationAuth = \authorization fromServer scope requestTimestamp -> do
791 if hasPrefix "Signed " authorization then do
792 let payload = drop 7 authorization
793 if payload == "" || tooLong payload (signatureLen + 25) then yield False {}
795 -- Split payload on first ':' into {sig, nonce}.
796 let cs = chars payload
797 let sepIdx = case findFirst (withIndex cs) (\p -> p.v == ":") of
798 Just {value: p} -> p.i
800 let sig = take sepIdx payload
801 let nonce = drop (sepIdx + 1) payload
802 result <- fetch (federationUrl fromServer) (FederationPubkey {})
804 Ok {value: r} -> case bytesFromHex r.pubkey of
805 Just {value: pk} -> case bytesFromHex sig of
806 Just {value: sigBytes} -> yield (r.ok && r.pubkey != "" && verify pk (textToBytes (scope ++ "|" ++ show requestTimestamp ++ "|" ++ fromServer ++ (if nonce == "" then "" else "|" ++ nonce))) sigBytes)
812-- Combined gate for inbound federation calls. Order of checks (each short-
813-- circuits with an error code in the response):
814-- 1. peer not block-listed; 2. peer in allow-list (if any); 3. caller-supplied
815-- `precondition` holds; 4. clock skew within `clockSkewTolerance`;
816-- 5. Ed25519 signature verifies; 6. per-peer federation rate limit.
817withFederationAuth = \authorization t respond fromServer scope requestTimestamp precondition action -> do
818 blocked <- *blockedPeers
819 allowed <- *allowedPeers
820 let err = \e r -> yield (respond {ok: False {}, error: e, retryAfterMs: r})
821 let preErr = if elem fromServer blocked
823 else if count allowed > 0 && not elem fromServer allowed
825 else if not precondition
826 then "invalid_request"
827 else if requestTimestamp > 0 && abs (requestTimestamp - t) > clockSkewTolerance
833 verified <- verifyFederationAuth authorization fromServer scope requestTimestamp
835 then err "unauthorized" 0
837 limited <- checkServerRateGate fromServer t
838 if limited then err "rate_limited" federationRateWindowMs else action
840-- ============================================
841-- Rate limiting (plain count-per-window, no sliding interpolation)
842-- ============================================
843-- Two flavors of fixed-window counters:
844-- * Per-key (`checkRate`) — keyed by pubkey, ip, server, or pair string;
845-- entries live in a `RateEntry` list ref capped at 5000 with LRU eviction.
846-- * Global (`checkGlobalRate`/`checkAuthGlobalRate`) — single counter+windowStart
847-- pair, used for the request-wide and auth-wide ceilings.
848-- `authPreCheck` composes drain + global + ip + auth-global into a one-shot
849-- gate for unauthenticated handlers (challenge/verify).
851-- Returns {limited, entries}. Increments count when the key fits within the window.
852checkRate : [RateEntry] -> Text -> Int -> Int<Ms> -> Timestamp -> {limited: Bool, entries: [RateEntry]}
853checkRate = \entries key maxRate windowMs t -> (\cur -> if cur.count > maxRate then {limited: True {}, entries} else {limited: False {}, entries: capList (union (removeWhere entries (\x -> x.key == key)) [{key, cur.windowStart, cur.count}]) 5000 (\r -> r.windowStart)}) (case findFirst entries (\r -> r.key == key && t - r.windowStart < windowMs) of
854 Just {value: r} -> {r.windowStart, count: r.count + 1}
855 _ -> {windowStart: t, count: 1})
857-- Single-counter global rate gate. Resets the counter when the window expires;
858-- returns True when the limit has been hit (caller responds rate_limited).
859checkGlobalRate : Timestamp -> IO {rw *globalRateCount, rw *globalRateWindowStart} Bool
860checkGlobalRate = \t -> atomic (do
861 ws <- *globalRateWindowStart
862 c <- *globalRateCount
863 if t - ws >= globalRateWindowMs
865 *globalRateWindowStart = t
868 else if c >= maxGlobalRequestRate
871 *globalRateCount = c + 1
874-- Same shape as `checkGlobalRate` but charged only by the auth endpoints
875-- (challenge + verify). Stops a single attacker from exhausting the global
876-- counter via auth churn.
877checkAuthGlobalRate = \t -> atomic (do
878 ws <- *authGlobalRateWindowStart
879 c <- *authGlobalRateCount
880 if t - ws >= authGlobalRateWindowMs
882 *authGlobalRateWindowStart = t
883 *authGlobalRateCount = 1
885 else if c >= maxAuthGlobalRate
888 *authGlobalRateCount = c + 1
891-- Per-source-IP rate gate (skipped when the IP is unknown — see `requireClientIp`).
892checkIpRateGate : Text -> Int<Ms> -> IO {rw *ipRates} Bool
893checkIpRateGate = \ip t -> if ip == ""
897 let r = checkRate rates ip maxIpRatePerIp ipRateWindowMs t
901-- Per-peer federation rate gate; charged inside `withFederationAuth`.
902checkServerRateGate : Text -> Int -> IO {rw *federationRates} Bool
903checkServerRateGate = \server t -> atomic (do
904 rates <- *federationRates
905 let r = checkRate rates server maxFederationRatePerServer federationRateWindowMs t
906 *federationRates = r.entries
909-- Pre-auth gate: drain + global + ip + auth-global. Returns "" when ok, else an error code.
910authPreCheck : Text -> Int<Ms> -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates} Text
911authPreCheck = \clientIp t -> do
913 g <- checkGlobalRate t
914 i <- checkIpRateGate clientIp t
915 a <- checkAuthGlobalRate t
919 then "global_rate_limited"
920 else if i then "ip_rate_limited" else if a then "auth_rate_limited" else "")
922-- ============================================
923-- Federation failure tracking (simple, no circuit breaker)
924-- ============================================
925-- Linear backoff per peer: each consecutive failure multiplies the delay by
926-- `retryBackoffBase` (capped at `retryBackoffMax`). `shouldRetryServer` is the
927-- single read-side gate; `recordFailure`/`clearFailure` are the write side.
928-- There is no half-open state — a single success clears all accumulated failures.
930-- Linear backoff: `retryBackoffBase * failures`, capped at `retryBackoffMax`.
931retryBackoff : Int -> Int<Ms>
932retryBackoff = \failures -> min retryBackoffMax (retryBackoffBase * max 1 failures)
934-- True when enough time has passed since the last failure for `server` to be
935-- worth retrying (or no failure recorded at all).
936shouldRetryServer : Text -> Timestamp -> [ForwardFailure] -> Bool
937shouldRetryServer = \server t failures -> none failures (\f -> f.server == server && t - f.failedAt < retryBackoff f.failures)
939-- Bump the failure counter for `server`. Preserves `firstFailedAt` across
940-- subsequent failures so health views can show the outage age.
941recordFailure : Text -> IO {rw *forwardFailures, clock} {}
942recordFailure = \server -> do
945 failures <- *forwardFailures
946 let same = \f -> f.server == server
947 let prev = case findFirst failures same of
948 Just {value: f} -> {failures: f.failures + 1, f.firstFailedAt}
949 _ -> {failures: 1, firstFailedAt: t}
950 *forwardFailures = union (removeWhere failures same) [
951 {server, failedAt: t, prev.failures, prev.firstFailedAt}
954-- Drop the failure record for `server` (called after any successful peer call).
955-- Single-success recovery — there is no half-open state.
956clearFailure : Text -> IO {rw *forwardFailures} {}
957clearFailure = \server -> atomic (do
958 forwardFailures <- *forwardFailures
959 *forwardFailures = filter forwardFailures (\f -> f.server != server))
961-- ============================================
962-- Common utility queries
963-- ============================================
964-- Pure helpers shared by send/poll/federation paths: monotonic sequence
965-- generation, presence lookup, and duplicate detection across
966-- (messages, deliveredIds).
968-- Strictly-monotonic next sequence timestamp: max of wall clock and prior+1.
969-- Guarantees /poll cursors order events even when many arrive in the same ms.
970nextSeqTs : Timestamp -> Timestamp -> Timestamp
971nextSeqTs = \t seqCounter -> max t (seqCounter + 1)
973-- Drop duplicates and cap to `maxMessageIds` (defense against bloated requests).
974dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
975dedupMessageIds = \ids -> takeRelation maxMessageIds ids
977-- All unexpired presence rows for a given pubkey (may include both local and
978-- remote entries — the recipient may live on multiple servers simultaneously).
979presenceForKey : [PresenceEntry] -> PubkeyHex -> Timestamp -> [PresenceEntry]
980presenceForKey = \presence pubkey t -> filter presence (\p -> p.pubkey == pubkey && p.expiresAt > t)
982-- Resolve a key's server for client lookup, masking remote identities behind "remote".
983maskedServer : [PresenceEntry] -> PubkeyHex -> Timestamp -> Text
984maskedServer = \pres pubkey t -> (\here -> if any here (\p -> isLocalServer p.server) then serverHostN else if count here > 0 then "remote" else "") (presenceForKey pres pubkey t)
986-- Replay-protection check. A message is a duplicate if either:
987-- * a live mailbox row exists with the same (encryptedBlob, toKey) -- since
988-- id := messageIdOf encryptedBlob, same blob ⇒ same id; or
989-- * a delivered-id tombstone matches (messageId, toKey).
990-- The blob comparison avoids hashing every live row on each check; the
991-- tombstone log only stores the id, so we compare it directly.
992isDuplicateMessage : BlobHex -> PubkeyHex -> [Message] -> [DeliveredEntry] -> Bool
993isDuplicateMessage = \blob toKey msgs dids -> any msgs (\m -> m.toKey == toKey && m.encryptedBlob == blob) || (\mid -> any dids (\d -> d.toKey == toKey && d.messageId == mid)) (messageIdOf blob)
995-- Project items through `getServer`, drop our own hostname, and dedupe.
996-- Used to derive peer recipient lists from forwards/presence/etc.
997collectPeers : [a] -> (a -> Text) -> [Text]
998collectPeers = \items getServer -> filter (pluck items getServer) isRemote
1000-- All distinct remote hostnames currently appearing in `presence` (the gossip
1001-- broadcast set we know about, less ourselves).
1002knownPeerServersFromPresence : [PresenceEntry] -> [Text]
1003knownPeerServersFromPresence = \presence -> collectPeers presence (\p -> p.server)
1005-- ============================================
1007-- ============================================
1008-- Single sweep that drops expired entries from every state ref in one atomic
1009-- block. Returns the locally-expired presence rows whose pubkey has no other
1010-- live local session, so the caller can broadcast offline gossip for them.
1012-- One sweep over every state ref: drops expired rows in a single atomic block.
1013-- Returns the locally-expired presence rows whose pubkey has no other live
1014-- local session, so the caller can broadcast offline gossip for them.
1015pruneExpired : Int<Ms> -> IO {rw *activePolls, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *messages, rw *presence, rw *sessions} [{expiresAt: Timestamp, pubkey: PubkeyHex, server: AnyServerName}]
1016pruneExpired = \t -> atomic do
1018 *challenges = filter ch (\x -> x.expiresAt > t)
1020 *sessions = filter ses (\x -> x.expiresAt > t)
1022 *messages = filter msgs (\m -> t - m.receivedAt < messageMaxAge)
1024 *forwards = filter fwds (\f -> t - f.createdAt < messageMaxAge)
1025 dids <- *deliveredIds
1026 *deliveredIds = filter dids (\d -> t - d.deliveredAt < messageMaxAge)
1027 ff <- *forwardFailures
1028 *forwardFailures = filter ff (\f -> t - f.failedAt < 86400000)
1030 *activePolls = filter aps (\p -> t - p.startedAt < pollTimeoutMs * 3)
1031 -- presence: remember locally-expired keys for offline gossip
1033 *presence = filter pres (\p -> p.expiresAt > t)
1034 let stillLive = \pk -> any pres (\k -> isLocalServer k.server && k.pubkey == pk && k.expiresAt > t)
1035 yield (filter pres (\p -> p.expiresAt <= t && isLocalServer p.server && not (stillLive p.pubkey)))
1037-- ============================================
1039-- ============================================
1040-- Construction and outbound dispatch of presence gossip events. Failures from
1041-- a peer record into `forwardFailures` (shared with forward retries) so both
1042-- subsystems back off together.
1044-- Build a gossip event with the originating pubkey's status.
1045mkGossip : GossipEventType -> PubkeyHex -> GossipEvent
1046mkGossip = \etype pk -> {eventType: etype, pubkey: pk}
1048-- Fan out one bundle of gossip events to each peer, skipping peers in backoff.
1049-- Records or clears federation-failure counters per call.
1050sendBatchedGossip : [Text] -> [GossipEvent] -> IO {} {}
1051sendBatchedGossip = \peers events -> forEach peers (\server -> do
1053 failures <- *forwardFailures
1054 when (shouldRetryServer server t failures) (do
1055 auth <- generateFederationAuth "gossip" t
1056 result <- fetch (federationUrl server) (RecvGossip {
1057 authorization: auth,
1059 fromServer: serverHostN,
1060 requestTimestamp: t,
1064 Err _ -> recordFailure server
1065 Ok _ -> clearFailure server))
1067-- Tell known peers that the listed local pubkeys just went offline. No-op when
1068-- federation is disabled or the list is empty.
1069broadcastOfflineGossip : [{pubkey: PubkeyHex | a}] -> IO {r *presence} {}
1070broadcastOfflineGossip = \expiredLocal -> do
1071 presence <- *presence
1072 when (federationEnabled && count expiredLocal > 0) (sendBatchedGossip (knownPeerServersFromPresence presence) (pluck expiredLocal (\p -> mkGossip (GossipOffline {}) p.pubkey)))
1074-- ============================================
1076-- ============================================
1077-- `retryPendingForwards` walks queued cross-server forwards in batches per
1078-- destination, increments retry counts, and gives up after `maxForwardRetries`.
1079-- `forwardQueuedToServer` is the catch-up path triggered when a remote key
1080-- comes online: any messages already buffered for that pubkey are enqueued
1081-- as forwards to the newly-known server.
1083-- Walk pending forwards grouped by destination, retry each (up to maxBatchSize
1084-- per peer) using `/federation/forward`. Drops successful entries; for failures
1085-- bumps the retry count and gives up after `maxForwardRetries`.
1086retryPendingForwards : IO {r *blockedPeers, r *forwardFailures, r *forwards, clock} {}
1087retryPendingForwards = do
1090 failures <- *forwardFailures
1091 blocked <- *blockedPeers
1092 forEach (filter (collectPeers fwds (\f -> f.toServer)) (\s -> shouldRetryServer s t failures && not elem s blocked)) (\server -> forEach (takeRelation maxBatchSize (filter fwds (\f -> f.toServer == server))) (\f -> do
1094 let messageId = messageIdOf f.encryptedBlob
1095 auth <- generateFederationAuth ("forward:" ++ messageId) t2
1096 result <- fetch (federationUrl server) (ForwardMessage {
1100 fromServer: serverHostN,
1101 requestTimestamp: t2,
1102 authorization: auth,
1105 success <- case result of
1107 recordFailure server
1110 when r.ok (clearFailure server)
1112 let isMatch = \x -> x.encryptedBlob == f.encryptedBlob && x.toServer == server
1113 let isDone = \x -> isMatch x && (success || x.retries + 1 >= maxForwardRetries)
1116 *forwards = pluck (removeWhere cur isDone) (\x -> if isMatch x
1117 then {x.toServer, x.createdAt, x.toKey, x.encryptedBlob, retries: x.retries + 1}
1120-- Forward queued messages to a server when a remote key comes online.
1121forwardQueuedToServer : PubkeyHex -> ServerName -> Int<Ms> -> IO {rw *forwards, r *messages} {}
1122forwardQueuedToServer = \pubkey server t -> atomic (do
1125 let candidates = filter msgs (\m -> m.toKey == pubkey && none fwds (\f -> f.encryptedBlob == m.encryptedBlob && f.toServer == server))
1126 let newFwds = pluck (takeRelation maxForwardsPerServer candidates) (\m -> {
1133 *forwards = union fwds newFwds)
1135-- ============================================
1137-- ============================================
1138-- Composite tick run from `backgroundPrune` (and on demand via /maintenance/prune):
1139-- expire stale rows, then opportunistically retry pending forwards and refresh
1140-- local presence so peers don't see us go offline mid-session.
1142-- One composite background tick: prune expired state, gossip offline for
1143-- locally-expired keys, retry pending forwards, and (occasionally) refresh
1144-- local presence so peers don't see us drop off.
1145pruneAndRetry = \t -> do
1147 expiredLocal <- pruneExpired t
1148 fork (broadcastOfflineGossip expiredLocal)
1149 when federationEnabled (do
1150 fork retryPendingForwards
1151 lastRefresh <- *lastPresenceRefresh
1152 when (t - lastRefresh >= presenceRefreshInterval) (do
1153 *lastPresenceRefresh = t
1154 refreshLocalPresence t))
1156-- Extend the expiry of local presence rows whose backing session is still live
1157-- but the presence entry is past half its window. Re-broadcasts an `online`
1158-- event for the refreshed entries so peers don't time them out.
1159refreshLocalPresence : Int<Ms> -> IO {rw *presence, r *sessions} {}
1160refreshLocalPresence = \t -> do
1161 refreshState <- atomic (do
1162 sessions <- *sessions
1163 presence <- *presence
1164 let sessionKeys = map (\s -> s.pubkey) sessions
1165 let needsRefresh = \p -> isLocalServer p.server && elem p.pubkey sessionKeys && p.expiresAt - t < sessionExpiry / 2
1166 let kept = filter presence (\p -> not needsRefresh p)
1167 let refreshed = map (\p -> {p.pubkey, p.server, expiresAt: t + sessionExpiry}) (filter presence needsRefresh)
1168 let updated = union kept refreshed
1170 yield {refreshed, peers: knownPeerServersFromPresence updated})
1171 when (count refreshState.refreshed > 0) (fork (sendBatchedGossip refreshState.peers (pluck refreshState.refreshed (\p -> mkGossip (GossipOnline {}) p.pubkey))))
1173-- ============================================
1174-- Handler: challenge / verify / revoke / logout
1175-- ============================================
1176-- Identity is a public key; there are no accounts. The flow is:
1177-- 1. Client requests a fresh nonce (`handleChallenge`).
1178-- 2. Client signs the nonce with its private key and submits the signature
1179-- together with the original challenge (`handleVerifyAuth`).
1180-- 3. Server returns a Bearer token bound to (pubkey, ip, expiry); the new
1181-- session also installs a local presence row and gossips an `online`
1182-- event to known peers.
1183-- `handleRevoke` lets a session revoke its peers (same pubkey, different
1184-- token); `handleLogout` invalidates the caller's own token and, if no
1185-- sibling sessions remain, gossips an `offline` event.
1187-- POST /auth/challenge: issue a fresh signed-nonce challenge for `pubkey`.
1188-- Capped per-key (maxChallengesPerKey) and rate-limited; the IP is recorded
1189-- so /auth/verify can require the same client.
1190handleChallenge : PubkeyHex -> Text -> ({challenge: Text, error: Text, expiresAt: Int, ok: Bool, retryAfterMs: Int} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challengeRates, rw *challenges, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, clock, random} a
1191handleChallenge = \pubkey clientIp respond -> do
1193 preErr <- if requireClientIp && clientIp == ""
1194 then yield "client_ip_required"
1195 else authPreCheck clientIp t
1197 then yield (respond {challenge: "", ok: False {}, expiresAt: 0, retryAfterMs: 0, error: preErr})
1199 challenge <- genToken
1200 result <- atomic (do
1202 let cappedActive = capPerKey (filter ch (\c -> c.expiresAt > t)) (\c -> c.pubkey == pubkey) maxChallengesPerKey (\c -> c.expiresAt)
1203 if count cappedActive >= maxChallenges
1204 then yield {ok: False {}, error: "capacity"}
1206 chRates <- *challengeRates
1207 let chR = checkRate chRates pubkey maxChallengeRatePerKey challengeRateWindowMs t
1208 *challengeRates = chR.entries
1210 then yield {ok: False {}, error: "rate_limited"}
1212 *challenges = union cappedActive [
1213 {pubkey, challenge, expiresAt: t + challengeExpiry, ip: clientIp}
1215 yield {ok: True {}, error: ""})
1217 challenge: if result.ok then challenge else "",
1219 expiresAt: if result.ok then t + challengeExpiry else 0,
1224-- POST /auth/verify: validate `signature` over `challenge` for `pubkey`.
1225-- On success, mints a session token, records local presence (`pubkey` is now
1226-- live on this server), and gossips an `online` event to known peers. If
1227-- `revokeOthers` is set, all other sessions for the same pubkey are dropped.
1228handleVerifyAuth : PubkeyHex -> Text -> Text -> Bool -> Text -> ({error: Text, expiresAt: Int, token: Text} -> a) -> IO {rw *authGlobalRateCount, rw *authGlobalRateWindowStart, rw *challenges, r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, rw *ipRates, rw *presence, rw *sessions, clock, random} a
1229handleVerifyAuth = \pubkey challenge signature revokeOthers clientIp respond -> do
1231 preErr <- authPreCheck clientIp t
1233 then yield (respond {token: "", expiresAt: 0, error: preErr})
1235 let expiry = t + sessionExpiry
1237 result <- atomic (do
1240 let same = \c -> c.pubkey == pubkey && c.challenge == challenge
1241 let sigValid = case bytesFromHex pubkey of
1242 Just {value: pk} -> case bytesFromHex signature of
1243 Just {value: sigBytes} -> verify pk (textToBytes challenge) sigBytes
1246 let valid = dp < 1 && any ch (\c -> same c && c.expiresAt > t && matchIp clientIp c.ip) && sigValid
1248 then yield {valid: False {}, peers: []}
1250 *challenges = removeWhere ch same
1252 let mine = \s -> s.pubkey == pubkey
1253 let pruned = if revokeOthers
1254 then removeWhere ses mine
1255 else capPerKey ses mine maxSessionsPerKey (\s -> s.expiresAt)
1256 let newSession = {token, pubkey, expiresAt: expiry, ip: clientIp, createdAt: t}
1257 *sessions = capList (union pruned [newSession]) maxSessions (\s -> s.expiresAt)
1259 let prunedPres = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1260 let newPres = union prunedPres [{pubkey, server: serverHostN, expiresAt: expiry}]
1262 yield {valid: True {}, peers: knownPeerServersFromPresence newPres})
1264 then yield (respond {token: "", expiresAt: 0, error: "invalid"})
1266 when federationEnabled (fork (sendBatchedGossip result.peers [
1267 mkGossip (GossipOnline {}) pubkey
1269 yield (respond {token, expiresAt: expiry, error: ""})
1271-- POST /auth/revoke: caller (any session for the pubkey) revokes a sibling
1272-- session by token. Returns ok=False if no such sibling existed.
1273handleRevoke : Text -> Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1274handleRevoke = \targetToken authorization clientIp respond -> do
1276 gateAuth (False {}) authorization clientIp t respond (\_ -> {ok: False {}, authorized: False {}}) (\caller -> do
1277 result <- atomic (do
1279 let pred = \s -> s.pubkey == caller && s.token == targetToken
1280 *sessions = removeWhere ses pred
1281 yield (any ses pred))
1282 yield (respond {ok: result, authorized: True {}}))
1284-- POST /auth/logout: drop the caller's own session. If this was the last
1285-- session for the pubkey, also clear local presence and gossip offline.
1286handleLogout : Text -> Text -> ({authorized: Bool, ok: Bool} -> a) -> IO {rw *presence, rw *sessions, clock} a
1287handleLogout = \authorization clientIp respond -> do
1289 let myToken = case extractToken authorization of
1290 Just {value: tk} -> tk
1292 removed <- atomic withSessionAuth authorization t clientIp (yield {
1298 *sessions = removeWhere ses (\s -> s.token == myToken)
1300 peers <- if any after (\s -> s.pubkey == pubkey)
1304 *presence = removeWhere pres (\p -> p.pubkey == pubkey && isLocalServer p.server)
1305 yield (knownPeerServersFromPresence pres)
1306 yield {ok: True {}, pubkey, peers})
1307 when (removed.ok && federationEnabled) (fork (sendBatchedGossip removed.peers [
1308 mkGossip (GossipOffline {}) removed.pubkey
1310 yield (respond {removed.ok, authorized: removed.ok})
1312-- ============================================
1314-- ============================================
1315-- Long-poll. The client supplies a cursor (last-seen `receivedAt`) and the
1316-- server's last-known `generation` (server identity for cursor validity). We:
1317-- * If the generation mismatches, reset the cursor and signal `cursorReset`.
1318-- * Ack-and-drop everything <= the cursor (treating poll progress as ack).
1319-- * Wait on `pollWakeup` (bumped by send/heartbeat) and return new events
1320-- plus piggybacked sender-side expirations/deliveries.
1321-- Per-pubkey concurrency is capped via `activePolls` to bound long-poll holds.
1323-- POST /poll: top-level dispatch. Validates drain/global gates, detects a
1324-- generation mismatch (server restart) to force a cursor reset, then hands
1325-- off to `pollOnce` for the actual long-poll body.
1326handlePoll : 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
1327handlePoll = \cursor generation authorization clientIp respond -> do
1329 serverGen <- *serverGeneration
1331 g <- checkGlobalRate t
1332 let cursorWasReset = generation != serverGen && cursor > 0
1333 let er = \auth rate drain retryRate -> yield (respond (emptyPollResp 0 auth rate drain serverGen retryRate cursorWasReset))
1335 then er (True {}) (False {}) (True {}) 0
1337 then er (False {}) (True {}) (False {}) globalRateWindowMs
1338 else pollOnce authorization (if cursorWasReset then 0 else cursor) cursorWasReset clientIp serverGen respond t
1340-- Body of a single poll. Treats the cursor advance as an implicit ack
1341-- (`ackPolledMessages`), acquires a per-pubkey poll slot, runs `doPoll` to
1342-- collect events (or block until wakeup), then assembles the response.
1343pollOnce : 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
1344pollOnce = \authorization cursor cursorWasReset clientIp serverGen respond t -> do
1345 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do
1346 let clampedCursor = clamp 0 t cursor
1347 when (clampedCursor > 0) (ackPolledMessages pubkey clampedCursor)
1348 startWakeup <- *pollWakeup
1349 yield (Just {value: {pubkey, clampedCursor, startWakeup}}))
1351 Nothing {} -> yield (respond (emptyPollResp 0 (False {}) (False {}) (False {}) serverGen 0 cursorWasReset))
1352 Just {value: s} -> do
1353 acquired <- atomic (do
1355 if countWhere aps (\p -> p.pubkey == s.pubkey) >= maxPollsPerKey
1356 then yield (False {})
1358 *activePolls = union aps [{s.pubkey, startedAt: t}]
1361 then yield (respond (emptyPollResp cursor (True {}) (True {}) (False {}) serverGen pollTimeoutMs cursorWasReset))
1363 pollResult <- doPoll s t
1364 atomic (do aps <- *activePolls; *activePolls = filter aps (\p -> p.pubkey != s.pubkey))
1368 authorized: True {},
1369 rateLimited: False {},
1371 generation: serverGen,
1372 heartbeat: count pollResult.events == 0,
1374 cursorReset: cursorWasReset,
1376 cursorStale: False {},
1380-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`
1381-- (bumped by send/heartbeat) and re-collect once at the new wall time.
1382doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}], hasMore: Bool}
1384 let collectAt = \tnow -> atomic (do
1386 yield (collectPollEvents msgs s.pubkey s.clampedCursor tnow))
1387 initial <- collectAt t
1388 if count initial.events > 0
1391 atomic (do wakeup <- *pollWakeup; if wakeup > s.startWakeup then yield {} else retry)
1395-- Pull this pubkey's unexpired messages newer than `cursor`, sort by
1396-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.
1397collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}
1398collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {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))
1400-- Treat the supplied cursor as an ack: drop all messages with
1401-- receivedAt <= cursor, plus their pending forwards.
1402ackPolledMessages : PubkeyHex -> Int -> IO {rw *forwards, rw *messages} {}
1403ackPolledMessages = \pubkey cursor -> do
1405 let matches = \m -> m.toKey == pubkey && m.receivedAt <= cursor
1406 let toAck = filter msgs matches
1407 let blobs = pluck toAck (\m -> m.encryptedBlob)
1408 *messages = removeWhere msgs matches
1410 *forwards = filter fwds (\f -> not (f.toKey == pubkey && elem f.encryptedBlob blobs))
1412-- ============================================
1414-- ============================================
1415-- Accept a message for a recipient pubkey and decide its placement:
1416-- * `MsDelivered` -- recipient has a live local session
1417-- * `MsFederated` -- recipient lives on a remote peer; queued as a forward
1418-- * `MsQueued` -- no presence yet; held until presence appears
1419-- * `MsRejected` / `MsRateLimited` / `MsDuplicate` / `MsDraining` -- refused
1420-- Send is gated by drain phase, global rate, and the per-sender / per-pair /
1421-- per-recipient rate limits charged together inside `sendMessageCore`.
1422-- Federated forwards are kicked off in a background fork after the atomic.
1424-- POST /messages: send one ciphertext to `to`. Rejects on draining/rate
1425-- limit; otherwise calls `sendMessageCore` inside an atomic block and forks
1426-- the federation forward (if any) once the local commit lands.
1427handleSendMessage : 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
1428handleSendMessage = \to encryptedBlob authorization clientIp respond -> do
1431 g <- checkGlobalRate t
1432 let effMsgId = messageIdOf encryptedBlob
1433 let err = \s e -> yield (respond (sendErrResp s e effMsgId))
1435 then err (MsDraining {}) "draining"
1437 then err (MsRateLimited {}) "rate_limited"
1439 result <- atomic (do
1440 withSessionAuth authorization t clientIp (yield {status: MsUnauthorized {}, remoteServers: []}) (\senderKey -> sendMessageCore senderKey to encryptedBlob t))
1441 when federationEnabled (fork (forkForwardsTo to encryptedBlob result.remoteServers))
1442 let ok = isSendSuccess result.status
1445 messageId: effMsgId,
1447 error: if ok then "" else "rejected",
1451-- Atomic core of /messages and /messages/batch. Charges send/pair/recv rate
1452-- counters together (so any limit denies the message), enforces caps and
1453-- self-send/duplicate rejections, allocates a monotonic receivedAt via
1454-- `seqCounter`, appends the message, and pre-stages forwards for any remote
1455-- presence rows that already exist for `to`. Returns the chosen status and
1456-- the list of remote servers the caller should kick forwards to.
1457sendMessageCore : 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}
1458sendMessageCore = \senderKey to encryptedBlob t -> do
1460 dids <- *deliveredIds
1461 sRates <- *sendRates
1462 let sR = checkRate sRates senderKey maxSendRatePerKey sendRateWindowMs t
1463 *sendRates = sR.entries
1464 let srLim = sR.limited
1465 pRates <- *pairRates
1466 let pR = checkRate pRates (senderKey ++ "|" ++ to) maxPairRatePerPair pairRateWindowMs t
1467 *pairRates = pR.entries
1468 let prLim = pR.limited
1469 rRates <- *recvRates
1470 let rR = checkRate rRates to maxRecvRatePerKey recvRateWindowMs t
1471 *recvRates = rR.entries
1472 let rrLim = rR.limited
1473 let reject = \s -> yield {status: s, remoteServers: []}
1475 then reject (MsRejected {})
1476 else if isDuplicateMessage encryptedBlob to msgs dids
1477 then reject (MsDuplicate {})
1478 else if srLim || prLim || rrLim
1479 then reject (MsRateLimited {})
1483 let seqTs = nextSeqTs t seq
1485 *messages = union msgs [{toKey: to, encryptedBlob, receivedAt: seqTs}]
1487 let toPresence = presenceForKey pres to t
1488 let remoteServers = collectPeers toPresence (\p -> p.server)
1489 forwardsEnqueued <- if federationEnabled
1492 let newFwds = mapWhere remoteServers (\s -> none fwds (\f -> f.encryptedBlob == encryptedBlob && f.toServer == s)) (\s -> {
1499 *forwards = union fwds newFwds
1500 yield (count newFwds > 0)
1501 else yield (False {})
1503 status: if any toPresence (\p -> isLocalServer p.server)
1505 else if forwardsEnqueued then MsFederated {} else MsQueued {},
1509-- Fire one /federation/forward call per pre-staged server. On success the
1510-- corresponding row in `forwards` is pruned, and if no other forwards remain
1511-- for this blob AND the recipient has no local presence, the backing
1512-- `*messages` row is dropped too: the origin's job is done and the message
1513-- now lives on the receiving peer's mailbox. On failure the forwards row
1514-- stays so `retryPendingForwards` can re-attempt with backoff.
1515forkForwardsTo : PubkeyHex -> BlobHex -> [Text] -> IO {} {}
1516forkForwardsTo = \to encryptedBlob servers -> forEach servers (\server -> do
1517 let effMsgId = messageIdOf encryptedBlob
1519 let same = \x -> x.encryptedBlob == encryptedBlob && x.toServer == server
1520 case findFirst fwds same of
1521 Nothing {} -> yield {}
1522 Just {value: f} -> do
1524 auth <- generateFederationAuth ("forward:" ++ effMsgId) t
1525 result <- fetch (federationUrl server) (ForwardMessage {
1528 messageId: effMsgId,
1529 fromServer: serverHostN,
1530 requestTimestamp: t,
1531 authorization: auth,
1535 Err _ -> recordFailure server
1536 Ok {value: r} -> when r.ok (do
1540 let remaining = filter fwds2 (\x -> not same x)
1541 *forwards = remaining
1543 let stillLocal = any pres (\p -> p.pubkey == to && isLocalServer p.server && p.expiresAt > t)
1544 let othersPending = any remaining (\x -> x.encryptedBlob == encryptedBlob)
1545 when (not stillLocal && not othersPending) (do
1547 *messages = filter msgs (\m -> m.encryptedBlob != encryptedBlob)))))
1549-- ============================================
1551-- ============================================
1552-- Presence resolution for clients deciding where to send. Remote presence is
1553-- masked to the literal string "remote" rather than leaking the peer hostname
1554-- (the routing decision is the server's, not the sender's).
1556-- GET /lookup/{pubkey}: return our local server name if the key is here,
1557-- "remote" if known on a peer, "" otherwise. Peer hostnames are deliberately
1558-- masked (see section header).
1559handleLookupPresence : PubkeyHex -> Text -> Text -> ({authorized: Bool, error: Text, server: Text} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1560handleLookupPresence = \pubkey authorization clientIp respond -> do
1562 let mkErr = \e -> {server: "", authorized: False {}, error: if e == "rate_limited" then e else ""}
1563 gateLookup authorization clientIp t respond mkErr {
1565 authorized: True {},
1566 error: "rate_limited"
1569 yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))
1571-- POST /lookup/batch: same masking as `handleLookupPresence`, but for up to
1572-- 50 pubkeys in one round-trip.
1573handleBatchLookupPresence : [PubkeyHex] -> Text -> Text -> ({authorized: Bool, error: Text, results: [{pubkey: PubkeyHex, server: Text}]} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1574handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1578 authorized: False {},
1579 error: if e == "rate_limited" then e else ""
1581 gateLookup authorization clientIp t respond mkErr {
1583 authorized: True {},
1584 error: "rate_limited"
1587 let results = pluck (takeRelation 50 pubkeys) (\k -> {pubkey: k, server: maskedServer pres k t})
1588 yield (respond {results, authorized: True {}, error: ""}))
1590-- POST /messages/status: classify a message as
1591-- pending (in mailbox), forwarding (en-route to a peer),
1592-- delivered (acked or terminal), or unknown.
1593handleMessageStatus : PubkeyHex -> MessageIdHex -> Text -> Text -> ({authorized: Bool, status: Text} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1594handleMessageStatus = \to messageId authorization clientIp respond -> do
1596 let mkErr = \_ -> {status: "", authorized: False {}}
1597 gateAuth (False {}) authorization clientIp t respond mkErr (\_ -> do
1600 dids <- *deliveredIds
1601 let blobMatches = \x -> messageIdOf x.encryptedBlob == messageId && x.toKey == to
1602 let didMatches = \d -> d.messageId == messageId && d.toKey == to
1603 let s = if any msgs blobMatches
1605 else if any fwds blobMatches
1607 else if any dids didMatches then "delivered" else "unknown"
1608 yield (respond {status: s, authorized: True {}}))
1610-- ============================================
1611-- Handler: ack / list / delete
1612-- ============================================
1613-- Mailbox mutation by the recipient (ack, list, delete).
1615-- POST /messages/ack: explicit ack by message id (alternative to the implicit
1616-- cursor-based ack in /poll). Drops mailbox+forward rows.
1617handleAckMessages : [MessageIdHex] -> Text -> Text -> ({ackedCount: Int, authorized: Bool, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1618handleAckMessages = \messageIds authorization clientIp respond -> do
1620 let mkErr = \e -> {ok: False {}, authorized: False {}, ackedCount: 0, error: e}
1621 gateClientWrite authorization clientIp t respond mkErr {
1623 authorized: True {},
1625 error: "rate_limited"
1627 ackedCount <- atomic (do
1628 toAck <- removeForKey (\x -> x.toKey == pubkey && elem (messageIdOf x.encryptedBlob) (dedupMessageIds messageIds))
1629 yield (count toAck))
1630 yield (respond {ok: True {}, authorized: True {}, ackedCount, error: ""}))
1632-- POST /messages/list: paginated peek at the caller's mailbox without acking.
1633-- Returns metadata only (no ciphertext) plus the total count.
1634handleListMessages : 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
1635handleListMessages = \offset authorization clientIp respond -> do
1639 authorized: False {},
1641 error: if e == "rate_limited" then e else ""
1643 gateAuth (True {}) authorization clientIp t respond mkErr (\pubkey -> do
1645 rates <- *lookupRates
1646 let r = checkRate rates pubkey maxLookupRatePerKey lookupRateWindowMs t
1647 *lookupRates = r.entries
1650 then yield (respond {messages: [], authorized: True {}, count: 0, error: "rate_limited"})
1653 let myMsgs = sortBy (\m -> m.receivedAt) (filter msgs (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge))
1654 let total = count myMsgs
1655 let page = pluck (takeRelation maxPollEvents (dropRelation (clamp 0 total offset) myMsgs)) (\m -> {
1656 messageId: messageIdOf m.encryptedBlob,
1659 yield (respond {messages: page, authorized: True {}, count: total, error: ""}))
1661-- POST /messages/delete: same effect as ack from the recipient's side, with
1662-- a "delete" semantic for the response.
1663handleDeleteMessages : [MessageIdHex] -> Text -> Text -> ({authorized: Bool, deletedCount: Int, error: Text, ok: Bool} -> a) -> IO {r *drainPhase, rw *globalRateCount, rw *globalRateWindowStart, r *sessions, clock} a
1664handleDeleteMessages = \messageIds authorization clientIp respond -> do
1668 authorized: False {},
1670 error: if e == "draining" then "" else e
1672 gateClientWrite authorization clientIp t respond mkErr {
1674 authorized: True {},
1676 error: "rate_limited"
1678 deletedCount <- atomic (do
1679 toDel <- removeForKey (\x -> x.toKey == pubkey && elem (messageIdOf x.encryptedBlob) (dedupMessageIds messageIds))
1680 yield (count toDel))
1681 yield (respond {ok: True {}, authorized: True {}, deletedCount, error: ""}))
1683-- POST /messages/batch: multi-recipient send in one call. Each entry runs
1684-- through `sendMessageCore` separately so per-entry rate limits/errors don't
1685-- abort the rest; federated forwards are forked per entry.
1686handleSendBatch : [{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
1687handleSendBatch = \messages authorization clientIp respond -> do
1689 let mkErr = \_ -> {results: [], authorized: False {}, retryAfterMs: 0}
1690 gateAuth (True {}) authorization clientIp t respond mkErr (\senderKey -> do
1691 results <- mapEach (takeRelation maxBatchSize messages) (\m -> do
1692 let effMsgId = messageIdOf m.encryptedBlob
1693 r <- atomic (do sendMessageCore senderKey m.to m.encryptedBlob t)
1694 when federationEnabled (fork (forkForwardsTo m.to m.encryptedBlob r.remoteServers))
1695 yield {messageId: effMsgId, r.status})
1696 yield (respond {results, authorized: True {}, retryAfterMs: 0}))
1698-- ============================================
1699-- Handler: federation receive
1700-- ============================================
1701-- Inbound endpoints called peer-to-peer. Every handler funnels through
1702-- `withFederationAuth` which checks: peer block/allow lists, request
1703-- precondition (e.g. expected origin/server constraints), clock skew vs the
1704-- request timestamp, Ed25519 signature auth, and per-peer rate limit.
1706-- GET /federation/pubkey: publish the Ed25519 public key peers should use to
1707-- verify our `Signed ...` federation auth headers. Returns ok=False when the
1708-- keypair has not yet been generated (should not happen post-startup).
1709handleFederationPubkey : ({pubkey: Text, ok: Bool} -> ret) -> IO {r *federationPublicKeyHex} ret
1710handleFederationPubkey = \respond -> do
1711 hex <- *federationPublicKeyHex
1712 yield (respond {pubkey: hex, ok: hex != ""})
1714-- POST /federation/gossip: ingest peer presence events. The transport-level
1715-- `fromServer` is the home server for every event. Drops every (pubkey,
1716-- fromServer) row touched by this batch, then re-adds fresh-TTL rows for
1717-- pubkeys with a `GossipOnline` event (so Online wins on collision). For keys
1718-- that transition offline → online, flushes any locally-buffered messages via
1719-- `forwardQueuedToServer`.
1720handleRecvGossip : [{eventType: <GossipOffline {} | GossipOnline {} | a>, pubkey: PubkeyHex | b}] -> Text -> Int -> Text -> ({error: Text, ok: Bool, retryAfterMs: Int} -> c) -> IO {r *allowedPeers, r *blockedPeers, rw *federationRates, rw *presence, clock, network} c
1721handleRecvGossip = \events fromServer requestTimestamp authorization respond -> do
1723 withFederationAuth authorization t respond fromServer "gossip" requestTimestamp (isRemote fromServer && count events <= maxGossipEvents) (do
1724 newlyOnline <- atomic (do
1726 let touchedPubkeys = pluck events (\e -> e.pubkey)
1727 let onlinePubkeys = pluck (filter events (\e -> case e.eventType of
1728 GossipOnline {} -> True {}
1729 _ -> False {})) (\e -> e.pubkey)
1730 let kept = filter pres (\p -> p.server != fromServer || not elem p.pubkey touchedPubkeys)
1731 let added = map (\pk -> {pubkey: pk, server: fromServer, expiresAt: t + onlineGossipTtl}) onlinePubkeys
1732 *presence = union kept added
1733 yield (filter onlinePubkeys (\pk -> none pres (\p -> p.pubkey == pk && p.server == fromServer && p.expiresAt > t))))
1734 fork (forEach newlyOnline (\pk -> forwardQueuedToServer pk fromServer t))
1735 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1737-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
1738-- Returns "" on success (including duplicates), otherwise an error code:
1739-- "no_presence" | "rate_limited" (only when checkRecvRate).
1741-- The presence gate requires the recipient to be **currently locally online**
1742-- (not just visible to gossip somewhere). Without that, an A → B forward based
1743-- on stale gossip about R could land on B with no actively-polling client and
1744-- get stuck. If B refuses, A retries via `retryPendingForwards` against the
1745-- next presence target.
1746storeFederated : PubkeyHex -> BlobHex -> Bool -> Int<Ms> -> IO {r *deliveredIds, rw *messages, rw *pollWakeup, r *presence, rw *recvRates, rw *seqCounter} Text
1747storeFederated = \toKey encryptedBlob checkRecvRate t -> do
1750 dids <- *deliveredIds
1751 if not any pres (\p -> p.pubkey == toKey && isLocalServer p.server && p.expiresAt > t)
1752 then yield "no_presence"
1753 else if isDuplicateMessage encryptedBlob toKey msgs dids
1756 limited <- if checkRecvRate
1758 rRates <- *recvRates
1759 let r = checkRate rRates toKey maxRecvRatePerKey recvRateWindowMs t
1760 *recvRates = r.entries
1762 else yield (False {})
1764 then yield "rate_limited"
1767 let seqTs = nextSeqTs t seq
1769 *messages = union msgs [{toKey, encryptedBlob, receivedAt: seqTs}]
1773-- POST /federation/forward: receive a single federated message destined for
1774-- a local key. Delegates the actual store decision to `storeFederated`,
1775-- which returns "" on success (including the duplicate case). The peer-
1776-- supplied messageId in the request body is ignored: we derive the id from
1777-- the blob ourselves so a hostile peer cannot bypass dedup by varying the
1778-- id. Honest senders derive the same value, so the "forward:" ++ messageId
1779-- auth scope still verifies.
1780handleForwardMessage : 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
1781handleForwardMessage = \toKey encryptedBlob _ fromServer requestTimestamp authorization respond -> do
1782 let messageId = messageIdOf encryptedBlob
1784 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (isRemote fromServer) (do
1785 result <- atomic storeFederated toKey encryptedBlob (True {}) t
1786 yield (respond {ok: result == "", error: result, retryAfterMs: 0}))
1788-- POST /federation/forward-batch: bulk variant of `handleForwardMessage`.
1789-- Skips per-entry recv-rate charging (the per-server federation rate limit
1790-- already gates the call as a whole).
1791handleFederationBatchForward : [{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
1792handleFederationBatchForward = \forwards fromServer requestTimestamp authorization respond -> do
1794 let batchRespond = \r -> respond {r.ok, r.error, results: [], r.retryAfterMs}
1795 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (isRemote fromServer && count forwards <= maxBatchSize) (do
1796 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> do
1797 -- Recompute id from blob; ignore e.messageId (untrusted, peer-chosen).
1798 let messageId = messageIdOf e.encryptedBlob
1799 result <- atomic storeFederated e.toKey e.encryptedBlob (False {}) t
1800 yield {messageId, ok: result == ""})
1801 yield (respond {ok: True {}, error: "", results, retryAfterMs: 0}))
1803-- ============================================
1804-- Maintenance handlers
1805-- ============================================
1806-- Operator endpoints, all gated by `runtimeMaintenanceSecret` (Bearer header).
1807-- Drain is two-phase: phase 1 stops accepting new auth, phase 2 also clears
1808-- local presence and gossips offline. Admin endpoints can revoke sessions by
1809-- pubkey or IP, purge a recipient's mailbox, list aggregated session activity,
1810-- block/allow peers, rotate the maintenance secret, and read live config.
1812-- POST /maintenance/drain: phase 1 stops accepting new auth (challenges,
1813-- verify); phase 2 additionally clears local presence and broadcasts offline
1814-- gossip for every locally-live key so peers can reroute. Idempotent.
1815handleDrain : Int -> Text -> ({drained: Int, ok: Bool, phase: Int} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, rw *pollWakeup, rw *presence, r *runtimeMaintenanceSecret, clock} a
1816handleDrain = \phase authorization respond -> withMaintAuth authorization (yield (respond {
1821 let effectivePhase = if phase == 1 then 1 else 2
1823 drainResult <- atomic (do
1824 *drainPhase = effectivePhase
1825 *drainCompletedAt = t
1826 if effectivePhase == 1
1827 then yield {peers: [], localKeys: []}
1830 let isLocalLive = \p -> isLocalServer p.server && p.expiresAt > t
1831 *presence = removeWhere pres isLocalLive
1833 yield {peers: knownPeerServersFromPresence pres, localKeys: filter pres isLocalLive})
1834 when (effectivePhase == 2 && count drainResult.localKeys > 0 && federationEnabled) (fork (sendBatchedGossip drainResult.peers (pluck drainResult.localKeys (\p -> mkGossip (GossipOffline {}) p.pubkey))))
1835 yield (respond {ok: True {}, drained: count drainResult.localKeys, phase: effectivePhase}))
1837-- POST /maintenance/undrain: clear drain state and resume normal operation.
1838-- Does not restore presence — clients must re-authenticate to repopulate.
1839handleUndrain : Text -> ({ok: Bool} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, r *runtimeMaintenanceSecret} a
1840handleUndrain = \authorization respond -> withMaintAuth authorization (yield (respond {ok: False {}})) (do
1841 atomic (do *drainPhase = 0; *drainCompletedAt = 0)
1842 yield (respond {ok: True {}}))
1844-- POST /maintenance/shutdown: convenience wrapper that performs a phase-2
1845-- drain and reports the count of presence rows cleared.
1846handleShutdown : Text -> ({drained: Int, ok: Bool} -> a) -> IO {rw *drainCompletedAt, rw *drainPhase, rw *pollWakeup, rw *presence, r *runtimeMaintenanceSecret, clock} a
1847handleShutdown = \authorization respond -> handleDrain 2 authorization (\r -> respond {
1852-- POST /maintenance/prune: trigger an out-of-band `pruneAndRetry` tick.
1853-- Useful when the operator wants to force expiry/forward retries without
1854-- waiting for the next `backgroundPrune` interval.
1855handlePrune : Text -> ({ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *sessions, clock} a
1856handlePrune = \authorization respond -> do
1858 authorized <- verifyMaintenanceAuth authorization
1859 when authorized (fork (pruneAndRetry t))
1860 yield (respond {ok: authorized})
1862-- POST /maintenance/reconcile: stub. Production builds run a state
1863-- reconciliation pass here; this simplified server reports zero drift.
1864handleReconcile : Text -> ({messageDrift: Int, ok: Bool, presenceDrift: Int, sessionDrift: Int} -> a) -> IO {r *runtimeMaintenanceSecret} a
1865handleReconcile = \authorization respond -> do
1866 authorized <- verifyMaintenanceAuth authorization
1867 yield (respond {ok: authorized, messageDrift: 0, presenceDrift: 0, sessionDrift: 0})
1869-- GET /health: aggregate readiness flags. During an initial warmup window
1870-- (3 prune intervals after startup) prune/heartbeat checks always pass to
1871-- avoid false negatives before the background loops have ticked.
1872handleHealth : a -> ({capacityHealthy: Bool, capacityWarning: Bool, degradedPeers: Int, draining: Bool, federationHealthy: Bool, forwardQueueHealthy: Bool, heartbeatHealthy: Bool, nonceHealthy: Bool, ok: Bool, pruneHealthy: Bool, requestPressureHealthy: Bool, version: Text, watchdogHealthy: Bool} -> b) -> IO {r *drainPhase, r *lastHeartbeatAt, r *lastPrunedAt, r *startedAt, clock} b
1873handleHealth = \authorization respond -> do
1875 lastPrune <- *lastPrunedAt
1876 lastHb <- *lastHeartbeatAt
1879 let warmup = t - sa < backgroundPruneInterval * 3
1880 let pruneOk = warmup || t - lastPrune < backgroundPruneInterval * 3
1881 let heartbeatOk = warmup || t - lastHb < pollTimeoutMs * 3
1883 ok: dp == 0 && pruneOk && heartbeatOk,
1884 version: serverBuildVersion,
1885 pruneHealthy: pruneOk,
1886 heartbeatHealthy: heartbeatOk,
1887 capacityHealthy: True {},
1888 capacityWarning: False {},
1889 federationHealthy: True {},
1890 watchdogHealthy: True {},
1891 nonceHealthy: True {},
1892 forwardQueueHealthy: True {},
1893 requestPressureHealthy: True {},
1898-- GET /ready: signals readiness once the first background prune has run
1899-- (load-balancer / orchestrator probe).
1900handleReady : ({ready: Bool} -> a) -> IO {r *lastPrunedAt, r *startedAt} a
1901handleReady = \respond -> do
1903 lastPrune <- *lastPrunedAt
1904 yield (respond {ready: lastPrune > 0 && lastPrune >= sa})
1906-- POST /maintenance/block: deny-list a peer. Outbound retries skip blocked
1907-- peers; inbound federation calls from blocked peers fail in `withFederationAuth`.
1908handleBlockPeer : Text -> Text -> ({blocked: Int, ok: Bool} -> a) -> IO {rw *blockedPeers, r *runtimeMaintenanceSecret} a
1909handleBlockPeer = \server authorization respond -> withMaintAuth authorization (yield (respond {
1914 current <- *blockedPeers
1915 let updated = union current [server]
1916 *blockedPeers = updated
1917 yield (count updated))
1918 yield (respond {ok: True {}, blocked: n}))
1920-- POST /maintenance/unblock: remove a peer from the deny-list.
1921handleUnblockPeer : Text -> Text -> ({ok: Bool, unblocked: Bool} -> a) -> IO {rw *blockedPeers, r *runtimeMaintenanceSecret} a
1922handleUnblockPeer = \server authorization respond -> withMaintAuth authorization (yield (respond {
1926 removed <- atomic (do
1927 current <- *blockedPeers
1928 *blockedPeers = removeWhere current (\x -> x == server)
1929 yield (elem server current))
1930 yield (respond {ok: True {}, unblocked: removed}))
1932-- GET /maintenance/blocked: list the deny-listed peers (empty for unauthorized).
1933handleListBlocked : Text -> ({ok: Bool, peers: [Text]} -> a) -> IO {r *blockedPeers, r *runtimeMaintenanceSecret} a
1934handleListBlocked = \authorization respond -> do
1935 ok <- verifyMaintenanceAuth authorization
1936 list <- *blockedPeers
1937 yield (respond {peers: if ok then list else [], ok})
1939-- POST /maintenance/allow: when the allowed-peers list is non-empty, only
1940-- federation calls from listed peers are accepted (whitelist mode). An empty
1941-- list disables the whitelist.
1942handleAllowPeer : Text -> Text -> ({allowed: Int, ok: Bool} -> a) -> IO {rw *allowedPeers, r *runtimeMaintenanceSecret} a
1943handleAllowPeer = \server authorization respond -> withMaintAuth authorization (yield (respond {
1948 current <- *allowedPeers
1949 let updated = union current [server]
1950 *allowedPeers = updated
1951 yield (count updated))
1952 yield (respond {ok: True {}, allowed: n}))
1954-- POST /maintenance/unallow: remove an entry from the allow-list. If the
1955-- list becomes empty, the whitelist gate disengages.
1956handleUnallowPeer : Text -> Text -> ({ok: Bool, removed: Bool} -> a) -> IO {rw *allowedPeers, r *runtimeMaintenanceSecret} a
1957handleUnallowPeer = \server authorization respond -> withMaintAuth authorization (yield (respond {
1961 removed <- atomic (do
1962 current <- *allowedPeers
1963 *allowedPeers = removeWhere current (\x -> x == server)
1964 yield (elem server current))
1965 yield (respond {ok: True {}, removed}))
1967-- GET /maintenance/allowed: list the allow-listed peers (empty for unauthorized).
1968handleListAllowed : Text -> ({ok: Bool, peers: [Text]} -> a) -> IO {r *allowedPeers, r *runtimeMaintenanceSecret} a
1969handleListAllowed = \authorization respond -> do
1970 ok <- verifyMaintenanceAuth authorization
1971 list <- *allowedPeers
1972 yield (respond {peers: if ok then list else [], ok})
1974-- POST /maintenance/ping: send a zero-event gossip to a peer and return the
1975-- round-trip latency. Useful for diagnosing federation reachability without
1976-- waiting for organic traffic.
1977handlePingPeer : Text -> Text -> ({error: Text, latencyMs: Int, ok: Bool} -> a) -> IO {r *federationPrivateKeyHex, r *runtimeMaintenanceSecret, network, clock, random} a
1978handlePingPeer = \server authorization respond -> withMaintAuth authorization (yield (respond {
1981 error: "unauthorized"
1984 auth <- generateFederationAuth "gossip" startTime
1985 result <- fetch (federationUrl server) (RecvGossip {
1986 authorization: auth,
1988 fromServer: serverHostN,
1989 requestTimestamp: startTime,
1994 Err e -> yield (respond {ok: False {}, latencyMs: endTime - startTime, error: show e})
1995 Ok {value: r} -> yield (respond {r.ok, latencyMs: endTime - startTime, r.error}))
1997-- GET /maintenance/sessions/{pubkey}: list a single user's active sessions.
1998-- Token values are masked ("***") — operators see expiry, ip, createdAt only.
1999handleAdminListSessions : PubkeyHex -> Text -> ({count: Int, ok: Bool, sessions: [{createdAt: Int<Ms>, expiresAt: Int<Ms>, ip: Text, token: Text}]} -> a) -> IO {r *runtimeMaintenanceSecret, r *sessions} a
2000handleAdminListSessions = \pubkey authorization respond -> do
2001 ok <- verifyMaintenanceAuth authorization
2004 then mapWhere ses (\s -> s.pubkey == pubkey) (\s -> {token: "***", s.expiresAt, s.ip, s.createdAt})
2006 yield (respond {sessions: mine, count: count mine, ok})
2008-- POST /maintenance/sessions/revoke-all: drop every session and local presence
2009-- row for `pubkey`. Operator-side kick for a compromised key.
2010handleAdminRevokeSessions : PubkeyHex -> Text -> ({ok: Bool, revokedCount: Int} -> a) -> IO {rw *presence, r *runtimeMaintenanceSecret, rw *sessions} a
2011handleAdminRevokeSessions = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
2015 revoked <- atomic (do
2017 let same = \s -> s.pubkey == pubkey
2018 *sessions = removeWhere ses same
2020 *presence = filter pres (\p -> p.pubkey != pubkey || isRemote p.server)
2021 yield (countWhere ses same))
2022 yield (respond {ok: True {}, revokedCount: revoked}))
2024-- POST /maintenance/messages/purge: drop all queued/forwarding messages
2025-- destined for `pubkey`. With the federation ack flow gone, there's nothing
2026-- to ack back — peers will discover the absence on their next forward attempt.
2027handleAdminPurgeMessages : PubkeyHex -> Text -> ({ok: Bool, purgedCount: Int} -> a) -> IO {rw *forwards, rw *messages, r *runtimeMaintenanceSecret} a
2028handleAdminPurgeMessages = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
2032 purged <- atomic (do
2034 let mine = \m -> m.toKey == pubkey
2035 let purgedMsgs = filter msgs mine
2036 *messages = removeWhere msgs mine
2038 *forwards = filter fwds (\f -> f.toKey != pubkey)
2039 yield (count purgedMsgs))
2040 yield (respond {ok: True {}, purgedCount: purged}))
2042-- POST /maintenance/sessions/revoke-ip: kill every session bound to a given
2043-- client IP, regardless of pubkey. Reports total revoked + distinct keys.
2044handleAdminRevokeIp : Text -> Text -> ({affectedKeys: Int, ok: Bool, revokedCount: Int} -> a) -> IO {r *runtimeMaintenanceSecret, rw *sessions} a
2045handleAdminRevokeIp = \ip authorization respond -> withMaintAuth authorization (yield (respond {
2054 let same = \s -> s.ip == ip
2055 *sessions = removeWhere ses same
2056 yield (mapWhere ses same (\s -> s.pubkey)))
2057 yield (respond {ok: ip != "", revokedCount: count keys, affectedKeys: count keys}))
2059-- POST /maintenance/sessions/list: paged session inventory aggregated by
2060-- pubkey (count, distinct ips, latest expiry). Sorted by descending count.
2061handleAdminListAllSessions : Int -> Text -> ({ok: Bool, sessions: [{count: Int, ips: [Text], latestExpiry: Int, pubkey: PubkeyHex}], total: Int} -> a) -> IO {r *runtimeMaintenanceSecret, r *sessions, clock} a
2062handleAdminListAllSessions = \limit authorization respond -> withMaintAuth authorization (yield (respond {
2069 let active = filter ses (\s -> s.expiresAt > t)
2070 let grouped = pluck (pluck active (\s -> s.pubkey)) (\pk -> (\mine -> {pubkey: pk, count: count mine, ips: mapWhere mine (\s -> s.ip != "") (\s -> s.ip), latestExpiry: fold (\acc s -> max acc s.expiresAt) 0 mine}) (filter active (\s -> s.pubkey == pk)))
2071 let cap = clamp 1 500 (if limit <= 0 then 100 else limit)
2072 let top = takeRelation cap (sortBy (\g -> 0 - g.count) grouped)
2073 yield (respond {sessions: top, ok: True {}, total: count grouped}))
2075-- GET /maintenance/rate-limits: expose the configured rate ceilings. Returns
2076-- zeros for unauthenticated callers (response shape stays stable).
2077handleGetRateLimits : Text -> ({authGlobal: Int, challengeRate: Int, clientWriteRate: Int, federationRate: Int, globalRequest: Int, ipRate: Int, lookupRate: Int, ok: Bool, pairRate: Int, recvRate: Int, sendRate: Int} -> a) -> IO {r *runtimeMaintenanceSecret} a
2078handleGetRateLimits = \authorization respond -> do
2079 ok <- verifyMaintenanceAuth authorization
2080 let v = \x -> if ok then x else 0
2083 globalRequest: v maxGlobalRequestRate,
2084 ipRate: v maxIpRatePerIp,
2085 sendRate: v maxSendRatePerKey,
2086 federationRate: v maxFederationRatePerServer,
2087 authGlobal: v maxAuthGlobalRate,
2088 recvRate: v maxRecvRatePerKey,
2089 challengeRate: v maxChallengeRatePerKey,
2090 pairRate: v maxPairRatePerPair,
2091 lookupRate: v maxLookupRatePerKey,
2092 clientWriteRate: v maxClientWriteRatePerKey
2095-- POST /maintenance/rate-limits: stub. Rate ceilings are compile-time constants
2096-- in this simplified server, so this just authorizes and returns ok=true.
2097handleSetRateLimits : a -> b -> c -> d -> e -> f -> g -> h -> i -> j -> Text -> ({ok: Bool} -> k) -> IO {r *runtimeMaintenanceSecret} k
2098handleSetRateLimits = \globalRequest ipRate sendRate federationRate authGlobal recvRate challengeRate pairRate lookupRate clientWriteRate authorization respond -> do
2099 authorized <- verifyMaintenanceAuth authorization
2100 yield (respond {ok: authorized})
2102-- POST /maintenance/nonces/flush: stub. Nonce caches were removed; the
2103-- endpoint is preserved so older operator tooling keeps working.
2104handleFlushNonces : Text -> ({flushedCount: Int, ok: Bool} -> a) -> IO {r *runtimeMaintenanceSecret} a
2105handleFlushNonces = \authorization respond -> do
2106 authorized <- verifyMaintenanceAuth authorization
2107 yield (respond {ok: authorized, flushedCount: 0})
2109-- POST /maintenance/rotate-secret: replace the maintenance Bearer secret.
2110-- The old secret authenticates the rotation; subsequent calls must use the
2112handleRotateSecret : Text -> Text -> ({error: Text, ok: Bool} -> a) -> IO {rw *runtimeMaintenanceSecret} a
2113handleRotateSecret = \newSecret authorization respond -> withMaintAuth authorization (yield (respond {
2115 error: "unauthorized"
2117 atomic *runtimeMaintenanceSecret = newSecret
2118 yield (respond {ok: True {}, error: ""}))
2120-- GET /maintenance/config: expose a small set of operationally-relevant
2121-- compile-time constants (poll timeout, ack grace).
2122handleGetConfig : Text -> ({ackGracePeriod: Int, ok: Bool, pollTimeoutMs: Int} -> a) -> IO {r *runtimeMaintenanceSecret} a
2123handleGetConfig = \authorization respond -> do
2124 ok <- verifyMaintenanceAuth authorization
2125 let v = \x -> if ok then x else 0
2126 yield (respond {ok, pollTimeoutMs: v pollTimeoutMs, ackGracePeriod: v ackGracePeriod})
2128-- GET /stats: capacity gauges + minimal federation/timing snapshot. Most of
2129-- the historical metric fields are zero-filled (see `zeroStats`) — only the
2130-- counts that come straight from the state refs are populated.
2131handleStats : Text -> ({capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text} -> a) -> IO {r *activePolls, r *blockedPeers, r *challenges, r *deliveredIds, r *drainPhase, r *forwards, r *heartbeatCount, r *lastHeartbeatAt, r *lastPrunedAt, r *messages, r *presence, r *pruneCount, r *runtimeMaintenanceSecret, r *sessions, r *startedAt, clock} a
2132handleStats = \authorization respond -> withMaintAuth authorization (yield (respond zeroStats)) (do
2135 sessions <- *sessions
2139 dids <- *deliveredIds
2144 hc <- *heartbeatCount
2146 lh <- *lastHeartbeatAt
2148 let sCount = count sessions
2149 let mCount = count msgs
2150 let pCount = count pres
2151 let fCount = count fwds
2154 version: serverBuildVersion,
2157 sessionsMax: maxSessions,
2158 sessionsPct: pct sCount maxSessions,
2160 totalMessageCount: mCount,
2163 challenges: count chs,
2164 challengesMax: maxChallenges,
2165 deliveredIds: count dids,
2166 deliveredIdsMax: maxDeliveredIds,
2168 forwardNoncesMax: 0,
2170 activePolls: count aps,
2172 messageBucketMax: 0,
2176 enabled: federationEnabled,
2185 oldestForwardAge: 0,
2188 blockedPeers: count bp,
2189 drainCompletedAt: 0,
2191 maxForwardQueueDepth: 0,
2192 topFailingPeers: [],
2195 latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}
2210 pendingAckCleanups: 0,
2215 rejectedFederation: 0
2223 lastHeartbeatAt: lh,
2226 totalEntries: sCount + mCount + pCount + fCount,
2228 peakActiveRequests: 0,
2232 pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0},
2236 sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0},
2243-- All-zero /stats payload returned to unauthorized callers. The shape is kept
2244-- in lockstep with the response in `handleStats` so older operator tooling
2245-- never sees a missing field.
2246zeroStats : {capacity: {activePolls: Int, challenges: Int, challengesMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardNonces: Int, forwardNoncesMax: Int, forwards: Int, messageBucketMax: Int, messageBucketMin: Int, messages: Int, nonceEvictions: Int, presence: Int, recipientCounts: Int, sessions: Int, sessionsMax: Int, sessionsPct: Int, totalMessageCount: Int}, federation: {avgLatencyMs: Int, blockedPeers: Int, callCount: Int, drainCompletedAt: Timestamp, enabled: Bool, forwardFailures: Int, forwardServers: Int, gossipFail: Int, gossipSuccess: Int, hungSlots: Int, inFlight: Int, latencyBuckets: {gte2000: Int, lt100: Int, lt2000: Int, lt500: Int}, maxForwardQueueDepth: Int, maxLatencyMs: Int, oldestAckAge: Int, oldestForwardAge: Int, recentGossip: Int, slowVerifies: Int, topFailingPeers: [{failures: Int, server: Text}], verifyFailures: Int}, ok: Bool, rates: {challenge: Int, federation: Int, federationPair: Int, global: Int, gossipEvent: Int, ip: Int, lookup: Int, pair: Int, pendingAckCleanups: Int, recv: Int, rejectedAuth: Int, rejectedFederation: Int, rejectedGlobal: Int, rejectedIp: Int, rejectedPerKey: Int, send: Int, uniqueKeyIp: Int, verifyIp: Int}, timing: {activeRequests: Int, authAvgMs: Int, authCount: Int, authMaxMs: Int, drainPhase: Int, draining: Bool, heartbeatCount: Int, lastHeartbeatAt: Timestamp, lastPruneAt: Timestamp, peakActiveRequests: Int, pollAvgMs: Int, pollCount: Int, pollLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, pollMaxMs: Int, pruneCount: Int, pruneRunning: Timestamp, sendAvgMs: Int, sendCount: Int, sendLatencyBuckets: {gte1000: Int, lt1000: Int, lt200: Int, lt50: Int}, sendMaxMs: Int, totalEntries: Int, uptime: Int}, version: Text}
2249 version: serverBuildVersion,
2255 totalMessageCount: 0,
2263 forwardNoncesMax: 0,
2267 messageBucketMax: 0,
2280 oldestForwardAge: 0,
2284 drainCompletedAt: 0,
2286 maxForwardQueueDepth: 0,
2287 topFailingPeers: [],
2290 latencyBuckets: {lt100: 0, lt500: 0, lt2000: 0, gte2000: 0}
2305 pendingAckCleanups: 0,
2310 rejectedFederation: 0
2323 peakActiveRequests: 0,
2327 pollLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0},
2331 sendLatencyBuckets: {lt50: 0, lt200: 0, lt1000: 0, gte1000: 0},
2338-- GET /stats/peers: per-peer failure summary derived from `forwardFailures`,
2339-- including current backoff and pending forward queue depth.
2340handleStatsPeers : Text -> ({ok: Bool, peers: [{avgLatencyMs: Int, backoffMs: Int<Ms>, failures: Int, firstFailedAt: Int<Ms>, gossipEventsPerMin: Int, lastFailedAt: Int<Ms>, latencySamples: Int, maxLatencyMs: Int, pendingForwards: Int, retriable: Bool, server: Text}]} -> a) -> IO {r *forwardFailures, r *forwards, r *runtimeMaintenanceSecret, clock} a
2341handleStatsPeers = \authorization respond -> withMaintAuth authorization (yield (respond {
2346 failures <- *forwardFailures
2349 peers: pluck failures (\f -> {
2352 lastFailedAt: f.failedAt,
2354 backoffMs: retryBackoff f.failures,
2355 retriable: shouldRetryServer f.server t failures,
2356 pendingForwards: countWhere fwds (\x -> x.toServer == f.server),
2360 gossipEventsPerMin: 0
2365-- GET /stats/key/{pubkey}: per-key footprint — message count, presence rows,
2366-- session count, queued forwards (with details), and active long-poll count.
2367handleStatsKey : PubkeyHex -> Text -> ({activePolls: Int, forwardDetails: [{createdAt: Int<Ms>, retries: Int, server: ServerName}], forwards: Int, messages: Int, ok: Bool, presence: [{expiresAt: Int<Ms>, server: AnyServerName}], sessions: Int} -> a) -> IO {r *activePolls, r *forwards, r *messages, r *presence, r *runtimeMaintenanceSecret, r *sessions, clock} a
2368handleStatsKey = \pubkey authorization respond -> withMaintAuth authorization (yield (respond {
2383 let myFwds = filter fwds (\x -> x.toKey == pubkey)
2385 messages: countWhere msgs (\x -> x.toKey == pubkey),
2386 presence: mapWhere pres (\x -> x.pubkey == pubkey && x.expiresAt > t) (\x -> {
2390 sessions: countWhere ses (\x -> x.pubkey == pubkey),
2391 forwards: count myFwds,
2392 forwardDetails: pluck myFwds (\x -> {server: x.toServer, x.retries, x.createdAt}),
2393 activePolls: countWhere aps (\x -> x.pubkey == pubkey),
2397-- POST /stats/top-recipients: ranked list of pubkeys by mailbox depth.
2398-- `limit` defaults to 20, capped at 100, used to spot hot-spots / abuse.
2399handleTopRecipients : Int -> Text -> ({ok: Bool, recipients: [RecipientCount], total: Int} -> a) -> IO {r *messages, r *runtimeMaintenanceSecret} a
2400handleTopRecipients = \limit authorization respond -> withMaintAuth authorization (yield (respond {
2406 let counts = pluck (pluck msgs (\m -> m.toKey)) (\pk -> {
2408 count: countWhere msgs (\m -> m.toKey == pk)
2410 let cap = clamp 1 100 (if limit <= 0 then 20 else limit)
2411 let top = takeRelation cap (sortBy (\c -> 0 - c.count) counts)
2412 yield (respond {recipients: top, ok: True {}, total: count counts}))
2414-- GET /stats/forwards: per-destination forward queue summary — count, age of
2415-- oldest pending entry, and total accumulated retries.
2416handleStatsForwards : Text -> ({ok: Bool, servers: [{count: Int<u>, oldestAge: Int<Ms>, server: ServerName, totalRetries: Int}], total: Int} -> a) -> IO {r *forwards, r *runtimeMaintenanceSecret, clock} a
2417handleStatsForwards = \authorization respond -> withMaintAuth authorization (yield (respond {
2425 servers: pluck (collectPeers fwds (\f -> f.toServer)) (\s -> (\svFwds -> {server: s, count: count svFwds, oldestAge: t - fold (\acc f -> min acc f.createdAt) t svFwds, totalRetries: fold (\acc f -> acc + f.retries) 0 svFwds}) (filter fwds (\f -> f.toServer == s))),
2430-- GET /stats/rates: same shape as /maintenance/rate-limits with an extra
2431-- scalePct field. Kept separate for backwards compatibility with older clients.
2432handleStatsRates : Text -> ({authGlobal: Int, challengeRate: Int, clientWriteRate: Int, federationRate: Int, globalRequest: Int, ipRate: Int, lookupRate: Int, ok: Bool, pairRate: Int, recvRate: Int, scalePct: Int, sendRate: Int} -> a) -> IO {r *runtimeMaintenanceSecret} a
2433handleStatsRates = \authorization respond -> do
2434 ok <- verifyMaintenanceAuth authorization
2435 let v = \x -> if ok then x else 0
2439 globalRequest: v maxGlobalRequestRate,
2440 ipRate: v maxIpRatePerIp,
2441 sendRate: v maxSendRatePerKey,
2442 federationRate: v maxFederationRatePerServer,
2443 authGlobal: v maxAuthGlobalRate,
2444 recvRate: v maxRecvRatePerKey,
2445 challengeRate: v maxChallengeRatePerKey,
2446 pairRate: v maxPairRatePerPair,
2447 lookupRate: v maxLookupRatePerKey,
2448 clientWriteRate: v maxClientWriteRatePerKey
2451-- GET /metrics: Prometheus-style text exposition of the most useful gauges.
2452-- The `authorization` header is accepted but not validated here — operators
2453-- relying on auth must front this endpoint with a reverse proxy that enforces it.
2454handleMetrics : a -> ({metrics: Text, ok: Bool} -> b) -> IO {r *activePolls, r *drainPhase, r *forwards, r *messages, r *presence, r *sessions, r *startedAt, clock} b
2455handleMetrics = \authorization respond -> do
2465 metrics: fold (\acc r -> acc ++ r.k ++ " " ++ show r.v ++ "\n") "" [
2466 {k: "skrepka_messages_total", v: count msgs},
2467 {k: "skrepka_sessions_total", v: count ses},
2468 {k: "skrepka_presence_total", v: count pres},
2469 {k: "skrepka_forwards_total", v: count fwds},
2470 {k: "skrepka_active_polls", v: count aps},
2471 {k: "skrepka_uptime_seconds", v: (t - sa) / 1000},
2472 {k: "skrepka_draining", v: if dp > 0 then 1 else 0}
2477-- ============================================
2479-- ============================================
2480-- Single match from generated request constructor (one per route entry above)
2481-- to the corresponding `handleXxx` function. Adding a new route means adding
2482-- a `route ... = NewReq` line, a `handleNewReq` definition, and a clause here.
2484-- Top-level request dispatcher: maps each generated request constructor to
2485-- its `handleXxx` function. Every route entry above must have a clause here.
2486serve = \req -> case req of
2487 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
2488 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
2489 RevokeSession {targetToken, authorization, clientIp, respond} -> handleRevoke targetToken authorization clientIp respond
2490 Logout {authorization, clientIp, respond} -> handleLogout authorization clientIp respond
2491 Poll {cursor, generation, authorization, clientIp, respond} -> handlePoll cursor generation authorization clientIp respond
2492 SendMessage {to, encryptedBlob, authorization, clientIp, respond} -> handleSendMessage to encryptedBlob authorization clientIp respond
2493 MessageStatus {to, messageId, authorization, clientIp, respond} -> handleMessageStatus to messageId authorization clientIp respond
2494 AckMessages {messageIds, authorization, clientIp, respond} -> handleAckMessages messageIds authorization clientIp respond
2495 SendBatch {messages, authorization, clientIp, respond} -> handleSendBatch messages authorization clientIp respond
2496 ListMessages {offset, authorization, clientIp, respond} -> handleListMessages offset authorization clientIp respond
2497 DeleteMessages {messageIds, authorization, clientIp, respond} -> handleDeleteMessages messageIds authorization clientIp respond
2498 LookupPresence {pubkey, authorization, clientIp, respond} -> handleLookupPresence pubkey authorization clientIp respond
2499 BatchLookupPresence {pubkeys, authorization, clientIp, respond} -> handleBatchLookupPresence pubkeys authorization clientIp respond
2500 RecvGossip {events, fromServer, requestTimestamp, authorization, respond} -> handleRecvGossip events fromServer requestTimestamp authorization respond
2501 ForwardMessage {toKey, encryptedBlob, messageId, fromServer, requestTimestamp, authorization, respond} -> handleForwardMessage toKey encryptedBlob messageId fromServer requestTimestamp authorization respond
2502 FederationBatchForward {forwards, fromServer, requestTimestamp, authorization, respond} -> handleFederationBatchForward forwards fromServer requestTimestamp authorization respond
2503 FederationPubkey {respond} -> handleFederationPubkey respond
2504 Prune {authorization, respond} -> handlePrune authorization respond
2505 Drain {phase, authorization, respond} -> handleDrain phase authorization respond
2506 Shutdown {authorization, respond} -> handleShutdown authorization respond
2507 Undrain {authorization, respond} -> handleUndrain authorization respond
2508 BlockPeer {server, authorization, respond} -> handleBlockPeer server authorization respond
2509 UnblockPeer {server, authorization, respond} -> handleUnblockPeer server authorization respond
2510 ListBlocked {authorization, respond} -> handleListBlocked authorization respond
2511 PingPeer {server, authorization, respond} -> handlePingPeer server authorization respond
2512 AdminListSessions {pubkey, authorization, respond} -> handleAdminListSessions pubkey authorization respond
2513 AdminRevokeSessions {pubkey, authorization, respond} -> handleAdminRevokeSessions pubkey authorization respond
2514 AdminListAllSessions {limit, authorization, respond} -> handleAdminListAllSessions limit authorization respond
2515 AdminPurgeMessages {pubkey, authorization, respond} -> handleAdminPurgeMessages pubkey authorization respond
2516 AdminRevokeIp {ip, authorization, respond} -> handleAdminRevokeIp ip authorization respond
2517 GetRateLimits {authorization, respond} -> handleGetRateLimits authorization respond
2518 SetRateLimits {globalRequest, ipRate, sendRate, federationRate, authGlobal, recvRate, challengeRate, pairRate, lookupRate, clientWriteRate, authorization, respond} -> handleSetRateLimits globalRequest ipRate sendRate federationRate authGlobal recvRate challengeRate pairRate lookupRate clientWriteRate authorization respond
2519 FlushNonces {authorization, respond} -> handleFlushNonces authorization respond
2520 AllowPeer {server, authorization, respond} -> handleAllowPeer server authorization respond
2521 UnallowPeer {server, authorization, respond} -> handleUnallowPeer server authorization respond
2522 ListAllowed {authorization, respond} -> handleListAllowed authorization respond
2523 Reconcile {authorization, respond} -> handleReconcile authorization respond
2524 RotateSecret {newSecret, authorization, respond} -> handleRotateSecret newSecret authorization respond
2525 GetConfig {authorization, respond} -> handleGetConfig authorization respond
2526 Health {authorization, respond} -> handleHealth authorization respond
2527 Stats {authorization, respond} -> handleStats authorization respond
2528 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
2529 StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
2530 TopRecipients {limit, authorization, respond} -> handleTopRecipients limit authorization respond
2531 StatsForwards {authorization, respond} -> handleStatsForwards authorization respond
2532 StatsRates {authorization, respond} -> handleStatsRates authorization respond
2533 Metrics {authorization, respond} -> handleMetrics authorization respond
2534 Ready {respond} -> handleReady respond
2536-- ============================================
2538-- ============================================
2539-- `backgroundPrune` runs `pruneAndRetry` every `backgroundPruneInterval` ms
2540-- (skipped while drain phase >= 2). `pollHeartbeat` bumps `pollWakeup` every
2541-- `pollTimeoutMs` so long-poll holders unblock and return an empty heartbeat
2542-- response, keeping connections fresh.
2544-- Endless tick: every `backgroundPruneInterval` run `pruneAndRetry` (skipped
2545-- when drain phase >= 2) and bump `pruneCount` for /stats and /health.
2546backgroundPrune : IO {r *drainPhase, rw *activePolls, r *blockedPeers, rw *challenges, rw *deliveredIds, rw *forwardFailures, rw *forwards, rw *lastPresenceRefresh, rw *lastPrunedAt, rw *messages, rw *presence, rw *pruneCount, rw *sessions, clock} a
2548 sleep backgroundPruneInterval
2552 atomic (do pc <- *pruneCount; *pruneCount = pc + 1)
2556-- Endless tick that bumps `pollWakeup` every `pollTimeoutMs` so any blocked
2557-- /poll holders unblock and return an empty heartbeat response.
2558pollHeartbeat : IO {rw *heartbeatCount, rw *lastHeartbeatAt, rw *pollWakeup, clock} a
2562 atomic (do *pollWakeup = t; *lastHeartbeatAt = t; hc <- *heartbeatCount; *heartbeatCount = hc + 1)
2565-- ============================================
2567-- ============================================
2568-- Stamp `startedAt` and a fresh `serverGeneration` (used by clients to detect
2569-- server restarts and reset their poll cursor), spawn the two background
2570-- loops, and start the listener.
2572-- Entry point: stamp startup metadata, lazily generate the federation keypair
2573-- on first run, fork the background loops, and start the HTTP listener.
2577 genJitter <- randomInt 1000000
2578 atomic (do *startedAt = t; *serverGeneration = t * 1000 + genJitter)
2579 priv <- *federationPrivateKeyHex
2580 when (priv == "") (do
2581 kp <- generateKeyPair
2583 *federationPrivateKeyHex = bytesToHex kp.privateKey
2584 *federationPublicKeyHex = bytesToHex kp.publicKey)
2585 logInfo "Generated federation keypair on first startup")
2586 logInfo ("Skrepka relay server starting on " ++ bindHost ++ ":" ++ show listenPort ++ " ...")
2587 logInfo (" federation: " ++ (if federationEnabled then "enabled" else "disabled"))
2588 fork backgroundPrune
2590 listenOn bindHost listenPort serve
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list