84ef369 · skrepka · git.iko.soy

664 let ackOriginServers = collectPeerServers result.pendingAcks (\a -> a.originServer)

665 sendAcksAndClean result.pendingAcks ackOriginServers)

666

667−pollOnce : Text -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}

668pollOnce = \token cursor -> do

667pollOnce = \token cursor -> do

669 t <- now

668 t <- now

670 atomic do

669 atomic do

672 let auth = authenticateSession sessions token t

671 let auth = authenticateSession sessions token t

673 case auth of

672 case auth of

674 Unauthorized {} ->

673 Unauthorized {} ->

675− yield {events: [], cursor: -1, authorized: False {}}

674+ yield PollUnauthorized {}

676 Authorized {pubkey} -> do

675 Authorized {pubkey} -> do

677 seqCounter <- *seqCounter

676 seqCounter <- *seqCounter

678 messages <- *messages

677 messages <- *messages

692 encryptedBlob: m.encryptedBlob,

691 encryptedBlob: m.encryptedBlob,

693 timestamp: m.timestamp}}

692 timestamp: m.timestamp}}

694 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs

693 let lastReceivedAt = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) clampedCursor limitedMsgs

695− yield {events: evts, cursor: lastReceivedAt, authorized: True {}}

694+ yield PollAuthorized {events: evts, cursor: lastReceivedAt}

696

695

697−pollLoop : Text -> Int -> Int -> IO {clock} {authorized: Bool, cursor: Int, events: [PollEvent]}

698pollLoop = \token cursor deadline -> do

696pollLoop = \token cursor deadline -> do

699 result <- pollOnce token cursor

697 result <- pollOnce token cursor

700 t <- now

698 t <- now

699+ case result of

700+ PollUnauthorized {} -> yield result

701+ PollAuthorized {events} ->

701− if result.authorized && count result.events == 0 && t < deadline

702+ if count events == 0 && t < deadline

702− then do

703+ then do

703− let remaining = deadline - t

704+ let remaining = deadline - t

704− let sleepMs = if remaining > 500 then 500<Ms> else remaining

705+ let sleepMs = if remaining > 500 then 500<Ms> else remaining

705− sleep sleepMs

706+ sleep sleepMs

706− pollLoop token cursor deadline

707+ pollLoop token cursor deadline

707− else yield result

708+ else yield result

708

709

709-- ============================================

710-- ============================================

710-- Request Handler

711-- Request Handler

749 Nothing {} -> False {}

750 Nothing {} -> False {}

750 _ -> verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

751 _ -> verify (bytesFromHex pubkey) (textToBytes challenge) (bytesFromHex signature)

751 if not sigValid

752 if not sigValid

752− then yield VerifyFailed {pr: pr}

753+ then yield {pr: pr, verify: VerifyFailed {}}

753 else do

754 else do

754 set *challenges = do

755 set *challenges = do

755 c <- challenges

756 c <- challenges

769 let peerServers = if federationEnabled

770 let peerServers = if federationEnabled

770 then collectPeerServers updatedPresence (\p -> p.server)

771 then collectPeerServers updatedPresence (\p -> p.server)

771 else []

772 else []

772− yield VerifyOk {peerServers: peerServers, pr: pr}

773+ yield {pr: pr, verify: VerifyOk {peerServers: peerServers}}

774+ fork (do

773− fork (case result of

775+ let expiredLocal = case result.verify of

774− VerifyFailed {pr} ->

775− broadcastOfflineGossip t pr.peerServers pr.expiredLocal

776− VerifyOk {peerServers, pr} -> do

776+ VerifyOk _ -> do

777− when federationEnabled

778− (sendToEach peerServers (\server -> do

779− _ <- fetch (federationScheme ++ server) (RecvGossip {

780− authorization: "Bearer " ++ federationSecret,

781− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})

782− yield {}))

783− broadcastOfflineGossip t pr.peerServers (do

784− p <- pr.expiredLocal

777+ p <- result.pr.expiredLocal

785 where not (p.pubkey == pubkey)

778 where not (p.pubkey == pubkey)

786− yield p))

779+ yield p

780+ _ -> result.pr.expiredLocal

781+ case result.verify of

782+ VerifyOk {peerServers} ->

783+ when federationEnabled

784+ (sendToEach peerServers (\server -> do

785+ _ <- fetch (federationScheme ++ server) (RecvGossip {

786+ authorization: "Bearer " ++ federationSecret,

787+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})

788+ yield {}))

789+ _ -> yield {}

790+ broadcastOfflineGossip t result.pr.peerServers expiredLocal)

787− yield (case result of

791+ yield (case result.verify of

788 VerifyFailed _ -> respond {token: "", expiresAt: 0}

792 VerifyFailed _ -> respond {token: "", expiresAt: 0}

789 VerifyOk _ -> respond {token: token, expiresAt: expiry})

793 VerifyOk _ -> respond {token: token, expiresAt: expiry})

