@@ -238,10 +238,17 @@ serve = \req -> case req of
239 let token = extractToken authorization
239 let token = extractToken authorization
240 result <- atomic do
240 result <- atomic do
241+ presenceBeforePrune <- *presence
242+ let expiredLocal = do
243+ p <- presenceBeforePrune
244+ where p.expiresAt <= t
245+ where p.server == serverHost
241 pruneExpired t
247 pruneExpired t
242 sessions <- *sessions
248 sessions <- *sessions
243 messages <- *messages
249 messages <- *messages
244 let session = lookupSession sessions token t
250 let session = lookupSession sessions token t
251+ let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
245 let evts = case session of
252 let evts = case session of
246 Nothing {} -> []
253 Nothing {} -> []
247 Just {value: s} -> do
254 Just {value: s} -> do
@@ -255,9 +262,28 @@ serve = \req -> case req of
255 timestamp: m.timestamp}}
262 timestamp: m.timestamp}}
256 let newCursor = case session of
263 let newCursor = case session of
257 Nothing {} -> 0
264 Nothing {} -> 0
258− _ -> t
265+ _ -> maxReceived
259− yield {events: evts, cursor: newCursor}
266+ yield {events: evts, cursor: newCursor, expiredLocal: expiredLocal}
260− respond result
267+ respond {events: result.events, cursor: result.cursor}
268+ let expiredPubkeys = fold (\acc p ->
269+ if count (do a <- acc; where a == p.pubkey; yield a) > 0 then acc
270+ else union acc [p.pubkey]) [] result.expiredLocal
271+ if not (federationSecret == "") && count expiredPubkeys > 0
273+ peerServers <- atomic do
274+ presence <- *presence
275+ yield (fold (\acc p ->
276+ if p.server == serverHost then acc
277+ else if count (do a <- acc; where a == p.server; yield a) > 0 then acc
278+ else union acc [p.server]) [] presence)
279+ let gossipEvents = do
280+ pk <- expiredPubkeys
281+ yield {eventType: "offline", pubkey: pk, server: serverHost, timestamp: t, ttl: 3600}
282+ sendToEach peerServers (\server ->
283+ httpPost (federationScheme ++ server ++ "/federation/gossip")
284+ {authorization: "Bearer " ++ federationSecret}
285+ {events: gossipEvents, hopsRemaining: 2, fromServer: serverHost})
262 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
288 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
@@ -270,12 +296,14 @@ serve = \req -> case req of
270 messages <- *messages
296 messages <- *messages
271 presence <- *presence
297 presence <- *presence
272 let session = lookupSession sessions (extractToken authorization) t
298 let session = lookupSession sessions (extractToken authorization) t
299+ let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
300+ let seqTs = if t > maxReceived then t else maxReceived + 1
273 set *messages = case session of
301 set *messages = case session of
274 Nothing {} -> messages
302 Nothing {} -> messages
275 Just {value: s} -> if validInput
303 Just {value: s} -> if validInput
276 then union messages
304 then union messages
277 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
305 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
278− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: t}]
306+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
279 else messages
307 else messages
280 let localOnline = case session of
308 let localOnline = case session of
281 Nothing {} -> False {}
309 Nothing {} -> False {}
@@ -355,36 +383,39 @@ serve = \req -> case req of
355 else yield {}
383 else yield {}
356 respond {ok: result.ok}
384 respond {ok: result.ok}
358− LookupPresence {pubkey, authorization, respond} -> do
386+ LookupPresence {pubkey, authorization, respond} ->
388+ then respond {online: False {}, server: ""}
359− t <- now
390+ t <- now
360− result <- atomic do
391+ result <- atomic do
361− pruneExpired t
392+ pruneExpired t
362− sessions <- *sessions
393+ sessions <- *sessions
363− presence <- *presence
394+ presence <- *presence
364− let session = lookupSession sessions (extractToken authorization) t
395+ let session = lookupSession sessions (extractToken authorization) t
365− let valid = do
396+ let valid = do
366− p <- presence
397+ p <- presence
367− where p.pubkey == pubkey
398+ where p.pubkey == pubkey
368− where p.expiresAt > t
399+ where p.expiresAt > t
370− let foundServer = case session of
401+ let foundServer = case session of
371− Nothing {} -> ""
402+ Nothing {} -> ""
372− Just {value: _} ->
403+ Just {value: _} ->
373− if count valid == 0
404+ if count valid == 0
375− else case single (do
406+ else case single (do
376− p <- valid
407+ p <- valid
377− where p.server == serverHost
408+ where p.server == serverHost
378− yield p) of
409+ yield p) of
379− Just {value: p} -> p.server
410+ Just {value: p} -> p.server
380− Nothing {} -> fold (\_ p -> p.server) "" valid
411+ Nothing {} -> fold (\_ p -> p.server) "" valid
381− let authorized = case session of
412+ let authorized = case session of
382− Nothing {} -> False {}
413+ Nothing {} -> False {}
383− _ -> True {}
414+ _ -> True {}
384− yield {foundServer: foundServer, authorized: authorized}
415+ yield {foundServer: foundServer, authorized: authorized}
385− if result.authorized
416+ if result.authorized
386− then respond {online: not (result.foundServer == ""), server: result.foundServer}
417+ then respond {online: not (result.foundServer == ""), server: result.foundServer}
387− else respond {online: False {}, server: "unauthorized"}
418+ else respond {online: False {}, server: ""}
389 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
420 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
@@ -442,19 +473,44 @@ serve = \req -> case req of
442 atomic do
473 atomic do
443 pruneExpired t
474 pruneExpired t
444 presence <- *presence
475 presence <- *presence
445− let cleaned = diff presence (do
476+ let merged = fold (\acc e ->
479+ where p.pubkey == e.pubkey
480+ where p.server == e.server
481+ where p.expiresAt >= e.expiresAt
487+ where not (p.pubkey == e.pubkey && p.server == e.server)
446− p <- presence
489+ [e]) presence deduped
490+ let offlineOnlyKeys = do
447 k <- affectedKeys
491 k <- affectedKeys
492+ where count (do d <- deduped; where d.pubkey == k.pubkey; where d.server == k.server; yield d) == 0
494+ let afterOffline = diff merged (do
496+ k <- offlineOnlyKeys
448 where p.pubkey == k.pubkey
497 where p.pubkey == k.pubkey
449 where p.server == k.server
498 where p.server == k.server
501+ where o.pubkey == k.pubkey
502+ where o.server == k.server
503+ where o.timestamp >= p.expiresAt - maxPresenceTtl * 1000
451 set *presence = if validRequest
506 set *presence = if validRequest
452− then union cleaned deduped
507+ then afterOffline
453 else presence
508 else presence
454 let validEvents = do
509 let validEvents = do
455 e <- events
510 e <- events
456 where not (e.server == "")
511 where not (e.server == "")
457 where not (e.pubkey == "")
512 where not (e.pubkey == "")
513+ where not (e.server == serverHost)
458 where e.timestamp <= t
514 where e.timestamp <= t
459 where e.timestamp + (clampTtl e.ttl) * 1000 > t
515 where e.timestamp + (clampTtl e.ttl) * 1000 > t
@@ -489,6 +545,8 @@ serve = \req -> case req of
489 atomic do
545 atomic do
490 pruneExpired t
546 pruneExpired t
491 messages <- *messages
547 messages <- *messages
548+ let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
549+ let seqTs = if t > maxReceived then t else maxReceived + 1
492 let exists = count (do
550 let exists = count (do
493 m <- messages
551 m <- messages
494 where m.messageId == messageId
552 where m.messageId == messageId
@@ -498,7 +556,7 @@ serve = \req -> case req of
498 set *messages = if shouldInsert
556 set *messages = if shouldInsert
499 then union messages
557 then union messages
500 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
558 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
501− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: t}]
559+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
502 else messages
560 else messages
503 respond {ok: authorized && validOrigin && validFields}
561 respond {ok: authorized && validOrigin && validFields}