@@ -18,6 +18,10 @@ serverHost = "localhost:8080"
1818federationSecret = "" -- must be set to enable federation
1919maintenanceSecret = "" -- set to enable /maintenance/prune independently of federation
2020federationScheme = "https://" -- scheme for federation HTTP calls
21+-- Per-peer federation secrets (overrides global federationSecret for specific peers)
22+-- Format: [{server: "peer.example.com", secret: "peer-specific-secret"}]
23+federationPeerSecrets : [{server: Text, secret: Text}]
24+federationPeerSecrets = []
2125-- Message and data size limits
2226messageMaxAge : Int<Ms>
2327messageMaxAge = 2592000000 -- 30 days
@@ -31,6 +35,7 @@ maxGossipEvents = 100
3135maxPollEvents = 50
3236maxPresenceEntries = 10000
3337maxMessagesPerRecipient = 500
38+maxMessages = 100000 -- global cap on total stored messages across all recipients
3439-- Global capacity limits for in-memory collections
3540clockSkewTolerance : Int<Ms>
3641clockSkewTolerance = 5000 -- 5 seconds tolerance for clock skew between federated servers
@@ -225,6 +230,7 @@ type LookupRateEntry = {pubkey: PubkeyHex, windowStart: Timestamp, count: Int}
225230-- Tracks recently re-broadcast gossip events to suppress amplification from duplicate cycles
226231type RecentGossipEntry = {pubkey: PubkeyHex, server: Text, timestamp: Timestamp, seenAt: Timestamp}
227232type PeerStatus = {server: Text, failures: Int, lastFailedAt: Timestamp, firstFailedAt: Timestamp, backoffMs: Int, retriable: Bool}
233+type KeyPresenceEntry = {server: Text, expiresAt: Timestamp}
228234
229235-- ============================================
230236-- Persisted State
@@ -380,11 +386,14 @@ route HealthApi where
380386-- Returns counts of in-memory collections (requires federation or maintenance secret)
381387route StatsApi where
382388 GET /stats headers {authorization: Text}
383− -> {sessions: Int, sessionsMax: Int, messages: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, ok: Bool}
389+ -> {sessions: Int, sessionsMax: Int, messages: Int, messagesMax: Int, presence: Int, presenceMax: Int, forwards: Int, forwardsMax: Int, challenges: Int, challengesMax: Int, ackedIds: Int, ackedIdsMax: Int, ackOverflow: Int, ackOverflowMax: Int, deliveredIds: Int, deliveredIdsMax: Int, forwardFailures: Int, activePolls: Int, activePollsMax: Int, sendRates: Int, challengeRates: Int, federationRates: Int, globalRateCount: Int, ipRates: Int, lookupRates: Int, recvRates: Int, pruneCount: Int, heartbeatCount: Int, uptime: Int, lastPruneAt: Timestamp, lastHeartbeatAt: Timestamp, ok: Bool}
384390 = Stats
385391 GET /stats/peers headers {authorization: Text}
386392 -> {peers: [PeerStatus], ok: Bool}
387393 = StatsPeers
394+ GET /stats/key/{pubkey: PubkeyHex} headers {authorization: Text}
395+ -> {messages: Int, presence: [KeyPresenceEntry], sessions: Int, forwards: Int, activePolls: Int, ok: Bool}
396+ = StatsKey
388397
389398route Api = AuthApi | ClientApi | FederationApi | MaintenanceApi | HealthApi | StatsApi
390399
@@ -639,6 +648,15 @@ verifyAuthWith = \check authorization ->
639648verifyFederationAuth : Text -> Bool
640649verifyFederationAuth = verifyAuthWith (\token -> verifySecret token federationSecret)
641650
651+-- Per-peer federation auth: checks peer-specific secret first, falls back to global
652+verifyFederationAuthForServer : Text -> Text -> Bool
653+verifyFederationAuthForServer = \fromServer authorization ->
654+ verifyAuthWith (\token ->
655+ case headOf (do p <- federationPeerSecrets; where normalizeServerName p.server == fromServer; yield p) of
656+ Just {value: p} -> verifySecret token p.secret
657+ Nothing {} -> verifySecret token federationSecret
658+ ) authorization
659+
642660verifyMaintenanceAuth : Text -> Bool
643661verifyMaintenanceAuth = verifyAuthWith (\token -> verifySecret token federationSecret || verifySecret token maintenanceSecret)
644662
@@ -664,10 +682,11 @@ nextSeqTs = \t seqCounter -> if t > seqCounter then t else seqCounter + 1
664682-- Deduplicate message IDs, preserving order, capped at maxMessageIds
665683dedupMessageIds : [MessageIdHex] -> [MessageIdHex]
666684dedupMessageIds = \messageIds ->
667− fold (\acc mid ->
668− if count acc >= maxMessageIds then acc
669− else if any acc (\a -> a == mid) then acc
670− else union acc [mid]) [] messageIds
685+ let result = fold (\acc mid ->
686+ if acc.n >= maxMessageIds then acc
687+ else if any acc.items (\a -> a == mid) then acc
688+ else {n: acc.n + 1, items: union acc.items [mid]}) {n: 0, items: []} messageIds
689+ in result.items
671690
672691-- Remove the N oldest acked entries (by ackedAt timestamp)
673692evictOldestAckedIds = \items n ->
@@ -1001,7 +1020,7 @@ checkPreAuth = \clientIp t -> do
10011020 else yield ""
10021021
10031022withFederationAuth = \authorization t respond fromServer precondition action ->
1004− if not (verifyFederationAuth authorization)
1023+ if not (verifyFederationAuthForServer fromServer authorization)
10051024 then do
10061025 maybePruneAndBroadcast t
10071026 yield (respond {ok: False {}, error: "unauthorized"})
@@ -1249,8 +1268,9 @@ hexCharIndex = \c ->
12491268-- Map pubkey to one of 16 signal buckets based on first hex digit.
12501269-- This partitions the signal space so storing a message for one pubkey
12511270-- only wakes polls in the same bucket, not all active polls.
1271+-- Defensive clamp ensures an invalid first char never indexes out of bounds
12521272pubkeyBucket : PubkeyHex -> Int
1253−pubkeyBucket = \pubkey -> hexCharIndex (take 1 pubkey)
1273+pubkeyBucket = \pubkey -> clamp 0 15 (hexCharIndex (take 1 pubkey))
12541274
12551275-- Update the signal timestamp for a pubkey's bucket, waking any blocked polls in that bucket
12561276signalMessageForKey = \pubkey ts ->
@@ -1452,7 +1472,7 @@ forwardCandidates = \targets messages forwards t ->
14521472
14531473
14541474challengeMatch = \challenges pubkey challenge ip t ->
1455− any challenges (\c -> c.pubkey == pubkey && c.challenge == challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))
1475+ any challenges (\c -> c.pubkey == pubkey && safeCompare c.challenge challenge && c.expiresAt > t && (ip == "" || c.ip == "" || c.ip == ip))
14561476
14571477challengesExcluding = \challenges pubkey challenge ->
14581478 do c <- challenges; where c.pubkey != pubkey || c.challenge != challenge; yield c
@@ -1680,7 +1700,9 @@ handleSendMessage = \to encryptedBlob timestamp messageId authorization respond
16801700 then yield {pr: pr, send: SendFailed {status: MsRateLimited {}}, senderRateLimited: False {}}
16811701 else do
16821702 set *recvRates = recvResult.entries
1683− if countMessagesFor messages to t >= maxMessagesPerRecipient
1703+ if count messages >= maxMessages
1704+ then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}
1705+ else if countMessagesFor messages to t >= maxMessagesPerRecipient
16841706 then yield {pr: pr, send: SendFailed {status: MsRejected {}}, senderRateLimited: False {}}
16851707 else do
16861708 presence <- *presence
@@ -1817,6 +1839,8 @@ handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fro
18171839 deliveredIds <- *deliveredIds
18181840 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds || hasMessage messageId toKey deliveredIds
18191841 then yield {pr: pr, forward: ForwardDuplicate {}}
1842+ else if count messages >= maxMessages
1843+ then yield {pr: pr, forward: ForwardRejected {}}
18201844 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
18211845 then yield {pr: pr, forward: ForwardRejected {}}
18221846 else do
@@ -1919,9 +1943,11 @@ handleHealth = \respond -> do
19191943 forwards <- *forwards
19201944 sessions <- *sessions
19211945 activePolls <- *activePolls
1946+ messages <- *messages
19221947 let warmup = t - sa < healthPruneMaxAge
19231948 -- Capacity pressure: flag when any capped collection exceeds 90% full
1924− let capacityOk = count presence * 10 <= maxPresenceEntries * 9
1949+ let capacityOk = count messages * 10 <= maxMessages * 9
1950+ && count presence * 10 <= maxPresenceEntries * 9
19251951 && count forwards * 10 <= maxForwardsEntries * 9
19261952 && count sessions * 10 <= maxSessions * 9
19271953 && count activePolls * 10 <= maxActivePolls * 9
@@ -1962,8 +1988,8 @@ handleStats = \authorization respond -> do
19621988 lp <- *lastPrunedAt
19631989 lh <- *lastHeartbeatAt
19641990 yield {sessions: count s, messages: count m, presence: count p, forwards: count f, challenges: count c, ackedIds: count a, ackOverflow: count ao, deliveredIds: count d, forwardFailures: count ff, activePolls: count ap, sendRates: count sr, challengeRates: count cr, federationRates: count fr, globalRateCount: grc, ipRates: count ir, lookupRates: count lr, recvRates: count rr, pruneCount: pc, heartbeatCount: hc, uptime: t - sa, lastPruneAt: lp, lastHeartbeatAt: lh}
1965− yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})
1966− else yield (respond {sessions: 0, sessionsMax: 0, messages: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})
1991+ yield (respond {sessions: result.sessions, sessionsMax: maxSessions, messages: result.messages, messagesMax: maxMessages, presence: result.presence, presenceMax: maxPresenceEntries, forwards: result.forwards, forwardsMax: maxForwardsEntries, challenges: result.challenges, challengesMax: maxChallenges, ackedIds: result.ackedIds, ackedIdsMax: maxAckedIds, ackOverflow: result.ackOverflow, ackOverflowMax: maxAckOverflow, deliveredIds: result.deliveredIds, deliveredIdsMax: maxDeliveredIds, forwardFailures: result.forwardFailures, activePolls: result.activePolls, activePollsMax: maxActivePolls, sendRates: result.sendRates, challengeRates: result.challengeRates, federationRates: result.federationRates, globalRateCount: result.globalRateCount, ipRates: result.ipRates, lookupRates: result.lookupRates, recvRates: result.recvRates, pruneCount: result.pruneCount, heartbeatCount: result.heartbeatCount, uptime: result.uptime, lastPruneAt: result.lastPruneAt, lastHeartbeatAt: result.lastHeartbeatAt, ok: True {}})
1992+ else yield (respond {sessions: 0, sessionsMax: 0, messages: 0, messagesMax: 0, presence: 0, presenceMax: 0, forwards: 0, forwardsMax: 0, challenges: 0, challengesMax: 0, ackedIds: 0, ackedIdsMax: 0, ackOverflow: 0, ackOverflowMax: 0, deliveredIds: 0, deliveredIdsMax: 0, forwardFailures: 0, activePolls: 0, activePollsMax: 0, sendRates: 0, challengeRates: 0, federationRates: 0, globalRateCount: 0, ipRates: 0, lookupRates: 0, recvRates: 0, pruneCount: 0, heartbeatCount: 0, uptime: 0, lastPruneAt: 0, lastHeartbeatAt: 0, ok: False {}})
19671993
19681994handleStatsPeers = \authorization respond -> do
19691995 let authorized = verifyMaintenanceAuth authorization
@@ -1975,6 +2001,26 @@ handleStatsPeers = \authorization respond -> do
19752001 yield (respond {peers: peers, ok: True {}})
19762002 else yield (respond {peers: [], ok: False {}})
19772003
2004+handleStatsKey = \pubkey authorization respond -> do
2005+ let authorized = verifyMaintenanceAuth authorization
2006+ if authorized
2007+ then do
2008+ t <- now
2009+ result <- atomic do
2010+ messages <- *messages
2011+ presence <- *presence
2012+ sessions <- *sessions
2013+ forwards <- *forwards
2014+ activePolls <- *activePolls
2015+ let keyMessages = count (do m <- messages; where m.toKey == pubkey; yield m)
2016+ let keyPresence = do p <- presence; where p.pubkey == pubkey && p.expiresAt > t; yield {server: p.server, expiresAt: p.expiresAt}
2017+ let keySessions = count (do s <- sessions; where s.pubkey == pubkey; yield s)
2018+ let keyForwards = count (do f <- forwards; where f.toKey == pubkey; yield f)
2019+ let keyPolls = count (do p <- activePolls; where p.pubkey == pubkey; yield p)
2020+ yield {messages: keyMessages, presence: keyPresence, sessions: keySessions, forwards: keyForwards, activePolls: keyPolls}
2021+ yield (respond {messages: result.messages, presence: result.presence, sessions: result.sessions, forwards: result.forwards, activePolls: result.activePolls, ok: True {}})
2022+ else yield (respond {messages: 0, presence: [], sessions: 0, forwards: 0, activePolls: 0, ok: False {}})
2023+
19782024serve = \req -> case req of
19792025 ReqChallenge {pubkey, clientIp, respond} -> handleChallenge pubkey clientIp respond
19802026 VerifyAuth {pubkey, challenge, signature, revokeOthers, clientIp, respond} -> handleVerifyAuth pubkey challenge signature revokeOthers clientIp respond
@@ -1989,6 +2035,7 @@ serve = \req -> case req of
19892035 Health {respond} -> handleHealth respond
19902036 Stats {authorization, respond} -> handleStats authorization respond
19912037 StatsPeers {authorization, respond} -> handleStatsPeers authorization respond
2038+ StatsKey {pubkey, authorization, respond} -> handleStatsKey pubkey authorization respond
19922039
19932040-- ============================================
19942041-- Entry Point
@@ -2003,16 +2050,18 @@ backgroundPrune = do
20032050 if draining then println "backgroundPrune: stopping (server draining)"
20042051 else do
20052052 t <- now
2006− canRun <- atomic do
2053+ status <- atomic do
20072054 running <- *pruneRunning
2008− if running > 0 && t - running < backgroundPruneInterval * 5
2009− then yield (False {})
2055+ if running > 0 && t - running < backgroundPruneInterval * 3
2056+ then yield {canRun: False {}, recovered: False {}}
20102057 else do
20112058 set *pruneRunning = t
20122059 pc <- *pruneCount
20132060 set *pruneCount = pc + 1
2014− yield (True {})
2015− if canRun
2061+ yield {canRun: True {}, recovered: running > 0}
2062+ when status.recovered
2063+ (println "backgroundPrune: previous cycle appears stuck, forcing new cycle")
2064+ if status.canRun
20162065 then fork (do
20172066 pruneAndRetry t
20182067 atomic do set *pruneRunning = 0)
@@ -2039,12 +2088,19 @@ pollHeartbeat = do
20392088 pollHeartbeat
20402089
20412090main = do
2042− t <- now
2043− atomic do set *startedAt = t
2044− println "Skrepka relay server starting on port 8080..."
2045− println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
2046− println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
2047− println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
2048− fork backgroundPrune
2049− fork pollHeartbeat
2050− listen 8080 serve
2091+ let configOk = maxMessages > 0 && maxMessagesPerRecipient > 0 && maxPresenceEntries > 0
2092+ && maxSessions > 0 && maxChallenges > 0 && backgroundPruneInterval > 0
2093+ && pollTimeoutMs > 0 && challengeExpiry > 0 && sessionExpiry > 0
2094+ && maxPresenceTtl > 0 && maxForwardsEntries > 0 && maxAckedIds > 0
2095+ if not configOk
2096+ then println "FATAL: invalid configuration — capacity limits and intervals must be > 0, server not started"
2097+ else do
2098+ t <- now
2099+ atomic do set *startedAt = t
2100+ println "Skrepka relay server starting on port 8080..."
2101+ println (" federation: " ++ if federationEnabled then "enabled" else "disabled")
2102+ println (" prune interval: " ++ show backgroundPruneInterval ++ "ms")
2103+ println (" poll timeout: " ++ show pollTimeoutMs ++ "ms")
2104+ fork backgroundPrune
2105+ fork pollHeartbeat
2106+ listen 8080 serve