790

794

798 pruneResult <- atomic do

802 pruneResult <- atomic do

799 maybePrune tAfterPoll

803 maybePrune tAfterPoll

800 forkBroadcastOffline tAfterPoll pruneResult

804 forkBroadcastOffline tAfterPoll pruneResult

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

805+ yield (case result of

806+ PollUnauthorized {} -> respond {events: [], cursor: -1, authorized: False {}}

807+ PollAuthorized {events, cursor: c} -> respond {events: events, cursor: c, authorized: True {}})

802

808

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

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

804 let to = normalizeHex to

810 let to = normalizeHex to

812 let auth = authenticateSession sessions (extractToken authorization) t

818 let auth = authenticateSession sessions (extractToken authorization) t

813 case auth of

819 case auth of

814 Unauthorized {} ->

820 Unauthorized {} ->

815− yield SendUnauthorized {pr: pr}

821+ yield {pr: pr, send: SendUnauthorized {}}

816 Authorized {pubkey: fromKey} -> do

822 Authorized {pubkey: fromKey} -> do

817 messages <- *messages

823 messages <- *messages

818 if countMessagesFor messages to t >= maxMessagesPerRecipient

824 if countMessagesFor messages to t >= maxMessagesPerRecipient

819− then yield SendRejected {pr: pr}

825+ then yield {pr: pr, send: SendRejected {}}

820 else do

826 else do

821 presence <- *presence

827 presence <- *presence

822 seqCounter <- *seqCounter

828 seqCounter <- *seqCounter

836 set *forwards = if federationEnabled && count forwardServers > 0

842 set *forwards = if federationEnabled && count forwardServers > 0

837 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})

843 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})

838 else forwards

844 else forwards

839− yield SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey, pr: pr}

845+ yield {pr: pr, send: SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey}}

846+ fork (do

840− fork (case result of

847+ case result.send of

841− SendUnauthorized {pr} ->

842− broadcastOfflineGossip t pr.peerServers pr.expiredLocal

843− SendRejected {pr} ->

844− broadcastOfflineGossip t pr.peerServers pr.expiredLocal

845− SendDone {status, remoteServers, fromKey, pr} -> do

848+ SendDone {status, remoteServers, fromKey} -> do

846− when (count remoteServers > 0 && federationEnabled)

849+ when (count remoteServers > 0 && federationEnabled)

847− (sendToEach remoteServers (\server -> do

850+ (sendToEach remoteServers (\server -> do

848− fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {

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

849− toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,

852+ toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,

850− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

853+ timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

851− authorization: "Bearer " ++ federationSecret})

854+ authorization: "Bearer " ++ federationSecret})

852− case fwdResult of

855+ case fwdResult of

853− Ok _ -> removeForward msgId server

856+ Ok _ -> removeForward msgId server

854− Err _ -> yield {}))

857+ Err _ -> yield {}))

855− when (status == "queued" && count remoteServers > 0 && federationEnabled)

858+ when (status == "queued" && count remoteServers > 0 && federationEnabled)

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

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

860+ _ -> yield {}

857− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)

861+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

858− yield (case result of

862+ yield (case result.send of

859 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}

863 SendUnauthorized _ -> respond {status: "unauthorized", messageId: ""}

860 SendRejected _ -> respond {status: "rejected", messageId: ""}

864 SendRejected _ -> respond {status: "rejected", messageId: ""}

861 SendDone {status} -> respond {status: status, messageId: msgId})

865 SendDone {status} -> respond {status: status, messageId: msgId})

869 let auth = authenticateSession sessions (extractToken authorization) t

873 let auth = authenticateSession sessions (extractToken authorization) t

870 case auth of

874 case auth of

871 Unauthorized {} ->

875 Unauthorized {} ->

872− yield AckUnauthorized {pr: pr}

876+ yield {pr: pr, ack: AckUnauthorized {}}

873 Authorized {pubkey} -> do

877 Authorized {pubkey} -> do

874 messages <- *messages

878 messages <- *messages

875 let ackedMessages = do

879 let ackedMessages = do

891 let evictedEntries = if excessCount <= 0 then []

895 let evictedEntries = if excessCount <= 0 then []

892 else diff combinedAckedIds afterEviction

896 else diff combinedAckedIds afterEviction

893 set *ackedIds = afterEviction

897 set *ackedIds = afterEviction

894− yield AckDone {pendingAcks: ackedFederated, pr: pr, evictedEntries: evictedEntries}

898+ yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}

