git.iko.soy/skrepka
on master help sign in
Simplify server: extract storeFederated and maskedServer helpers, reduce duplication
caa93f4iko
server.knot+47−62
compare with master
unifiedsplit
@@ -812,6 +812,11 @@ bestServerForKey = \entries ->
812 Just {value: p} -> p.server
812 Just {value: p} -> p.server
813 _ -> ""
813 _ -> ""
814
814
815+-- Resolve a key's server for client lookup, masking remote identities behind "remote".
816+maskedServer = \pres pubkey t ->
817+ let found = bestServerForKey (presenceForKey pres pubkey t) in
818+ if found == "" then "" else if isLocalServer found then found else "remote"
819+
815hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
820hasMessage : MessageIdHex -> PubkeyHex -> [{messageId: MessageIdHex, toKey: PubkeyHex | a}] -> Bool
816hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
821hasMessage = \mid toKey items -> any items (\x -> x.messageId == mid && x.toKey == toKey)
817
822
@@ -1280,9 +1285,7 @@ handleLookupPresence = \pubkey authorization clientIp respond -> do
1280 if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})
1285 if lim then yield (respond {server: "", authorized: True {}, error: "rate_limited"})
1281 else do
1286 else do
1282 pres <- *presence
1287 pres <- *presence
1283− let found = bestServerForKey (presenceForKey pres pubkey t)
1288+ yield (respond {server: maskedServer pres pubkey t, authorized: True {}, error: ""}))
1284− let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1285− yield (respond {server: masked, authorized: True {}, error: ""}))
1286
1289
1287handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1290handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1288 t <- now
1291 t <- now
@@ -1294,9 +1297,7 @@ handleBatchLookupPresence = \pubkeys authorization clientIp respond -> do
1294 pres <- *presence
1297 pres <- *presence
1295 let results = do
1298 let results = do
1296 k <- takeRelation 50 pubkeys
1299 k <- takeRelation 50 pubkeys
1297− let found = bestServerForKey (presenceForKey pres k t)
1298− let masked = if found == "" then "" else if isLocalServer found then found else "remote"
1299− yield {pubkey: k, server: masked}
1300+ yield {pubkey: k, server: maskedServer pres k t}
1300 yield (respond {results: results, authorized: True {}, error: ""}))
1301 yield (respond {results: results, authorized: True {}, error: ""}))
1301
1302
1302handleMessageStatus = \to messageId authorization clientIp respond -> do
1303handleMessageStatus = \to messageId authorization clientIp respond -> do
@@ -1500,43 +1501,49 @@ handleRecvGossip = \events hopsRemaining fromServer requestTimestamp correlation
1500 (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))
1501 (fork (forEach valid.newlyOnline (\tgt -> tgt.pubkey ++ "|" ++ tgt.server) (\tgt -> forwardQueuedToServer tgt.pubkey tgt.server t)))
1501 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1502 yield (respond {ok: True {}, error: "", retryAfterMs: 0}))
1502
1503
1504+-- Store a federated/forwarded message into the local mailbox. Caller wraps in `atomic`.
1505+-- Returns "" on success (including duplicates), otherwise an error code:
1506+-- "no_presence" | "global_full" | "mailbox_full" | "rate_limited" (only when checkRecvRate).
1507+storeFederated = \toKey encryptedBlob timestamp originServer messageId expiresAt priority checkRecvRate t -> do
1508+ pres <- *presence
1509+ if not (any pres (\p -> p.pubkey == toKey && p.expiresAt > t)) then yield "no_presence"
1510+ else do
1511+ msgs <- *messages
1512+ aids <- *ackedIds
1513+ dids <- *deliveredIds
1514+ if any msgs (\m -> m.messageId == messageId && m.toKey == toKey)
1515+ || hasMessage messageId toKey aids
1516+ || hasMessage messageId toKey dids
1517+ then yield ""
1518+ else if count msgs >= maxMessages then yield "global_full"
1519+ else if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
1520+ else do
1521+ limited <- if checkRecvRate then do
1522+ rr <- *recvRates
1523+ let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t
1524+ set *recvRates = r.entries
1525+ yield r.limited
1526+ else yield False
1527+ if limited then yield "rate_limited"
1528+ else do
1529+ seq <- *seqCounter
1530+ let seqTs = nextSeqTs t seq
1531+ set *seqCounter = seqTs
1532+ let clampedTs = clampTimestamp timestamp t
1533+ let fwdExpiresAt = if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
1534+ let clampedPri = clamp 0 maxMessagePriority priority
1535+ set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1536+ set *pollWakeup = seqTs
1537+ yield ""
1538+
1503handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
1539handleForwardMessage = \toKey encryptedBlob timestamp originServer messageId fromServer requestTimestamp correlationId expiresAt priority authorization respond -> do
1504 t <- now
1540 t <- now
1505 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
1541 withFederationAuth authorization t respond fromServer ("forward:" ++ messageId) requestTimestamp (not (isLocalServer originServer) && not (isLocalServer fromServer)) (do
1506 if t - timestamp > messageMaxAge
1542 if t - timestamp > messageMaxAge
1507 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
1543 then yield (respond {ok: False {}, error: "stale_timestamp", retryAfterMs: 0})
1508 else do
1544 else do
1509− let clampedTs = clampTimestamp timestamp t
1545+ result <- atomic (storeFederated toKey encryptedBlob timestamp originServer messageId expiresAt priority (True {}) t)
1510− result <- atomic do
1511− pres <- *presence
1512− let hasPresence = any pres (\p -> p.pubkey == toKey && p.expiresAt > t)
1513− if not hasPresence then yield "no_presence"
1514− else do
1515− msgs <- *messages
1516− aids <- *ackedIds
1517− dids <- *deliveredIds
1518− let isDup = any msgs (\m -> m.messageId == messageId && m.toKey == toKey)
1519− || hasMessage messageId toKey aids
1520− || hasMessage messageId toKey dids
1521− if isDup then yield "duplicate"
1522− else if count msgs >= maxMessages then yield "global_full"
1523− else
1524− if countWhere msgs (\m -> m.toKey == toKey) >= maxMessagesPerRecipient then yield "mailbox_full"
1525− else do
1526− rr <- *recvRates
1527− let r = checkRate rr toKey maxRecvRatePerKey recvRateWindowMs 5000 t
1528− if r.limited then yield "rate_limited"
1529− else do
1530− set *recvRates = r.entries
1531− seq <- *seqCounter
1532− let seqTs = nextSeqTs t seq
1533− set *seqCounter = seqTs
1534− let fwdExpiresAt = if expiresAt > t then min expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
1535− let clampedPri = clamp 0 maxMessagePriority priority
1536− set *messages = union msgs [{messageId: messageId, toKey: toKey, encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1537− set *pollWakeup = seqTs
1538− yield "stored"
1539− if result == "stored" || result == "duplicate"
1546+ if result == ""
1540 then yield (respond {ok: True {}, error: "", retryAfterMs: 0})
1547 then yield (respond {ok: True {}, error: "", retryAfterMs: 0})
1541 else yield (respond {ok: False {}, error: result, retryAfterMs: 0}))
1548 else yield (respond {ok: False {}, error: result, retryAfterMs: 0}))
1542
1549
@@ -1604,31 +1611,9 @@ handleFederationBatchForward = \forwards fromServer requestTimestamp correlation
1604 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1611 withFederationAuth authorization t batchRespond fromServer "forward-batch" requestTimestamp (not (isLocalServer fromServer) && count forwards <= maxBatchSize) (do
1605 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> e.messageId) (\e -> do
1612 results <- mapEach (takeRelation maxBatchSize forwards) (\e -> e.messageId) (\e -> do
1606 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
1613 ok <- if isLocalServer e.originServer || t - e.timestamp > messageMaxAge then yield False
1607− else atomic do
1614+ else do
1608− pres <- *presence
1615+ result <- atomic (storeFederated e.toKey e.encryptedBlob e.timestamp e.originServer e.messageId e.expiresAt e.priority (False {}) t)
1609− let hasPresence = any pres (\p -> p.pubkey == e.toKey && p.expiresAt > t)
1616+ yield (result == "")
1610− if not hasPresence then yield False
1611− else do
1612− msgs <- *messages
1613− aids <- *ackedIds
1614− dids <- *deliveredIds
1615− let isDup = any msgs (\m -> m.messageId == e.messageId && m.toKey == e.toKey)
1616− || hasMessage e.messageId e.toKey aids
1617− || hasMessage e.messageId e.toKey dids
1618− if isDup then yield True
1619− else if count msgs >= maxMessages then yield False
1620− else
1621− if countWhere msgs (\m -> m.toKey == e.toKey) >= maxMessagesPerRecipient then yield False
1622− else do
1623− seq <- *seqCounter
1624− let seqTs = nextSeqTs t seq
1625− set *seqCounter = seqTs
1626− let clampedTs = clampTimestamp e.timestamp t
1627− let fwdExpiresAt = if e.expiresAt > t then min e.expiresAt (seqTs + unackedMessageMaxAge) else seqTs + unackedMessageMaxAge
1628− let clampedPri = clamp 0 maxMessagePriority e.priority
1629− set *messages = union msgs [{messageId: e.messageId, toKey: e.toKey, encryptedBlob: e.encryptedBlob, timestamp: clampedTs, originServer: e.originServer, receivedAt: seqTs, expiresAt: fwdExpiresAt, senderKey: "", priority: clampedPri}]
1630− set *pollWakeup = seqTs
1631− yield True
1632 yield {messageId: e.messageId, ok: ok})
1617 yield {messageId: e.messageId, ok: ok})
1633 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1618 yield (respond {ok: True {}, error: "", results: results, retryAfterMs: 0}))
1634
1619