@@ -110,10 +110,11 @@ type PresenceEntry = {pubkey: PubkeyHex, server: AnyServerName, expiresAt: Times
110type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
110type ForwardEntry = {messageId: MessageIdHex, toServer: ServerName, createdAt: Timestamp, toKey: PubkeyHex, fromKey: PubkeyHex, encryptedBlob: BlobHex, timestamp: Timestamp}
111type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
111type AckedEntry = {messageId: MessageIdHex, toKey: PubkeyHex, ackedAt: Timestamp, originServer: ServerName}
112data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
112data AuthResult = Authorized {pubkey: PubkeyHex} | Unauthorized {}
113−data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], fromKey: PubkeyHex}
113+data SendResult = SendFailed {status: MessageStatus} | SendDone {delivery: DeliveryStatus, remoteServers: [ServerName], forwardServers: [ServerName], fromKey: PubkeyHex}
114data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
114data VerifyResult = VerifyFailed {} | VerifyOk {peerServers: [ServerName]}
115data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
115data AckResult = AckUnauthorized {} | AckDone {pendingAcks: [AckedEntry], evictedEntries: [AckedEntry]}
116−type PollResponse = {events: [PollEvent], cursor: Timestamp, authorized: Bool}
116+data PollResult = PollUnauthorized {} | PollOk {events: [PollEvent], cursor: Timestamp}
117+data ForwardResult = ForwardDuplicate {} | ForwardRejected {} | ForwardStored {}
119-- ============================================
120-- ============================================
@@ -200,6 +201,10 @@ headOf = \items -> fold (\acc x -> case acc of
200 Nothing {} -> Just {value: x}
201 Nothing {} -> Just {value: x}
201 _ -> acc) Nothing {} items
202 _ -> acc) Nothing {} items
204+maxBy = \items getKey -> fold (\acc x -> case acc of
205+ Nothing {} -> Just {value: x}
206+ Just {value: v} -> if getKey x > getKey v then Just {value: x} else acc) Nothing {} items
203hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
208hexDigit = \n -> take 1 (drop (clamp 0 15 n) "0123456789abcdef")
205byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
210byteToHex = \n -> hexDigit (n / 16) ++ hexDigit (n - (n / 16) * 16)
@@ -220,7 +225,7 @@ genToken = genTokenN 8
221tooLong = \s maxLen -> drop maxLen s != ""
226tooLong = \s maxLen -> drop maxLen s != ""
223−isHexChar = \c -> contains c "0123456789abcdefABCDEF"
228+isHexChar = \c -> contains c "0123456789abcdef"
225isValidHexPairs = \s ->
230isValidHexPairs = \s ->
226 if s == "" then True {}
231 if s == "" then True {}
@@ -352,10 +357,6 @@ authenticateSession = \sessions token t ->
352 Nothing {} -> Unauthorized {}
357 Nothing {} -> Unauthorized {}
353 Just {value: s} -> Authorized {pubkey: s.pubkey}
358 Just {value: s} -> Authorized {pubkey: s.pubkey}
355−authenticateRequest = \authorization t -> do
356− sessions <- *sessions
357− yield (authenticateSession sessions (extractToken authorization) t)
359extractToken = \authHeader ->
360extractToken = \authHeader ->
360 let token = drop 7 authHeader in
361 let token = drop 7 authHeader in
361 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
362 if toLower (take 7 authHeader) == "bearer " && not (tooLong token maxTokenLen)
@@ -621,10 +622,27 @@ effectiveGossipEvents = \entries appliedKeys eventType -> do
621 where any appliedKeys (\a -> samePubkeyServer a e)
622 where any appliedKeys (\a -> samePubkeyServer a e)
622 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
623 yield {eventType: eventType, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp, ttl: e.ttl}
625+mergeGossipPresence = \presence deduped trustedOffline fromServer t ->
626+ let trustedDeduped = do e <- deduped; where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t); yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp} in
627+ let appliedOnline = do e <- trustedDeduped; where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt); yield (pubkeyServer e) in
628+ let merged = mergeByPubkeyServer presence trustedDeduped in
629+ let afterOffline = removeOffline merged trustedOffline t in
630+ let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p) in
631+ let capResult = capGossipPresence {afterOffline: afterOffline, appliedOnline: appliedOnline, trustedDeduped: trustedDeduped, presence: presence, trustedOffline: trustedOffline, t: t} in
632+ {finalPresence: capResult.finalPresence,
633+ appliedKeys: union capResult.appliedOnline appliedOffline,
634+ gossipPeerServers: collectPeerServers capResult.finalPresence (\p -> p.server),
635+ withinCapacity: capResult.withinCapacity}
624rejectFederation = \t respond -> do
637rejectFederation = \t respond -> do
625 maybePruneAndBroadcast t
638 maybePruneAndBroadcast t
626 yield (respond {ok: False {}})
639 yield (respond {ok: False {}})
641+withFederationAuth = \authorization t respond extraCheck action ->
642+ if not (verifyFederationAuth authorization) || extraCheck
643+ then rejectFederation t respond
628maybePruneAndBroadcast = \t -> do
646maybePruneAndBroadcast = \t -> do
629 pruneResult <- atomic do maybePrune t
647 pruneResult <- atomic do maybePrune t
630 forkBroadcastOffline t pruneResult
648 forkBroadcastOffline t pruneResult
@@ -658,13 +676,16 @@ retryPendingForwards = \pending -> do
658 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
676 retryForwardsForServer (do x <- pending; where x.toServer == server; yield x) maxForwardRetriesPerServer)
679+fetchFederationAck = \server messageIds ->
680+ fetch (federationUrl server) (FederationAck {
681+ authorization: federationAuth,
682+ messageIds: messageIds,
683+ originServer: server, fromServer: serverHostN})
661sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
685sendBatchedAcksToServer : [{messageId: MessageIdHex, toKey: PubkeyHex, originServer: ServerName, | a}] -> ServerName -> IO {network} {}
662sendBatchedAcksToServer = \serverAcks server ->
686sendBatchedAcksToServer = \serverAcks server ->
663 forEachBatch serverAcks maxMessageIds (\batch -> do
687 forEachBatch serverAcks maxMessageIds (\batch -> do
664− result <- fetch (federationUrl server) (FederationAck {
665− authorization: federationAuth,
666− messageIds: (do a <- batch; yield a.messageId),
688+ result <- fetchFederationAck server (do a <- batch; yield a.messageId)
667− originServer: server, fromServer: serverHostN})
668 whenOk result (atomic do
689 whenOk result (atomic do
669 ackedIds <- *ackedIds
690 ackedIds <- *ackedIds
670 set *ackedIds = do
691 set *ackedIds = do
@@ -686,10 +707,7 @@ retryEvictedAcks = \evictedEntries pendingAcks ->
687 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
708 let evictedOriginServers = collectPeerServers evictedToRetry (\a -> a.originServer)
688 sendToEach evictedOriginServers (\server -> do
709 sendToEach evictedOriginServers (\server -> do
689− fetchResult <- fetch (federationUrl server) (FederationAck {
690− authorization: federationAuth,
691− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),
710+ fetchResult <- fetchFederationAck server (do a <- evictedToRetry; where a.originServer == server; yield a.messageId)
692− originServer: server, fromServer: serverHostN})
693 case fetchResult of
711 case fetchResult of
694 Err _ -> do
712 Err _ -> do
695 dropped <- atomic do
713 dropped <- atomic do
@@ -725,7 +743,7 @@ pollOnce = \token cursor -> do
725 let auth = authenticateSession sessions token t
743 let auth = authenticateSession sessions token t
726 case auth of
744 case auth of
727 Unauthorized {} ->
745 Unauthorized {} ->
728− yield {events: [], cursor: 0, authorized: False {}}
746+ yield (PollUnauthorized {})
729 Authorized {pubkey} -> do
747 Authorized {pubkey} -> do
730 seqCounter <- *seqCounter
748 seqCounter <- *seqCounter
731 messages <- *messages
749 messages <- *messages
@@ -742,19 +760,20 @@ pollOnce = \token cursor -> do
742 encryptedBlob: m.encryptedBlob,
760 encryptedBlob: m.encryptedBlob,
743 timestamp: m.timestamp}}
761 timestamp: m.timestamp}}
744 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
762 let lastReceivedAt = fold (\acc m -> max acc m.receivedAt) clampedCursor limitedMsgs
745− yield {events: evts, cursor: lastReceivedAt, authorized: True {}}
763+ yield (PollOk {events: evts, cursor: lastReceivedAt})
747−pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResponse
765+pollLoop : Text -> Int -> Int<Ms> -> IO {clock} PollResult
748pollLoop = \token cursor deadline -> do
766pollLoop = \token cursor deadline -> do
749 result <- pollOnce token cursor
767 result <- pollOnce token cursor
750− if not result.authorized then yield result
768+ case result of
751− else do
769+ PollUnauthorized {} -> yield result
770+ PollOk {events} -> do
752− t <- now
771+ t <- now
753− if count result.events == 0 && t < deadline
772+ if count events == 0 && t < deadline
755− sleep (clamp 1 500 (deadline - t))
774+ sleep (clamp 1 500 (deadline - t))
756− pollLoop token cursor deadline
775+ pollLoop token cursor deadline
757− else yield result
776+ else yield result
759-- ============================================
778-- ============================================
760-- Request Handler
779-- Request Handler
@@ -827,7 +846,9 @@ serve = \req -> case req of
827 result <- pollLoop token cursor deadline
846 result <- pollLoop token cursor deadline
828 tAfterPoll <- now
847 tAfterPoll <- now
829 maybePruneAndBroadcast tAfterPoll
848 maybePruneAndBroadcast tAfterPoll
830− yield (respond result)
849+ yield (respond (case result of
850+ PollUnauthorized {} -> {events: [], cursor: 0, authorized: False {}}
851+ PollOk {events: evts, cursor: c} -> {events: evts, cursor: c, authorized: True {}}))
832 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
853 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
@@ -835,7 +856,8 @@ serve = \req -> case req of
835 let clampedTs = clampTimestamp timestamp t
856 let clampedTs = clampTimestamp timestamp t
836 result <- atomic do
857 result <- atomic do
837 pr <- maybePrune t
858 pr <- maybePrune t
838− auth <- authenticateRequest authorization t
859+ sessions <- *sessions
860+ let auth = authenticateSession sessions (extractToken authorization) t
839 case auth of
861 case auth of
840 Unauthorized {} ->
862 Unauthorized {} ->
841 yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}}
863 yield {pr: pr, send: SendFailed {status: MsUnauthorized {}}}
@@ -863,9 +885,9 @@ serve = \req -> case req of
863 set *forwards = if shouldForward
885 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})
886 then union forwards (do s <- forwardServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})
865 else forwards
887 else forwards
866− yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, fromKey: fromKey}}
888+ yield {pr: pr, send: SendDone {delivery: delivery, remoteServers: remoteServers, forwardServers: forwardServers, fromKey: fromKey}}
867 case result.send of
889 case result.send of
868− SendDone {delivery, remoteServers, fromKey} ->
890+ SendDone {delivery, remoteServers, forwardServers, fromKey} ->
870 when (count remoteServers > 0 && federationEnabled) (do
892 when (count remoteServers > 0 && federationEnabled) (do
871 sendToEach remoteServers (\server -> do
893 sendToEach remoteServers (\server -> do
@@ -873,7 +895,8 @@ serve = \req -> case req of
873 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
895 toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob,
874 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
896 timestamp: clampedTs, originServer: serverHostN, messageId: msgId,
875 authorization: federationAuth})
897 authorization: federationAuth})
898+ when (any forwardServers (\s -> s == server))
876− whenOk fwdResult (removeForward msgId server))
899+ (whenOk fwdResult (removeForward msgId server)))
877 case delivery of
900 case delivery of
878 DsQueued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
901 DsQueued {} -> println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")
879 _ -> yield {})
902 _ -> yield {})
@@ -888,7 +911,8 @@ serve = \req -> case req of
888 let dedupedIds = dedupMessageIds messageIds
911 let dedupedIds = dedupMessageIds messageIds
889 result <- atomic do
912 result <- atomic do
890 pr <- maybePrune t
913 pr <- maybePrune t
891− auth <- authenticateRequest authorization t
914+ sessions <- *sessions
915+ let auth = authenticateSession sessions (extractToken authorization) t
892 case auth of
916 case auth of
893 Unauthorized {} ->
917 Unauthorized {} ->
894 yield {pr: pr, ack: AckUnauthorized {}}
918 yield {pr: pr, ack: AckUnauthorized {}}
@@ -903,7 +927,7 @@ serve = \req -> case req of
903 m <- ackedMessages
927 m <- ackedMessages
904 case m.origin of
928 case m.origin of
905 FederatedOrigin {server} -> yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: server}
929 FederatedOrigin {server} -> yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: server}
906− _ -> []
930+ LocalOrigin {} -> []
907 ackedIds <- *ackedIds
931 ackedIds <- *ackedIds
908 let combinedAckedIds = union ackedIds ackedFederated
932 let combinedAckedIds = union ackedIds ackedFederated
909 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
933 let evictedEntries = let excessCount = count combinedAckedIds - maxAckedIds in
@@ -924,7 +948,8 @@ serve = \req -> case req of
925 result <- atomic do
949 result <- atomic do
926 pr <- maybePrune t
950 pr <- maybePrune t
927− auth <- authenticateRequest authorization t
951+ sessions <- *sessions
952+ let auth = authenticateSession sessions (extractToken authorization) t
928 case auth of
953 case auth of
929 Unauthorized {} ->
954 Unauthorized {} ->
930 yield {pr: pr, server: "", authorized: False {}}
955 yield {pr: pr, server: "", authorized: False {}}
@@ -936,9 +961,7 @@ serve = \req -> case req of
937 let localOnline = any valid (\p -> isLocalServer p.server)
962 let localOnline = any valid (\p -> isLocalServer p.server)
938 let remoteValid = do p <- valid; where not (isLocalServer p.server); yield p
963 let remoteValid = do p <- valid; where not (isLocalServer p.server); yield p
939− let bestRemote = fold (\acc p -> case acc of
964+ let bestRemote = maxBy remoteValid (\p -> p.expiresAt)
940− Nothing {} -> Just {value: p}
941− Just {value: v} -> if p.expiresAt > v.expiresAt then Just {value: p} else acc) Nothing {} remoteValid
942 let foundServer = if localOnline then serverHostN
965 let foundServer = if localOnline then serverHostN
943 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
966 else case bestRemote of Just {value: p} -> p.server; Nothing {} -> ""
944 yield {pr: pr, server: foundServer, authorized: True {}}
967 yield {pr: pr, server: foundServer, authorized: True {}}
@@ -947,91 +970,73 @@ serve = \req -> case req of
948 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
971 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
950− let authorized = verifyFederationAuth authorization
951 let cappedHops = clamp 0 maxGossipHops hopsRemaining
973 let cappedHops = clamp 0 maxGossipHops hopsRemaining
952− if not authorized || cappedHops < 1 || isLocalServer fromServer || count events > maxGossipEvents
974+ withFederationAuth authorization t respond (cappedHops < 1 || isLocalServer fromServer || count events > maxGossipEvents) (do
953− then rejectFederation t respond
955− let validated = validGossipEvents events t
975+ let validated = validGossipEvents events t
956− let onlineEntries = validated.online
976+ let onlineEntries = validated.online
957− let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e
977+ let trustedOffline = do e <- validated.offline; where e.server == fromServer; yield e
958− let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
978+ let filteredOnline = resolveByTimestamp onlineEntries trustedOffline
959− let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
979+ let filteredOffline = resolveByTimestamp trustedOffline onlineEntries
960− let deduped = mergeByPubkeyServer [] filteredOnline
980+ let deduped = mergeByPubkeyServer [] filteredOnline
961− gossipResult <- atomic do
981+ gossipResult <- atomic do
962− pr <- maybePrune t
982+ pr <- maybePrune t
963− presence <- *presence
983+ presence <- *presence
964− let trustedDeduped = do
984+ let gm = mergeGossipPresence presence deduped trustedOffline fromServer t
966− where e.server == fromServer || any presence (\p -> p.server == e.server && p.expiresAt > t)
967− yield {expiresAt: e.expiresAt, pubkey: e.pubkey, server: e.server, timestamp: e.timestamp}
968− let appliedOnline = do
970− where none presence (\p -> samePubkeyServer p e && p.expiresAt >= e.expiresAt)
971− yield (pubkeyServer e)
972− let merged = mergeByPubkeyServer presence trustedDeduped
973− let afterOffline = removeOffline merged trustedOffline t
974− let appliedOffline = do p <- diff merged afterOffline; yield (pubkeyServer p)
975− let capResult = capGossipPresence {afterOffline: afterOffline, appliedOnline: appliedOnline, trustedDeduped: trustedDeduped, presence: presence, trustedOffline: trustedOffline, t: t}
976− set *presence = capResult.finalPresence
985+ set *presence = gm.finalPresence
977− let gossipPeerServers = collectPeerServers capResult.finalPresence (\p -> p.server)
986+ yield {gm: gm, pr: pr}
978− yield {appliedKeys: union capResult.appliedOnline appliedOffline, pr: pr, gossipPeerServers: gossipPeerServers, withinCapacity: capResult.withinCapacity}
979− let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.appliedKeys (GossipOffline {}))
987+ let effectiveEvents = union (effectiveGossipEvents filteredOnline gossipResult.gm.appliedKeys (GossipOnline {})) (effectiveGossipEvents filteredOffline gossipResult.gm.appliedKeys (GossipOffline {}))
980− when (not gossipResult.withinCapacity)
988+ when (not gossipResult.gm.withinCapacity)
981− (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
989+ (println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer))
982− fork (do
990+ fork (do
983− when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
991+ when (cappedHops > 1 && federationEnabled && count effectiveEvents > 0) (
984− let targets = do
992+ let targets = do
985− s <- gossipResult.gossipPeerServers
993+ s <- gossipResult.gm.gossipPeerServers
986− where s != fromServer
994+ where s != fromServer
988− in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
996+ in sendBatchedGossip targets effectiveEvents (cappedHops - 1))
989− broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal)
997+ broadcastOfflineGossip t gossipResult.pr.peerServers gossipResult.pr.expiredLocal)
990− yield (respond {ok: True {}})
998+ yield (respond {ok: True {}}))
992 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
1000 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do
993 t <- now
1001 t <- now
994− let authorized = verifyFederationAuth authorization
995− if not authorized || isLocalServer originServer
1002+ withFederationAuth authorization t respond (isLocalServer originServer) (do
996− then rejectFederation t respond
998− let clampedTs = clampTimestamp timestamp t
1003+ let clampedTs = clampTimestamp timestamp t
999− result <- atomic do
1004+ result <- atomic do
1000− pr <- maybePrune t
1005+ pr <- maybePrune t
1001− messages <- *messages
1006+ messages <- *messages
1002− ackedIds <- *ackedIds
1007+ ackedIds <- *ackedIds
1003− if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1008+ if hasMessage messageId toKey messages || hasMessage messageId toKey ackedIds
1004− then yield {pr: pr, ok: True {}}
1009+ then yield {pr: pr, forward: ForwardDuplicate {}}
1005− else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1010+ else if countMessagesFor messages toKey t >= maxMessagesPerRecipient
1006− then yield {pr: pr, ok: False {}}
1011+ then yield {pr: pr, forward: ForwardRejected {}}
1007− else do
1012+ else do
1008− seqCounter <- *seqCounter
1013+ seqCounter <- *seqCounter
1009− let seqTs = nextSeqTs t seqCounter
1014+ let seqTs = nextSeqTs t seqCounter
1010− set *seqCounter = seqTs
1015+ set *seqCounter = seqTs
1011− set *messages = union messages
1016+ set *messages = union messages
1012− [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1017+ [{messageId: messageId, toKey: toKey, fromKey: fromKey,
1013− encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1018+ encryptedBlob: encryptedBlob, timestamp: clampedTs, origin: FederatedOrigin {server: originServer}, receivedAt: seqTs}]
1014− yield {pr: pr, ok: True {}}
1019+ yield {pr: pr, forward: ForwardStored {}}
1015− forkBroadcastOffline t result.pr
1020+ forkBroadcastOffline t result.pr
1016− yield (respond {ok: result.ok})
1021+ yield (respond {ok: case result.forward of
1022+ ForwardRejected {} -> False {}
1023+ ForwardDuplicate {} -> True {}
1024+ ForwardStored {} -> True {}}))
1018 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1026 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do
1019 t <- now
1027 t <- now
1020− let authorized = verifyFederationAuth authorization
1021− if not authorized || not (isLocalServer originServer) || isLocalServer fromServer
1028+ withFederationAuth authorization t respond (not (isLocalServer originServer) || isLocalServer fromServer) (do
1022− then rejectFederation t respond
1024− let dedupedIds = dedupMessageIds messageIds
1029+ let dedupedIds = dedupMessageIds messageIds
1025− pruneResult <- atomic do
1030+ pruneResult <- atomic do
1026− pr <- maybePrune t
1031+ pr <- maybePrune t
1027− forwards <- *forwards
1032+ forwards <- *forwards
1028− set *forwards = do
1033+ set *forwards = do
1029− f <- forwards
1034+ f <- forwards
1030− where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1035+ where not (f.toServer == fromServer && any dedupedIds (\mid -> f.messageId == mid))
1031− yield f
1036+ yield f
1032− yield pr
1037+ yield pr
1033− forkBroadcastOffline t pruneResult
1038+ forkBroadcastOffline t pruneResult
1034− yield (respond {ok: True {}})
1039+ yield (respond {ok: True {}}))
1036 Prune {authorization, respond} -> do
1041 Prune {authorization, respond} -> do
1037 t <- now
1042 t <- now
@@ -1047,7 +1052,7 @@ backgroundPrune = do
1047 sleep backgroundPruneInterval
1052 sleep backgroundPruneInterval
1048 t <- now
1053 t <- now
1049 pruneAndRetry t
1054 pruneAndRetry t
1050− fork backgroundPrune
1055+ backgroundPrune
1052main = do
1057main = do
1053 println "Skrepka relay server starting on port 8080..."
1058 println "Skrepka relay server starting on port 8080..."