@@ -58,7 +58,7 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,
59*forwards : [{messageId: Text, toServer: Text, createdAt: Int, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int}]
59*forwards : [{messageId: Text, toServer: Text, createdAt: Int, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int}]
61−*ackedIds : [{messageId: Text, toKey: Text, ackedAt: Int}]
61+*ackedIds : [{messageId: Text, toKey: Text, ackedAt: Int, originServer: Text}]
63*lastPrunedAt : Int
63*lastPrunedAt : Int
@@ -635,7 +635,8 @@ serve = \req -> case req of
635 else if seqCounter > clampedCursor then seqCounter else clampedCursor
635 else if seqCounter > clampedCursor then seqCounter else clampedCursor
636 yield {events: evts, cursor: newCursor, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
636 yield {events: evts, cursor: newCursor, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}
637 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
637 respond {events: result.events, cursor: result.cursor, authorized: result.authorized}
638− broadcastOfflineGossip t result.peerServers result.expiredLocal
639+ broadcastOfflineGossip tAfterPoll result.peerServers result.expiredLocal
640 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
641 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
641 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
642 let validInput = not (to == "") && not (encryptedBlob == "") && not (tooLong to maxPubkeyLen) && not (tooLong encryptedBlob maxBlobLen) && isValidHex to && isValidHex encryptedBlob
@@ -742,7 +743,7 @@ serve = \req -> case req of
742 m <- ackedMessages
743 m <- ackedMessages
743 where not (m.originServer == "")
744 where not (m.originServer == "")
744 where not (m.originServer == serverHostN)
745 where not (m.originServer == serverHostN)
745− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t}
746+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}
746 ackedIds <- *ackedIds
747 ackedIds <- *ackedIds
747 let combinedAckedIds = union ackedIds ackedFederated
748 let combinedAckedIds = union ackedIds ackedFederated
748 let oldestAckedAt = if count combinedAckedIds > maxAckedIds
749 let oldestAckedAt = if count combinedAckedIds > maxAckedIds
@@ -1133,17 +1134,35 @@ serve = \req -> case req of
1133 if authorized
1134 if authorized
1135 result <- atomic do
1136 result <- atomic do
1137+ set *lastPrunedAt = t
1136 pr <- pruneExpired t
1138 pr <- pruneExpired t
1137 forwards <- *forwards
1139 forwards <- *forwards
1138 let pendingForwards = do
1140 let pendingForwards = do
1139 f <- forwards
1141 f <- forwards
1140 yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1142 yield {toServer: f.toServer, toKey: f.toKey, fromKey: f.fromKey, encryptedBlob: f.encryptedBlob, timestamp: f.timestamp, messageId: f.messageId}
1143+ ackedIds <- *ackedIds
1144+ let pendingAcks = do
1146+ where not (a.originServer == "")
1147+ where not (a.originServer == serverHostN)
1141− yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards}
1149+ yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}
1142 respond {ok: True {}}
1150 respond {ok: True {}}
1143 broadcastOfflineGossip t result.peerServers result.expiredLocal
1151 broadcastOfflineGossip t result.peerServers result.expiredLocal
1144 if not (federationSecret == "") && count result.pendingForwards > 0
1152 if not (federationSecret == "") && count result.pendingForwards > 0
1145 then retryPendingForwards result.pendingForwards
1153 then retryPendingForwards result.pendingForwards
1146 else yield {}
1154 else yield {}
1155+ if not (federationSecret == "") && count result.pendingAcks > 0
1157+ let ackOriginServers = fold (\acc a ->
1158+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc
1159+ else union acc [a.originServer]) [] result.pendingAcks
1160+ sendToEach ackOriginServers (\server ->
1161+ fetch (federationScheme ++ server) (FederationAck {
1162+ authorization: "Bearer " ++ federationSecret,
1163+ messageIds: (do a <- result.pendingAcks; where a.originServer == server; yield a.messageId),
1164+ originServer: server, fromServer: serverHostN}))
1147 else respond {ok: False {}}
1166 else respond {ok: False {}}
1149-- ============================================
1168-- ============================================