@@ -87,6 +87,20 @@ impl ToJSON MessageStatus where
87 MsUnauthorized {} -> "unauthorized"
87 MsUnauthorized {} -> "unauthorized"
88 MsRejected {} -> "rejected")
88 MsRejected {} -> "rejected")
90+data DeliveryStatus = Delivered {} | Federated {} | Queued {}
92+impl ToJSON DeliveryStatus where
93+ toJson ds = toJson (case ds of
94+ Delivered {} -> "delivered"
95+ Federated {} -> "federated"
96+ Queued {} -> "queued")
98+deliveryToStatus : DeliveryStatus -> MessageStatus
99+deliveryToStatus = \ds -> case ds of
100+ Delivered {} -> MsDelivered {}
101+ Federated {} -> MsFederated {}
102+ Queued {} -> MsQueued {}
90isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
104isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
91type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
105type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
92type BlobHex = Text where isValidBoundedHex maxBlobLen
106type BlobHex = Text where isValidBoundedHex maxBlobLen
@@ -108,12 +122,9 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
108type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
122type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
109type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
123type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
110data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
124data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
111−data SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
125+data SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
112data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
126data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
113−data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
114−data LookupResult = LookupUnauthorized {} | LookupDone {server: AnyServerName}
115−data ForwardResult = ForwardDuplicate {} | ForwardStored {} | ForwardFull {}
116−data PollResult = PollUnauthorized {} | PollDone {events: [PollEvent], cursor: Timestamp}
127+type PollResponse = {events: [PollEvent], cursor: Timestamp, authorized: Bool}
119-- ============================================
130-- ============================================
@@ -714,7 +725,7 @@ pollOnce = \token cursor -> do
714 let auth = authenticateSession sessions token t
725 let auth = authenticateSession sessions token t
715 case auth of
726 case auth of
716 Unauthorized {} ->
727 Unauthorized {} ->
717− yield PollUnauthorized {}
728+ yield {events: [], cursor: 0, authorized: False {}}
718 Authorized {pubkey} -> do
729 Authorized {pubkey} -> do
719 seqCounter <- *seqCounter
730 seqCounter <- *seqCounter
720 messages <- *messages
731 messages <- *messages
@@ -731,20 +742,19 @@ pollOnce = \token cursor -> do
731 encryptedBlob: m.encryptedBlob,
742 encryptedBlob: m.encryptedBlob,
732 timestamp: m.timestamp}}
743 timestamp: m.timestamp}}
733 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
744 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
734− yield PollDone {events: evts, cursor: lastReceivedAt}
745+ yield {events: evts, cursor: lastReceivedAt, authorized: True {}}
736−pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResult
747+pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResponse
737pollLoop = \token cursor deadline -> do
748pollLoop = \token cursor deadline -> do
738 result <- pollOnce token cursor
749 result <- pollOnce token cursor
739− case result of
750+ if not result.authorized then yield result
740− PollUnauthorized {} -> yield result
751+ else do
741− PollDone {events} -> do
742− t <- now
752+ t <- now
743− if count events == 0 && t < deadline
753+ if count result.events == 0 && t < deadline
745− sleep (clamp 1 500 (deadline - t))
755+ sleep (clamp 1 500 (deadline - t))
746− pollLoop token cursor deadline
756+ pollLoop token cursor deadline
747− else yield result
757+ else yield result
749-- ============================================
759-- ============================================
750-- Request Handler
760-- Request Handler
@@ -817,9 +827,7 @@ serve = \req -> case req of
817 result <- pollLoop token cursor deadline
827 result <- pollLoop token cursor deadline
818 tAfterPoll <- now
828 tAfterPoll <- now
819 maybePruneAndBroadcast tAfterPoll
829 maybePruneAndBroadcast tAfterPoll
820− yield (respond (case result of
830+ yield (respond result)
821− PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
822− PollDone r -> {events: r.events, cursor: r.cursor, authorized: True {}}))
824 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
832 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
@@ -849,9 +857,9 @@ serve = \req -> case req of
849 forwards <- *forwards
857 forwards <- *forwards
850 let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
858 let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
851 let shouldForward = federationEnabled && count forwardServers > 0
859 let shouldForward = federationEnabled && count forwardServers > 0
852− let status = if localOnline then MsDelivered {}
860+ let status = if localOnline then Delivered {}
853− else if shouldForward then MsFederated {}
861+ else if shouldForward then Federated {}
854− else MsQueued {}
862+ else Queued {}
855 set *forwards = if shouldForward
863 set *forwards = if shouldForward
856 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
864 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
857 else forwards
865 else forwards
@@ -867,14 +875,14 @@ serve = \req -> case req of
867 authorization: federationAuth})
875 authorization: federationAuth})
868 whenOk fwdResult (removeForward msgId server))
876 whenOk fwdResult (removeForward msgId server))
869 case status of
877 case status of
870− MsQueued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
878+ Queued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
871 _ -> yield {})
879 _ -> yield {})
872 _ -> yield {}
880 _ -> yield {}
873 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
881 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
874 yield (case result.send of
882 yield (case result.send of
875 SendUnauthorized {} -> respond {status: MsUnauthorized {}, messageId: ""}
883 SendUnauthorized {} -> respond {status: MsUnauthorized {}, messageId: ""}
876 SendRejected {} -> respond {status: MsRejected {}, messageId: ""}
884 SendRejected {} -> respond {status: MsRejected {}, messageId: ""}
877− SendDone {status} -> respond {status: status, messageId: msgId})
885+ SendDone {status} -> respond {status: deliveryToStatus status, messageId: msgId})
879 AckMessages {messageIds, authorization, respond} -> do
887 AckMessages {messageIds, authorization, respond} -> do
@@ -884,7 +892,7 @@ serve = \req -> case req of
884 auth <- authenticateRequest authorization t
892 auth <- authenticateRequest authorization t
885 case auth of
893 case auth of
886 Unauthorized {} ->
894 Unauthorized {} ->
887− yield {pr: pr, ack: AckUnauthorized {}}
895+ yield {pr: pr, authorized: False {}, pendingAcks: [], evictedEntries: []}
888 Authorized {pubkey} -> do
896 Authorized {pubkey} -> do
889 messages <- *messages
897 messages <- *messages
890 let ackedMessages = do
898 let ackedMessages = do
@@ -902,16 +910,13 @@ serve = \req -> case req of
902 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
910 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
903 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
911 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
904 set *ackedIds = diff combinedAckedIds evictedEntries
912 set *ackedIds = diff combinedAckedIds evictedEntries
905− yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
913+ yield {pr: pr, authorized: True {}, pendingAcks: ackedFederated, evictedEntries: evictedEntries}
908− AckDone {pendingAcks, evictedEntries} ->
909− when federationEnabled (do
915+ when (result.authorized && federationEnabled) (do
910− sendAcksAndClean pendingAcks
916+ sendAcksAndClean result.pendingAcks
911− retryEvictedAcks evictedEntries pendingAcks)
917+ retryEvictedAcks result.evictedEntries result.pendingAcks)
913 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
918 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
914− yield (respond {ok: case result.ack of AckDone _ -> True {}; _ -> False {}})
919+ yield (respond {ok: result.authorized})
916 LookupPresence {pubkey, authorization, respond} -> do
921 LookupPresence {pubkey, authorization, respond} -> do
@@ -920,7 +925,7 @@ serve = \req -> case req of
920 auth <- authenticateRequest authorization t
925 auth <- authenticateRequest authorization t
921 case auth of
926 case auth of
922 Unauthorized {} ->
927 Unauthorized {} ->
923− yield {pr: pr, lookup: LookupUnauthorized {}}
928+ yield {pr: pr, server: "", authorized: False {}}
924 Authorized _ -> do
929 Authorized _ -> do
925 presence <- *presence
930 presence <- *presence
926 let valid = do
931 let valid = do
@@ -933,11 +938,9 @@ serve = \req -> case req of
933 if not (isLocalServer p.server) && dominated then Just {value: p} else acc) Nothing {} valid
938 if not (isLocalServer p.server) && dominated then Just {value: p} else acc) Nothing {} valid
934 let foundServer = if localOnline then serverHostN
939 let foundServer = if localOnline then serverHostN
935 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
940 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
936− yield {pr: pr, lookup: LookupDone {server: foundServer}}
941+ yield {pr: pr, server: foundServer, authorized: True {}}
937 forkBroadcastOffline t result.pr
942 forkBroadcastOffline t result.pr
938− yield (case result.lookup of
943+ yield (respond {server: result.server, authorized: result.authorized})
939− LookupUnauthorized {} -> respond {server: "", authorized: False {}}
940− LookupDone {server} -> respond {server: server, authorized: True {}})
942 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
945 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -999,9 +1002,9 @@ serve = \req -> case req of
999 messages <- *messages
1002 messages <- *messages
1000 ackedIds <- *ackedIds
1003 ackedIds <- *ackedIds
1001 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1004 if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1002− then yield {pr: pr, forward: ForwardDuplicate {}}
1005+ then yield {pr: pr, ok: True {}}
1003 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1006 else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1004− then yield {pr: pr, forward: ForwardFull {}}
1007+ then yield {pr: pr, ok: False {}}
1006 seqCounter <- *seqCounter
1009 seqCounter <- *seqCounter
1007 let seqTs = nextSeqTs t seqCounter
1010 let seqTs = nextSeqTs t seqCounter
@@ -1009,9 +1012,9 @@ serve = \req -> case req of
1009 set *messages = union messages
1012 set *messages = union messages
1010 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1013 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1011 encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1014 encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1012− yield {pr: pr, forward: ForwardStored {}}
1015+ yield {pr: pr, ok: True {}}
1013 forkBroadcastOffline t result.pr
1016 forkBroadcastOffline t result.pr
1014− yield (respond {ok: case result.forward of ForwardFull {} -> False {}; _ -> True {}})
1017+ yield (respond {ok: result.ok})
1016 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1019 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1017 t <- now
1020 t <- now