@@ -27,7 +27,7 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,
28*sessions : [{token: Text, pubkey: Text, expiresAt: Int}]
28*sessions : [{token: Text, pubkey: Text, expiresAt: Int}]
30−*messages : [{messageId: Text, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int}]
30+*messages : [{messageId: Text, toKey: Text, fromKey: Text, encryptedBlob: Text, timestamp: Int, originServer: Text}]
32*presence : [{pubkey: Text, server: Text, expiresAt: Int}]
32*presence : [{pubkey: Text, server: Text, expiresAt: Int}]
@@ -101,6 +101,23 @@ verifyFederationAuth = \authorization ->
102clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
102clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
104+forwardToAll = \servers path headers body ->
105+ if count servers == 0
108+ let server = fold (\acc s -> if acc == "" then s else acc) "" servers
109+ result <- httpPost ("http://" ++ server ++ path) headers body
110+ restOk <- forwardToAll (do s <- servers; where not (s == server); yield s) path headers body
111+ yield (if result.ok then True {} else restOk)
113+sendToEach = \servers action ->
114+ if count servers == 0
117+ let server = fold (\acc s -> if acc == "" then s else acc) "" servers
119+ sendToEach (do s <- servers; where not (s == server); yield s) action
104pruneExpired = \t -> do
121pruneExpired = \t -> do
105 challenges <- *challenges
122 challenges <- *challenges
106 set *challenges = do
123 set *challenges = do
@@ -164,10 +181,7 @@ serve = \req -> case req of
164 where not (c.pubkey == pubkey && c.challenge == challenge)
181 where not (c.pubkey == pubkey && c.challenge == challenge)
166 set *sessions = if sigValid
183 set *sessions = if sigValid
169− where not (s.pubkey == pubkey)
170− yield s) [{token: token, pubkey: pubkey, expiresAt: expiry}]
184+ then union sessions [{token: token, pubkey: pubkey, expiresAt: expiry}]
171 else sessions
185 else sessions
172 set *presence = if sigValid
186 set *presence = if sigValid
173 then union
187 then union
@@ -209,7 +223,7 @@ serve = \req -> case req of
209 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
223 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
211 msgId <- genToken
225 msgId <- genToken
212− let clampedTs = if timestamp > t then t else timestamp
226+ let clampedTs = if timestamp > t then t else if timestamp < t - messageMaxAge then t - messageMaxAge else timestamp
213 result <- atomic do
227 result <- atomic do
214 pruneExpired t
228 pruneExpired t
215 sessions <- *sessions
229 sessions <- *sessions
@@ -220,7 +234,7 @@ serve = \req -> case req of
220 Nothing {} -> messages
234 Nothing {} -> messages
221 Just {value: s} -> union messages
235 Just {value: s} -> union messages
222 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
236 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
223− encryptedBlob: encryptedBlob, timestamp: clampedTs}]
237+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: ""}]
224 let localOnline = case session of
238 let localOnline = case session of
225 Nothing {} -> False {}
239 Nothing {} -> False {}
226 Just {value: _} -> count (do
240 Just {value: _} -> count (do
@@ -229,51 +243,71 @@ serve = \req -> case req of
229 where p.server == serverHost
243 where p.server == serverHost
230 where p.expiresAt > t
244 where p.expiresAt > t
231 yield p) > 0
245 yield p) > 0
232− let remoteServer = case session of
246+ let remoteServers = case session of
233− Nothing {} -> ""
247+ Nothing {} -> []
234− Just {value: _} -> fold (\_ p -> p.server) "" (do
248+ Just {value: _} -> do
235 p <- presence
249 p <- presence
236 where p.pubkey == to
250 where p.pubkey == to
237 where not (p.server == serverHost)
251 where not (p.server == serverHost)
238 where p.expiresAt > t
252 where p.expiresAt > t
239− yield p)
253+ yield p.server
240 let fromKey = case session of
254 let fromKey = case session of
241 Nothing {} -> ""
255 Nothing {} -> ""
242 Just {value: s} -> s.pubkey
256 Just {value: s} -> s.pubkey
243 let status = case session of
257 let status = case session of
244 Nothing {} -> "unauthorized"
258 Nothing {} -> "unauthorized"
245 _ -> if localOnline then "delivered"
259 _ -> if localOnline then "delivered"
246− else if not (remoteServer == "") then "federated"
260+ else if count remoteServers > 0 then "federated"
247 else "queued"
261 else "queued"
248− yield {status: status, remoteServer: remoteServer, fromKey: fromKey}
262+ yield {status: status, remoteServers: remoteServers, fromKey: fromKey}
249− if result.status == "federated" && not (result.remoteServer == "") && not (federationSecret == "")
263+ finalStatus <- if result.status == "federated" && not (federationSecret == "")
250− then httpPost ("http://" ++ result.remoteServer ++ "/federation/forward")
265+ anyOk <- forwardToAll result.remoteServers "/federation/forward"
251− {authorization: "Bearer " ++ federationSecret}
266+ {authorization: "Bearer " ++ federationSecret}
252− {toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
267+ {toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
253− timestamp: clampedTs, originServer: serverHost, messageId: msgId}
268+ timestamp: clampedTs, originServer: serverHost, messageId: msgId}
269+ yield (if anyOk then "federated" else "queued")
254− else yield {}
270+ else yield result.status
255− respond {status: result.status, messageId: (if result.status == "unauthorized" then "" else msgId)}
271+ respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else msgId)}
257 AckMessages {messageIds, authorization, respond} -> do
273 AckMessages {messageIds, authorization, respond} -> do
259− ok <- atomic do
275+ result <- atomic do
260 pruneExpired t
276 pruneExpired t
261 sessions <- *sessions
277 sessions <- *sessions
262 messages <- *messages
278 messages <- *messages
263 let session = lookupSession sessions (extractToken authorization) t
279 let session = lookupSession sessions (extractToken authorization) t
264− set *messages = case session of
280+ let ackedMessages = case session of
265− Nothing {} -> messages
281+ Nothing {} -> []
266− Just {value: s} -> diff messages (do
282+ Just {value: s} -> do
267 m <- messages
283 m <- messages
268 mid <- messageIds
284 mid <- messageIds
269 where m.messageId == mid
285 where m.messageId == mid
270 where m.toKey == s.pubkey
286 where m.toKey == s.pubkey
271− yield m)
287+ yield m
288+ set *messages = case session of
289+ Nothing {} -> messages
290+ Just {value: _} -> diff messages ackedMessages
272 let ok = case session of
291 let ok = case session of
273 Nothing {} -> False {}
292 Nothing {} -> False {}
274 _ -> True {}
293 _ -> True {}
275− yield ok
294+ yield {ok: ok, ackedMessages: ackedMessages}
295+ let fedMessages = do
296+ m <- result.ackedMessages
297+ where not (m.originServer == "")
298+ where not (m.originServer == serverHost)
300+ let originServers = fold (\acc m ->
301+ if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc
302+ else union acc [m.originServer]) [] fedMessages
303+ if result.ok && count originServers > 0 && not (federationSecret == "")
304+ then sendToEach originServers (\server ->
305+ httpPost ("http://" ++ server ++ "/federation/ack")
306+ {authorization: "Bearer " ++ federationSecret}
307+ {messageIds: (do m <- fedMessages; where m.originServer == server; yield m.messageId),
308+ originServer: server})
276− respond {ok: ok}
310+ respond {ok: result.ok}
278 LookupPresence {pubkey, authorization, respond} -> do
312 LookupPresence {pubkey, authorization, respond} -> do
@@ -364,7 +398,7 @@ serve = \req -> case req of
364 set *messages = if shouldInsert
398 set *messages = if shouldInsert
365 then union messages
399 then union messages
366 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
400 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
367− encryptedBlob: encryptedBlob, timestamp: clampedTs}]
401+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer}]
368 else messages
402 else messages
369 respond {ok: authorized}
403 respond {ok: authorized}