@@ -27,7 +27,7 @@ type GossipEvent = {eventType: Text, pubkey: Text, server: Text, timestamp: Int,
2727
2828*sessions : [{token: Text, pubkey: Text, expiresAt: Int}]
2929
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}]
3131
3232*presence : [{pubkey: Text, server: Text, expiresAt: Int}]
3333
@@ -101,6 +101,23 @@ verifyFederationAuth = \authorization ->
101101
102102clampTtl = \ttl -> if ttl < 1 then 1 else if ttl > maxPresenceTtl then maxPresenceTtl else ttl
103103
104+forwardToAll = \servers path headers body ->
105+ if count servers == 0
106+ then yield False {}
107+ else do
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)
112+
113+sendToEach = \servers action ->
114+ if count servers == 0
115+ then yield {}
116+ else do
117+ let server = fold (\acc s -> if acc == "" then s else acc) "" servers
118+ action server
119+ sendToEach (do s <- servers; where not (s == server); yield s) action
120+
104121pruneExpired = \t -> do
105122 challenges <- *challenges
106123 set *challenges = do
@@ -164,10 +181,7 @@ serve = \req -> case req of
164181 where not (c.pubkey == pubkey && c.challenge == challenge)
165182 yield c
166183 set *sessions = if sigValid
167− then union (do
168− s <- sessions
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}]
171185 else sessions
172186 set *presence = if sigValid
173187 then union
@@ -209,7 +223,7 @@ serve = \req -> case req of
209223 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
210224 t <- now
211225 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
213227 result <- atomic do
214228 pruneExpired t
215229 sessions <- *sessions
@@ -220,7 +234,7 @@ serve = \req -> case req of
220234 Nothing {} -> messages
221235 Just {value: s} -> union messages
222236 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
223− encryptedBlob: encryptedBlob, timestamp: clampedTs}]
237+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: ""}]
224238 let localOnline = case session of
225239 Nothing {} -> False {}
226240 Just {value: _} -> count (do
@@ -229,51 +243,71 @@ serve = \req -> case req of
229243 where p.server == serverHost
230244 where p.expiresAt > t
231245 yield p) > 0
232− let remoteServer = case session of
233− Nothing {} -> ""
234− Just {value: _} -> fold (\_ p -> p.server) "" (do
246+ let remoteServers = case session of
247+ Nothing {} -> []
248+ Just {value: _} -> do
235249 p <- presence
236250 where p.pubkey == to
237251 where not (p.server == serverHost)
238252 where p.expiresAt > t
239− yield p)
253+ yield p.server
240254 let fromKey = case session of
241255 Nothing {} -> ""
242256 Just {value: s} -> s.pubkey
243257 let status = case session of
244258 Nothing {} -> "unauthorized"
245259 _ -> if localOnline then "delivered"
246− else if not (remoteServer == "") then "federated"
260+ else if count remoteServers > 0 then "federated"
247261 else "queued"
248− yield {status: status, remoteServer: remoteServer, fromKey: fromKey}
249− if result.status == "federated" && not (result.remoteServer == "") && not (federationSecret == "")
250− then httpPost ("http://" ++ result.remoteServer ++ "/federation/forward")
251− {authorization: "Bearer " ++ federationSecret}
252− {toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
253− timestamp: clampedTs, originServer: serverHost, messageId: msgId}
254− else yield {}
255− respond {status: result.status, messageId: (if result.status == "unauthorized" then "" else msgId)}
262+ yield {status: status, remoteServers: remoteServers, fromKey: fromKey}
263+ finalStatus <- if result.status == "federated" && not (federationSecret == "")
264+ then do
265+ anyOk <- forwardToAll result.remoteServers "/federation/forward"
266+ {authorization: "Bearer " ++ federationSecret}
267+ {toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,
268+ timestamp: clampedTs, originServer: serverHost, messageId: msgId}
269+ yield (if anyOk then "federated" else "queued")
270+ else yield result.status
271+ respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else msgId)}
256272
257273 AckMessages {messageIds, authorization, respond} -> do
258274 t <- now
259− ok <- atomic do
275+ result <- atomic do
260276 pruneExpired t
261277 sessions <- *sessions
262278 messages <- *messages
263279 let session = lookupSession sessions (extractToken authorization) t
264− set *messages = case session of
265− Nothing {} -> messages
266− Just {value: s} -> diff messages (do
280+ let ackedMessages = case session of
281+ Nothing {} -> []
282+ Just {value: s} -> do
267283 m <- messages
268284 mid <- messageIds
269285 where m.messageId == mid
270286 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
272291 let ok = case session of
273292 Nothing {} -> False {}
274293 _ -> True {}
275− yield ok
276− respond {ok: 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)
299+ yield m
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})
309+ else yield {}
310+ respond {ok: result.ok}
277311
278312 LookupPresence {pubkey, authorization, respond} -> do
279313 t <- now
@@ -364,7 +398,7 @@ serve = \req -> case req of
364398 set *messages = if shouldInsert
365399 then union messages
366400 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
367− encryptedBlob: encryptedBlob, timestamp: clampedTs}]
401+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer}]
368402 else messages
369403 respond {ok: authorized}
370404