899+ fork (do

895− fork (case result of

900+ case result.ack of

896− AckUnauthorized {pr} ->

897− broadcastOfflineGossip t pr.peerServers pr.expiredLocal

898− AckDone {pendingAcks, pr, evictedEntries} -> do

901+ AckDone {pendingAcks, evictedEntries} -> do

899− let originServers = collectPeerServers pendingAcks (\a -> a.originServer)

902+ let originServers = collectPeerServers pendingAcks (\a -> a.originServer)

900− when (count originServers > 0 && federationEnabled)

903+ when (count originServers > 0 && federationEnabled)

901− (sendAcksAndClean pendingAcks originServers)

904+ (sendAcksAndClean pendingAcks originServers)

902− let evictedToRetry = do

905+ let evictedToRetry = do

903− a <- evictedEntries

906+ a <- evictedEntries

904− where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)

907+ where none pendingAcks (\a2 -> a2.messageId == a.messageId && a2.toKey == a.toKey)

905− yield a

908+ yield a

906− when (count evictedToRetry > 0 && federationEnabled) (do

909+ when (count evictedToRetry > 0 && federationEnabled) (do

907− let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)

910+ let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)

908− sendToEach evictedOriginServers (\server -> do

911+ sendToEach evictedOriginServers (\server -> do

909− fetchResult <- fetch (federationScheme ++ server) (FederationAck {

912+ fetchResult <- fetch (federationScheme ++ server) (FederationAck {

910− authorization: "Bearer " ++ federationSecret,

913+ authorization: "Bearer " ++ federationSecret,

911− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),

914+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),

912− originServer: server, fromServer: serverHostN})

915+ originServer: server, fromServer: serverHostN})

913− case fetchResult of

916+ case fetchResult of

914− Err _ -> do

917+ Err _ -> do

915− dropped <- atomic do

918+ dropped <- atomic do

916− ackedIds <- *ackedIds

919+ ackedIds <- *ackedIds

917− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

920+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

918− let combined = union ackedIds reinsert

921+ let combined = union ackedIds reinsert

919− let excess = count combined - maxAckedIds

922+ let excess = count combined - maxAckedIds

920− let afterEviction = if excess <= 0 then combined

923+ let afterEviction = if excess <= 0 then combined

921− else evictOldestAckedIds combined excess

924+ else evictOldestAckedIds combined excess

922− set *ackedIds = afterEviction

925+ set *ackedIds = afterEviction

923− yield excess

926+ yield excess

924− when (dropped > 0)

927+ when (dropped > 0)

925− (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))

928+ (println ("ackMessages: ackedIds at capacity (" ++ show maxAckedIds ++ "), " ++ show dropped ++ " ack entries permanently lost for " ++ server))

926− _ -> yield {}))

929+ _ -> yield {}))

930+ _ -> yield {}

927− broadcastOfflineGossip t pr.peerServers pr.expiredLocal)

931+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)

928− yield (case result of

932+ yield (case result.ack of

929 AckUnauthorized _ -> respond {ok: False {}}

933 AckUnauthorized _ -> respond {ok: False {}}

930 AckDone _ -> respond {ok: True {}})

934 AckDone _ -> respond {ok: True {}})

931

935

938 let auth = authenticateSession sessions (extractToken authorization) t

942 let auth = authenticateSession sessions (extractToken authorization) t

939 case auth of

943 case auth of

940 Unauthorized {} ->

944 Unauthorized {} ->

941− yield {foundServer: "", authorized: False {}, pr: pr}

945+ yield {pr: pr, lookup: LookupUnauthorized {}}

942 Authorized _ -> do

946 Authorized _ -> do

943 presence <- *presence

947 presence <- *presence

944 let valid = do

948 let valid = do

949 let foundServer = if any valid (\p -> p.server == serverHostN)

953 let foundServer = if any valid (\p -> p.server == serverHostN)

950 then serverHostN

954 then serverHostN

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

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

952− yield {foundServer: foundServer, authorized: True {}, pr: pr}

956+ yield {pr: pr, lookup: LookupDone {foundServer: foundServer}}

953 forkBroadcastOffline t result.pr

957 forkBroadcastOffline t result.pr

958+ yield (case result.lookup of

959+ LookupUnauthorized _ -> respond {online: False {}, server: "", authorized: False {}}

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

960+ LookupDone {foundServer} -> respond {online: not (foundServer == ""), server: foundServer, authorized: True {}})

955

961

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

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

957 t <- now

963 t <- now

958 let authorized = verifyFederationAuth authorization

964 let authorized = verifyFederationAuth authorization

959 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

965 let cappedHops = if hopsRemaining > 5 then 5 else hopsRemaining

960 let fromServer = normalizeServerName fromServer

966 let fromServer = normalizeServerName fromServer

961− if not (authorized && cappedHops >= 1 && not (fromServer == serverHostN) && count events <= maxGossipEvents)

967+ if not authorized || cappedHops < 1 || fromServer == serverHostN || count events > maxGossipEvents

962 then do

968 then do

963 pruneResult <- atomic do maybePrune t

969 pruneResult <- atomic do maybePrune t

964 forkBroadcastOffline t pruneResult

970 forkBroadcastOffline t pruneResult