3964fd8 · skrepka · git.iko.soy

git.iko.soy/skrepka

Extract helpers (firstOf, sessionPubkey, forkBroadcastOffline, removeForward) and add MessageStatus type for improved clarity and type safety

3964fd8iko6 months ago

server.knot+32−24

compare with master

unifiedsplit

@@ -47,6 +47,8 @@ type GossipEventType = Text where \s -> s == "online" || s == "offline"

48type GossipEvent = {eventType: GossipEventType, pubkey: Text, server: Text, timestamp: Timestamp, ttl: Int<S>}

50+type MessageStatus = Text where \s -> s == "unauthorized" || s == "rejected" || s == "delivered" || s == "federated" || s == "queued"

51+

50type PubkeyHex = Text where \s -> isValidHex s && not (tooLong s maxPubkeyLen)

52type PubkeyHex = Text where \s -> isValidHex s && not (tooLong s maxPubkeyLen)

51type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)

53type BlobHex = Text where \s -> isValidHex s && not (tooLong s maxBlobLen)

52type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)

54type SignatureHex = Text where \s -> isValidHex s && not (tooLong s maxSignatureLen)

@@ -95,7 +97,7 @@ route ClientApi where

95 = Poll

97 = Poll

96 /messages

98 /messages

97 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}

99 POST {to: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp} / headers {authorization: Text}

98− -> {status: Text, messageId: Text}

100+ -> {status: MessageStatus, messageId: Text}

99 = SendMessage

101 = SendMessage

100 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}

102 POST {messageIds: [MessageIdHex]} /ack headers {authorization: Text}

101 -> {ok: Bool}

103 -> {ok: Bool}

@@ -133,6 +135,8 @@ isJust = \m -> case m of

133 Nothing {} -> False {}

135 Nothing {} -> False {}

134 _ -> True {}

136 _ -> True {}

135

137

138+firstOf = \items -> fold (\acc s -> if acc == "" then s else acc) "" items

139+

136msToS : Int<Ms> -> Int<S>

140msToS : Int<Ms> -> Int<S>

137msToS = \x -> x / 1000

141msToS = \x -> x / 1000

138

142

@@ -357,6 +361,10 @@ lookupSession = \sessions token t ->

357 _ -> acc

361 _ -> acc

358 else acc) Nothing {} sessions

362 else acc) Nothing {} sessions

359

363

364+sessionPubkey = \session -> case session of

365+ Nothing {} -> ""

366+ Just {value: s} -> s.pubkey

367+

360extractToken = \authHeader ->

368extractToken = \authHeader ->

361 if toLower (take 7 authHeader) == "bearer "

369 if toLower (take 7 authHeader) == "bearer "

362 && not (tooLong (drop 7 authHeader) maxTokenLen)

370 && not (tooLong (drop 7 authHeader) maxTokenLen)

@@ -478,6 +486,7 @@ countMessagesFor = \messages toKey t ->

478 where m.receivedAt >= t - messageMaxAge

486 where m.receivedAt >= t - messageMaxAge

479 yield m)

487 yield m)

480

488

489+validGossipEvents : [GossipEvent] -> Bool -> Timestamp -> GossipEventType -> [{pubkey: PubkeyHex, server: Text, expiresAt: Timestamp, timestamp: Timestamp, ttl: Int<S>}]

481validGossipEvents = \events validRequest t eventType ->

490validGossipEvents = \events validRequest t eventType ->

482 do

491 do

483 e <- events

492 e <- events

@@ -499,7 +508,7 @@ sendToEach = \servers action ->

499 if count servers == 0

508 if count servers == 0

500 then yield {}

509 then yield {}

501 else do

510 else do

502− let server = fold (\acc s -> if acc == "" then s else acc) "" servers

511+ let server = firstOf servers

503 let rest = do s <- servers; where not (s == server); yield s

512 let rest = do s <- servers; where not (s == server); yield s

504 if server == ""

513 if server == ""

505 then yield {}

514 then yield {}

@@ -586,6 +595,16 @@ broadcastOfflineGossip = \t peerServers expiredLocal -> do

586 sendBatchedGossip peerServers gossipEvents

595 sendBatchedGossip peerServers gossipEvents

587 else yield {}

596 else yield {}

588

597

598+forkBroadcastOffline = \t result ->

599+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)

600+

601+removeForward = \mid srv -> atomic do

602+ forwards <- *forwards

603+ set *forwards = do

604+ x <- forwards

605+ where not (x.messageId == mid && x.toServer == srv)

606+ yield x

607+

589retryForwardsForServer : [{encryptedBlob: BlobHex, fromKey: PubkeyHex, messageId: MessageIdHex, timestamp: Timestamp, toKey: PubkeyHex, toServer: ServerName}] -> Int -> IO {network} {}

608retryForwardsForServer : [{encryptedBlob: BlobHex, fromKey: PubkeyHex, messageId: MessageIdHex, timestamp: Timestamp, toKey: PubkeyHex, toServer: ServerName}] -> Int -> IO {network} {}

590retryForwardsForServer = \entries remaining ->

609retryForwardsForServer = \entries remaining ->

591 if count entries == 0 || remaining <= 0

610 if count entries == 0 || remaining <= 0

