@@ -87,20 +87,6 @@ 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 {}
104isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
90isValidBoundedHex = \maxLen s -> isValidHex s && not (tooLong s maxLen) && s == normalizeHex s
105type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
91type PubkeyHex = Text where isValidBoundedHex maxPubkeyLen
106type BlobHex = Text where isValidBoundedHex maxBlobLen
92type BlobHex = Text where isValidBoundedHex maxBlobLen
@@ -122,8 +108,10 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
122type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
108type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
123type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
109type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
124data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
110data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
125−data SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
111+data SendResult = SendUnauthorized {} | SendRejected {} | SendDone {status: MessageStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
126data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
112data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
113+data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
114+data LookupResult = LookupUnauthorized {} | LookupDone {server: AnyServerName}
127type PollResponse = {events: [PollEvent], cursor: Timestamp, authorized: Bool}
115type PollResponse = {events: [PollEvent], cursor: Timestamp, authorized: Bool}
@@ -621,6 +609,11 @@ broadcastOfflineGossip = \t peerServers expiredLocal ->
621forkBroadcastOffline = \t result ->
609forkBroadcastOffline = \t result ->
622 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
610 fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)
612+effectiveGossipEvents = \entries appliedKeys eventType -> do
614+ where any appliedKeys (\a -> samePubkeyServer a e)
615+ yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
624rejectFederation = \t respond -> do
617rejectFederation = \t respond -> do
625 maybePruneAndBroadcast t
618 maybePruneAndBroadcast t
626 yield (respond {ok: False {}})
619 yield (respond {ok: False {}})
@@ -857,16 +850,16 @@ serve = \req -> case req of
857 forwards <- *forwards
850 forwards <- *forwards
858 let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
851 let forwardServers = takeN remoteServers (maxForwardsEntries - count forwards)
859 let shouldForward = federationEnabled && count forwardServers > 0
852 let shouldForward = federationEnabled && count forwardServers > 0
860− let status = if localOnline then Delivered {}
853+ let status = if localOnline then MsDelivered {}
861− else if shouldForward then Federated {}
854+ else if shouldForward then MsFederated {}
862− else Queued {}
855+ else MsQueued {}
863 set *forwards = if shouldForward
856 set *forwards = if shouldForward
864 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
857 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
865 else forwards
858 else forwards
866 yield {pr: pr, send: SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey}}
859 yield {pr: pr, send: SendDone {status: status, remoteServers: remoteServers, fromKey: fromKey}}
868− case result.send of
860+ case result.send of
869− SendDone {status, remoteServers, fromKey} ->
861+ SendDone {status, remoteServers, fromKey} ->
870 when (count remoteServers > 0 && federationEnabled) (do
863 when (count remoteServers > 0 && federationEnabled) (do
871 sendToEach remoteServers (\server -> do
864 sendToEach remoteServers (\server -> do
872 fwdResult <- fetch (federationUrl server) (ForwardMessage {
865 fwdResult <- fetch (federationUrl server) (ForwardMessage {
@@ -875,14 +868,14 @@ serve = \req -> case req of
875 authorization: federationAuth})
868 authorization: federationAuth})
876 whenOk fwdResult (removeForward msgId server))
869 whenOk fwdResult (removeForward msgId server))
877 case status of
870 case status of
878− Queued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
871+ MsQueued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
879 _ -> yield {})
872 _ -> yield {})
881− broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
873+ broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
874+ _ -> forkBroadcastOffline t result.pr
882 yield (case result.send of
875 yield (case result.send of
883 SendUnauthorized {} -> respond {status: MsUnauthorized {}, messageId: ""}
876 SendUnauthorized {} -> respond {status: MsUnauthorized {}, messageId: ""}
884 SendRejected {} -> respond {status: MsRejected {}, messageId: ""}
877 SendRejected {} -> respond {status: MsRejected {}, messageId: ""}
885− SendDone {status} -> respond {status: deliveryToStatus status, messageId: msgId})
878+ SendDone {status} -> respond {status: status, messageId: msgId})
887 AckMessages {messageIds, authorization, respond} -> do
880 AckMessages {messageIds, authorization, respond} -> do
@@ -892,7 +885,7 @@ serve = \req -> case req of
892 auth <- authenticateRequest authorization t
885 auth <- authenticateRequest authorization t
893 case auth of
886 case auth of
894 Unauthorized {} ->
887 Unauthorized {} ->
895− yield {pr: pr, authorized: False {}, pendingAcks: [], evictedEntries: []}
888+ yield {pr: pr, ack: AckUnauthorized {}}
896 Authorized {pubkey} -> do
889 Authorized {pubkey} -> do
897 messages <- *messages
890 messages <- *messages
898 let ackedMessages = do
891 let ackedMessages = do
@@ -910,13 +903,16 @@ serve = \req -> case req of
910 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
903 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
911 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
904 if excessCount <= 0 then [] else boundedSmallest combinedAckedIds excessCount (\a -> a.ackedAt)
912 set *ackedIds = diff combinedAckedIds evictedEntries
905 set *ackedIds = diff combinedAckedIds evictedEntries
913− yield {pr: pr, authorized: True {}, pendingAcks: ackedFederated, evictedEntries: evictedEntries}
906+ yield {pr: pr, ack: AckDone {pendingAcks: ackedFederated, evictedEntries: evictedEntries}}
909+ AckDone {pendingAcks, evictedEntries} ->
915− when (result.authorized && federationEnabled) (do
910+ when federationEnabled (do
916− sendAcksAndClean result.pendingAcks
911+ sendAcksAndClean pendingAcks
917− retryEvictedAcks result.evictedEntries result.pendingAcks)
912+ retryEvictedAcks evictedEntries pendingAcks)
913+ AckUnauthorized {} -> yield {}
918 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
914 broadcastOfflineGossip t result.pr.peerServers result.pr.expiredLocal)
919− yield (respond {ok: result.authorized})
915+ yield (respond {ok: case result.ack of AckDone _ -> True {}; _ -> False {}})
921 LookupPresence {pubkey, authorization, respond} -> do
917 LookupPresence {pubkey, authorization, respond} -> do
@@ -925,7 +921,7 @@ serve = \req -> case req of
925 auth <- authenticateRequest authorization t
921 auth <- authenticateRequest authorization t
926 case auth of
922 case auth of
927 Unauthorized {} ->
923 Unauthorized {} ->
928− yield {pr: pr, server: "", authorized: False {}}
924+ yield {pr: pr, lookup: LookupUnauthorized {}}
929 Authorized _ -> do
925 Authorized _ -> do
930 presence <- *presence
926 presence <- *presence
931 let valid = do
927 let valid = do
@@ -938,9 +934,11 @@ serve = \req -> case req of
938 if not (isLocalServer p.server) && dominated then Just {value: p} else acc) Nothing {} valid
934 if not (isLocalServer p.server) && dominated then Just {value: p} else acc) Nothing {} valid
939 let foundServer = if localOnline then serverHostN
935 let foundServer = if localOnline then serverHostN
940 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
936 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
941− yield {pr: pr, server: foundServer, authorized: True {}}
937+ yield {pr: pr, lookup: LookupDone {server: foundServer}}
942 forkBroadcastOffline t result.pr
938 forkBroadcastOffline t result.pr
939+ yield (case result.lookup of
943− yield (respond {server: result.server, authorized: result.authorized})
940+ LookupUnauthorized {} -> respond {server: "", authorized: False {}}
941+ LookupDone {server} -> respond {server: server, authorized: True {}})
945 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
943 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -973,11 +971,7 @@ serve = \req -> case req of
973 set *presence = capResult.finalPresence
971 set *presence = capResult.finalPresence
974 let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
972 let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
975 yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: capResult.withinCapacity}
973 yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: capResult.withinCapacity}
976− let effectiveFor = \entries eventType -> do
978− where any gossipResult.appliedKeys (\a -> samePubkeyServer a e)
979− yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
980− let effectiveEvents = union (effectiveFor filteredOnline (GossipOnline {})) (effectiveFor filteredOffline (GossipOffline {}))
974+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.appliedKeys (GossipOffline {}))
981 when (not gossipResult.withinCapacity)
975 when (not gossipResult.withinCapacity)
982 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
976 (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))