@@ -238,10 +238,17 @@ serve = \req -> case req of
238238 t <- now
239239 let token = extractToken authorization
240240 result <- atomic do
241+ presenceBeforePrune <- *presence
242+ let expiredLocal = do
243+ p <- presenceBeforePrune
244+ where p.expiresAt <= t
245+ where p.server == serverHost
246+ yield p
241247 pruneExpired t
242248 sessions <- *sessions
243249 messages <- *messages
244250 let session = lookupSession sessions token t
251+ let maxReceived = fold (\acc m -> if m.receivedAt > acc then m.receivedAt else acc) 0 messages
245252 let evts = case session of
246253 Nothing {} -> []
247254 Just {value: s} -> do
@@ -255,9 +262,28 @@ serve = \req -> case req of
255262 timestamp: m.timestamp}}
256263 let newCursor = case session of
257264 Nothing {} -> 0
258− _ -> t
259− yield {events: evts, cursor: newCursor}
260− respond result
265+ _ -> maxReceived
266+ yield {events: evts, cursor: newCursor, expiredLocal: expiredLocal}
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
272+ then do
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})
286+ else yield {}
261287
262288 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do
263289 t <- now
@@ -270,12 +296,14 @@ serve = \req -> case req of
270296 messages <- *messages
271297 presence <- *presence
272298 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
273301 set *messages = case session of
274302 Nothing {} -> messages
275303 Just {value: s} -> if validInput
276304 then union messages
277305 [{messageId: msgId, toKey: to, fromKey: s.pubkey,
278− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: t}]
306+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: "", receivedAt: seqTs}]
279307 else messages
280308 let localOnline = case session of
281309 Nothing {} -> False {}
@@ -355,36 +383,39 @@ serve = \req -> case req of
355383 else yield {}
356384 respond {ok: result.ok}
357385
358− LookupPresence {pubkey, authorization, respond} -> do
359− t <- now
360− result <- atomic do
361− pruneExpired t
362− sessions <- *sessions
363− presence <- *presence
364− let session = lookupSession sessions (extractToken authorization) t
365− let valid = do
366− p <- presence
367− where p.pubkey == pubkey
368− where p.expiresAt > t
369− yield p
370− let foundServer = case session of
371− Nothing {} -> ""
372− Just {value: _} ->
373− if count valid == 0
374− then ""
375− else case single (do
376− p <- valid
377− where p.server == serverHost
378− yield p) of
379− Just {value: p} -> p.server
380− Nothing {} -> fold (\_ p -> p.server) "" valid
381− let authorized = case session of
382− Nothing {} -> False {}
383− _ -> True {}
384− yield {foundServer: foundServer, authorized: authorized}
385− if result.authorized
386− then respond {online: not (result.foundServer == ""), server: result.foundServer}
387− else respond {online: False {}, server: "unauthorized"}
386+ LookupPresence {pubkey, authorization, respond} ->
387+ if pubkey == ""
388+ then respond {online: False {}, server: ""}
389+ else do
390+ t <- now
391+ result <- atomic do
392+ pruneExpired t
393+ sessions <- *sessions
394+ presence <- *presence
395+ let session = lookupSession sessions (extractToken authorization) t
396+ let valid = do
397+ p <- presence
398+ where p.pubkey == pubkey
399+ where p.expiresAt > t
400+ yield p
401+ let foundServer = case session of
402+ Nothing {} -> ""
403+ Just {value: _} ->
404+ if count valid == 0
405+ then ""
406+ else case single (do
407+ p <- valid
408+ where p.server == serverHost
409+ yield p) of
410+ Just {value: p} -> p.server
411+ Nothing {} -> fold (\_ p -> p.server) "" valid
412+ let authorized = case session of
413+ Nothing {} -> False {}
414+ _ -> True {}
415+ yield {foundServer: foundServer, authorized: authorized}
416+ if result.authorized
417+ then respond {online: not (result.foundServer == ""), server: result.foundServer}
418+ else respond {online: False {}, server: ""}
388419
389420 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do
390421 t <- now
@@ -442,19 +473,44 @@ serve = \req -> case req of
442473 atomic do
443474 pruneExpired t
444475 presence <- *presence
445− let cleaned = diff presence (do
446− p <- presence
476+ let merged = fold (\acc e ->
477+ if count (do
478+ p <- acc
479+ where p.pubkey == e.pubkey
480+ where p.server == e.server
481+ where p.expiresAt >= e.expiresAt
482+ yield p) > 0
483+ then acc
484+ else union
485+ (do
486+ p <- acc
487+ where not (p.pubkey == e.pubkey && p.server == e.server)
488+ yield p)
489+ [e]) presence deduped
490+ let offlineOnlyKeys = do
447491 k <- affectedKeys
492+ where count (do d <- deduped; where d.pubkey == k.pubkey; where d.server == k.server; yield d) == 0
493+ yield k
494+ let afterOffline = diff merged (do
495+ p <- merged
496+ k <- offlineOnlyKeys
448497 where p.pubkey == k.pubkey
449498 where p.server == k.server
499+ where count (do
500+ o <- offlineEntries
501+ where o.pubkey == k.pubkey
502+ where o.server == k.server
503+ where o.timestamp >= p.expiresAt - maxPresenceTtl * 1000
504+ yield o) > 0
450505 yield p)
451506 set *presence = if validRequest
452− then union cleaned deduped
507+ then afterOffline
453508 else presence
454509 let validEvents = do
455510 e <- events
456511 where not (e.server == "")
457512 where not (e.pubkey == "")
513+ where not (e.server == serverHost)
458514 where e.timestamp <= t
459515 where e.timestamp + (clampTtl e.ttl) * 1000 > t
460516 yield e
@@ -489,6 +545,8 @@ serve = \req -> case req of
489545 atomic do
490546 pruneExpired t
491547 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
492550 let exists = count (do
493551 m <- messages
494552 where m.messageId == messageId
@@ -498,7 +556,7 @@ serve = \req -> case req of
498556 set *messages = if shouldInsert
499557 then union messages
500558 [{messageId: messageId, toKey: toKey, fromKey: fromKey,
501− encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: t}]
559+ encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]
502560 else messages
503561 respond {ok: authorized && validOrigin && validFields}
504562