@@ -602,12 +621,7 @@ retryForwardsForServer = \entries remaining ->

602 authorization: "Bearer " ++ federationSecret})

621 authorization: "Bearer " ++ federationSecret})

603 case result of

622 case result of

604 Ok _ -> do

623 Ok _ -> do

605− atomic do

606− forwards <- *forwards

607− set *forwards = do

608− x <- forwards

609− where not (x.messageId == f.messageId && x.toServer == f.toServer)

624+ removeForward f.messageId f.toServer

610− yield x

611 yield {}

625 yield {}

612 Err _ -> yield {}

626 Err _ -> yield {}

613 retryForwardsForServer rest (remaining - 1)

627 retryForwardsForServer rest (remaining - 1)

@@ -736,7 +750,7 @@ serve = \req -> case req of

736 globalCapped

750 globalCapped

737 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

751 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

738 yield pr

752 yield pr

739− fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)

753+ forkBroadcastOffline t pruneResult

740 yield (respond {challenge: challenge})

754 yield (respond {challenge: challenge})

741

755

742 VerifyAuth {pubkey, challenge, signature, respond} -> do

756 VerifyAuth {pubkey, challenge, signature, respond} -> do

@@ -813,7 +827,7 @@ serve = \req -> case req of

813 tAfterPoll <- now

827 tAfterPoll <- now

814 pruneResult <- atomic do

828 pruneResult <- atomic do

815 maybePrune tAfterPoll

829 maybePrune tAfterPoll

816− fork (broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal)

830+ forkBroadcastOffline tAfterPoll pruneResult

817 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

831 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

818

832

819 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do

833 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do

@@ -828,9 +842,7 @@ serve = \req -> case req of

828 messages <- *messages

842 messages <- *messages

829 presence <- *presence

843 presence <- *presence

830 let session = lookupSession sessions (extractToken authorization) t

844 let session = lookupSession sessions (extractToken authorization) t

831− let fromKey = case session of

845+ let fromKey = sessionPubkey session

832− Nothing {} -> ""

833− Just {value: s} -> s.pubkey

834 let authorized = isJust session

846 let authorized = isJust session

835 seqCounter <- *seqCounter

847 seqCounter <- *seqCounter

836 let seqTs = nextSeqTs t seqCounter

848 let seqTs = nextSeqTs t seqCounter

@@ -865,24 +877,20 @@ serve = \req -> case req of

865 else forwards

877 else forwards

866 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

878 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

867 let finalStatus = if result.status == "federated" && (not federationEnabled || result.forwardsAtCapacity) then "queued" else result.status

879 let finalStatus = if result.status == "federated" && (not federationEnabled || result.forwardsAtCapacity) then "queued" else result.status

880+ let actionable = not (result.status == "unauthorized") && not (result.status == "rejected")

868 fork (do

881 fork (do

869− if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "rejected") && federationEnabled

882+ if count result.remoteServers > 0 && actionable && federationEnabled

870 then sendToEach result.remoteServers (\server -> do

883 then sendToEach result.remoteServers (\server -> do

871 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {

884 fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {

872 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

885 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

873 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

886 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

874 authorization: "Bearer " ++ federationSecret})

887 authorization: "Bearer " ++ federationSecret})

875 case fwdResult of

888 case fwdResult of

876− Ok _ -> atomic do

889+ Ok _ -> removeForward msgId server

877− forwards <- *forwards

878− set *forwards = do

879− x <- forwards

880− where not (x.messageId == msgId && x.toServer == server)

881− yield x

882 Err _ -> yield {}

890 Err _ -> yield {}

883 )

891 )

884 else yield {}

892 else yield {}

885− if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "rejected")

893+ if result.forwardsAtCapacity && count result.remoteServers > 0 && actionable

886 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

894 then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

887 else yield {}

895 else yield {}

888 broadcastOfflineGossip t result.peerServers result.expiredLocal)

896 broadcastOfflineGossip t result.peerServers result.expiredLocal)

@@ -992,7 +1000,7 @@ serve = \req -> case req of

992 then serverHostN

1000 then serverHostN

993 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server

1001 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server

994 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}

1002 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}

995− fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)

1003+ forkBroadcastOffline t result

996 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})

1004 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})

997

1005

998 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

1006 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

@@ -1142,7 +1150,7 @@ serve = \req -> case req of

1142 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1150 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1143 else messages

1151 else messages

1144 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}

1152 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}

1145− fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)

1153+ forkBroadcastOffline t result

1146 yield (respond {ok: result.inserted || (authorized && validOrigin && result.alreadyExists)})

1154 yield (respond {ok: result.inserted || (authorized && validOrigin && result.alreadyExists)})

1147

1155

1148 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

1156 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

@@ -1166,7 +1174,7 @@ serve = \req -> case req of

1166 then diff forwards forwardsToRemove

1174 then diff forwards forwardsToRemove

1167 else forwards

1175 else forwards

1168 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1176 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1169− fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)

1177+ forkBroadcastOffline t pruneResult

1170 yield (respond {ok: pruneResult.validAck})

1178 yield (respond {ok: pruneResult.validAck})

1171

1179

1172 Prune {authorization, respond} -> do

1180 Prune {authorization, respond} -> do