e5192b2 · skrepka · git.iko.soy

479 authorization: "Bearer " ++ federationSecret,

480 events: batch, hopsRemaining: 2, fromServer: serverHostN})

481 case batchResult of

482− Ok _ -> sendBatchedGossipToServer server rest

482+ Ok _ -> yield {}

483 Err _ -> yield {}

484+ sendBatchedGossipToServer server rest

484

485

485sendBatchedGossip : [Text] -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}] -> IO {} {}

486sendBatchedGossip : [Text] -> [{eventType: Text, pubkey: Text, server: Text, timestamp: Int, ttl: Int}] -> IO {} {}

486sendBatchedGossip = \peerServers allEvents ->

487sendBatchedGossip = \peerServers allEvents ->

667 globalCapped

668 globalCapped

668 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

669 [{pubkey: pubkey, challenge: challenge, expiresAt: t + challengeExpiry}]

669 yield pr

670 yield pr

670− broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal

671+ fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)

671 yield (respond {challenge: challenge})

672 yield (respond {challenge: challenge})

672 else yield (respond {challenge: ""})

673 else yield (respond {challenge: ""})

673

674

750 else union acc [p.server]) [] updatedPresence

751 else union acc [p.server]) [] updatedPresence

751 else []

752 else []

752 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}

753 yield {sigValid: sigValid, peerServers: peerServers, expiredLocal: pr.expiredLocal, offlinePeers: pr.peerServers}

754+ fork (do

753− if result.sigValid && not (federationSecret == "")

755+ if result.sigValid && not (federationSecret == "")

754− then sendToEach result.peerServers (\server -> do

756+ then sendToEach result.peerServers (\server -> do

755− fetch (federationScheme ++ server) (RecvGossip {

757+ fetchResult <- fetch (federationScheme ++ server) (RecvGossip {

756− authorization: "Bearer " ++ federationSecret,

758+ authorization: "Bearer " ++ federationSecret,

757− events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})

759+ events: [{eventType: "online", pubkey: pubkey, server: serverHostN, timestamp: t, ttl: onlineGossipTtl}], hopsRemaining: 2, fromServer: serverHostN})

760+ case fetchResult of

758− yield {}

761+ Ok _ -> yield {}

762+ Err _ -> yield {}

759− )

763+ )

760−

761− else yield {}

764+ else yield {}

762− broadcastOfflineGossip t result.offlinePeers (do

765+ broadcastOfflineGossip t result.offlinePeers (do

763− p <- result.expiredLocal

766+ p <- result.expiredLocal

764− where not (result.sigValid && p.pubkey == pubkey)

767+ where not (result.sigValid && p.pubkey == pubkey)

765− yield p)

768+ yield p))

766 yield (if result.sigValid

769 yield (if result.sigValid

767 then respond {token: token, expiresAt: expiry}

770 then respond {token: token, expiresAt: expiry}

768 else respond {token: "", expiresAt: 0})

771 else respond {token: "", expiresAt: 0})

776 tAfterPoll <- now

779 tAfterPoll <- now

777 pruneResult <- atomic do

780 pruneResult <- atomic do

778 maybePrune tAfterPoll

781 maybePrune tAfterPoll

779− broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal

782+ fork (broadcastOfflineGossip tAfterPoll pruneResult.peerServers pruneResult.expiredLocal)

780 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

783 yield (respond {events: result.events, cursor: result.cursor, authorized: result.authorized})

781

784

782 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do

785 SendMessage {to, encryptedBlob, timestamp, authorization, respond} -> do

838 else "queued"

841 else "queued"

839 forwards <- *forwards

842 forwards <- *forwards

840 let forwardsAtCapacity = count forwards >= maxForwardsEntries

843 let forwardsAtCapacity = count forwards >= maxForwardsEntries

841− set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0

844+ set *forwards = if shouldInsert && not forwardsAtCapacity && count remoteServers > 0 && not (federationSecret == "")

842 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})

845 then union forwards (do s <- remoteServers; yield {messageId: msgId, toServer: s, createdAt: t, toKey: to, fromKey: fromKey, encryptedBlob: encryptedBlob, timestamp: clampedTs})

843 else forwards

846 else forwards

844 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

847 yield {status: status, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

845 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status

848 let finalStatus = if result.status == "federated" && (federationSecret == "" || result.forwardsAtCapacity) then "queued" else result.status

849+ fork (do

846− if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected") && not (federationSecret == "")

850+ if count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected") && not (federationSecret == "")

847− then sendToEach result.remoteServers (\server -> do

851+ then sendToEach result.remoteServers (\server -> do

848− fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {

852+ fwdResult <- fetch (federationScheme ++ server) (ForwardMessage {

849− toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

853+ toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

850− timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

854+ timestamp: clampedTs, originServer: serverHostN, messageId: msgId,

851− authorization: "Bearer " ++ federationSecret})

855+ authorization: "Bearer " ++ federationSecret})

852− case fwdResult of

856+ case fwdResult of

853− Ok _ -> atomic do

857+ Ok _ -> atomic do

854− forwards <- *forwards

858+ forwards <- *forwards

855− set *forwards = do

859+ set *forwards = do

856− x <- forwards

860+ x <- forwards

857− where not (x.messageId == msgId && x.toServer == server)

861+ where not (x.messageId == msgId && x.toServer == server)

858− yield x

862+ yield x

859− Err _ -> yield {})

863+ Err _ -> yield {})

860− else yield {}

864+ else yield {}

861− if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected")

865+ if result.forwardsAtCapacity && count result.remoteServers > 0 && not (result.status == "unauthorized") && not (result.status == "invalid") && not (result.status == "rejected")

862− then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

866+ then println ("sendMessage: forwards at capacity (" ++ show maxForwardsEntries ++ "), skipping forward tracking")

863− else yield {}

867+ else yield {}

864− broadcastOfflineGossip t result.peerServers result.expiredLocal

868+ broadcastOfflineGossip t result.peerServers result.expiredLocal)

865 yield (respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else if result.status == "rejected" then "" else msgId)})

869 yield (respond {status: finalStatus, messageId: (if result.status == "unauthorized" then "" else if result.status == "invalid" then "" else if result.status == "rejected" then "" else msgId)})

866

870

867 AckMessages {messageIds, authorization, respond} -> do

871 AckMessages {messageIds, authorization, respond} -> do

914 let originServers = fold (\acc m ->

918 let originServers = fold (\acc m ->

915 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc

919 if count (do a <- acc; where a == m.originServer; yield a) > 0 then acc

916 else union acc [m.originServer]) [] fedMessages

920 else union acc [m.originServer]) [] fedMessages

921+ fork (do

917− if result.ok && count originServers > 0 && not (federationSecret == "")

922+ if result.ok && count originServers > 0 && not (federationSecret == "")

918− then do

923+ then do

919− let pendingAcks = do

924+ let pendingAcks = do

920− m <- fedMessages

925+ m <- fedMessages

921− yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}

926+ yield {messageId: m.messageId, toKey: m.toKey, ackedAt: t, originServer: m.originServer}

922− sendAcksAndClean pendingAcks originServers

927+ sendAcksAndClean pendingAcks originServers

923− else yield {}

928+ else yield {}

924− if result.ok

929+ if result.ok

925− then do

930+ then do

926− let evictedToRetry = do

931+ let evictedToRetry = do

927− a <- result.evictedEntries

932+ a <- result.evictedEntries

928− where count (do m <- fedMessages; where m.messageId == a.messageId; where m.toKey == a.toKey; yield m) == 0

933+ where count (do m <- fedMessages; where m.messageId == a.messageId; where m.toKey == a.toKey; yield m) == 0

929− yield a

934+ yield a

930− if count evictedToRetry > 0 && not (federationSecret == "")

935+ if count evictedToRetry > 0 && not (federationSecret == "")

931− then do

936+ then do

932− let evictedOriginServers = fold (\acc a ->

937+ let evictedOriginServers = fold (\acc a ->

933− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc

938+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc

934− else union acc [a.originServer]) [] evictedToRetry

939+ else union acc [a.originServer]) [] evictedToRetry

935− sendToEach evictedOriginServers (\server -> do

940+ sendToEach evictedOriginServers (\server -> do

936− result <- fetch (federationScheme ++ server) (FederationAck {

941+ result <- fetch (federationScheme ++ server) (FederationAck {

937− authorization: "Bearer " ++ federationSecret,

942+ authorization: "Bearer " ++ federationSecret,

938− messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),

943+ messageIds: (do a <- evictedToRetry; where a.originServer == server; yield a.messageId),

939− originServer: server, fromServer: serverHostN})

944+ originServer: server, fromServer: serverHostN})

940− case result of

945+ case result of

941− Err _ -> atomic do

946+ Err _ -> atomic do

942− ackedIds <- *ackedIds

947+ ackedIds <- *ackedIds

943− let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

948+ let reinsert = do a <- evictedToRetry; where a.originServer == server; yield a

944− let combined = union ackedIds reinsert

949+ let combined = union ackedIds reinsert

945− let excess = count combined - maxAckedIds

950+ let excess = count combined - maxAckedIds

946− let afterEviction = if excess <= 0 then combined

951+ let afterEviction = if excess <= 0 then combined

947− else evictOldestAckedIds combined excess

952+ else evictOldestAckedIds combined excess

948− set *ackedIds = afterEviction

953+ set *ackedIds = afterEviction

949− yield {}

954+ yield {}

950− _ -> yield {}

955+ _ -> yield {}

951− )

956+ )

952− else yield {}

957+ else yield {}

953− else yield {}

958+ else yield {}

954− broadcastOfflineGossip t result.peerServers result.expiredLocal

959+ broadcastOfflineGossip t result.peerServers result.expiredLocal)

955 yield (respond {ok: result.ok})

960 yield (respond {ok: result.ok})

956

961

957 LookupPresence {pubkey, authorization, respond} -> do

962 LookupPresence {pubkey, authorization, respond} -> do

980 then serverHostN

985 then serverHostN

981 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server

986 else (fold (\acc p -> if p.expiresAt > acc.best then {server: p.server, best: p.expiresAt} else acc) {server: "", best: 0} valid).server

982 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}

987 yield {foundServer: foundServer, authorized: authorized, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers}

983− broadcastOfflineGossip t result.peerServers result.expiredLocal

988+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)

984 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})

989 yield (respond {online: not (result.foundServer == ""), server: result.foundServer, authorized: result.authorized})

985

990

986 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

991 RecvGossip {events, hopsRemaining, fromServer, authorization, respond} -> do

1212 if not gossipResult.withinCapacity

1217 if not gossipResult.withinCapacity

1213 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

1218 then println ("gossip: presence at capacity (" ++ show maxPresenceEntries ++ "), limiting new entries from " ++ fromServer)

1214 else yield {}

1219 else yield {}

1220+ fork (do

1215− if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0

1221+ if validRequest && cappedHops > 1 && not (federationSecret == "") && count effectiveEvents > 0

1216− then do

1222+ then do

1217− let targets = do

1223+ let targets = do

1218− s <- gossipResult.gossipPeerServers

1224+ s <- gossipResult.gossipPeerServers

1219− where not (s == fromServer)

1225+ where not (s == fromServer)

1220− yield s

1226+ yield s

1221− sendToEach targets (\server -> do

1227+ sendToEach targets (\server -> do

1222− fetch (federationScheme ++ server) (RecvGossip {

1228+ fetchResult <- fetch (federationScheme ++ server) (RecvGossip {

1223− authorization: "Bearer " ++ federationSecret,

1229+ authorization: "Bearer " ++ federationSecret,

1224− events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})

1230+ events: effectiveEvents, hopsRemaining: cappedHops - 1, fromServer: serverHostN})

1231+ case fetchResult of

1225− yield {}

1232+ Ok _ -> yield {}

1233+ Err _ -> yield {}

1226− )

1234+ )

1227− else yield {}

1235+ else yield {}

1228− broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal

1236+ broadcastOfflineGossip t gossipResult.peerServers gossipResult.expiredLocal)

1229 yield (respond {ok: validRequest})

1237 yield (respond {ok: validRequest})

1230

1238

1231 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do

1239 ForwardMessage {toKey, fromKey, encryptedBlob, timestamp, originServer, messageId, authorization, respond} -> do

1272 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1280 encryptedBlob: encryptedBlob, timestamp: clampedTs, originServer: originServer, receivedAt: seqTs}]

1273 else messages

1281 else messages

1274 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}

1282 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, inserted: shouldInsert, alreadyExists: exists}

1275− broadcastOfflineGossip t result.peerServers result.expiredLocal

1283+ fork (broadcastOfflineGossip t result.peerServers result.expiredLocal)

1276 yield (respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)})

1284 yield (respond {ok: result.inserted || (authorized && validOrigin && validFields && result.alreadyExists)})

1277

1285

1278 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

1286 FederationAck {messageIds, originServer, fromServer, authorization, respond} -> do

1302 then diff forwards forwardsToRemove

1310 then diff forwards forwardsToRemove

1303 else forwards

1311 else forwards

1304 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1312 yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1305− broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal

1313+ fork (broadcastOfflineGossip t pruneResult.peerServers pruneResult.expiredLocal)

1306 yield (respond {ok: pruneResult.validAck})

1314 yield (respond {ok: pruneResult.validAck})

1307

1315

1308 Prune {authorization, respond} -> do

1316 Prune {authorization, respond} -> do

1324 where not (a.originServer == serverHostN)

1332 where not (a.originServer == serverHostN)

1325 yield a

1333 yield a

1326 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}

1334 yield {expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, pendingForwards: pendingForwards, pendingAcks: pendingAcks}

1335+ fork (do

1327− broadcastOfflineGossip t result.peerServers result.expiredLocal

1336+ broadcastOfflineGossip t result.peerServers result.expiredLocal

1328− if not (federationSecret == "") && count result.pendingForwards > 0

1337+ if not (federationSecret == "") && count result.pendingForwards > 0

1329− then retryPendingForwards result.pendingForwards

1338+ then retryPendingForwards result.pendingForwards

1330− else yield {}

1339+ else yield {}

1331− if not (federationSecret == "") && count result.pendingAcks > 0

1340+ if not (federationSecret == "") && count result.pendingAcks > 0

1332− then do

1341+ then do

1333− let ackOriginServers = fold (\acc a ->

1342+ let ackOriginServers = fold (\acc a ->

1334− if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc

1343+ if count (do s <- acc; where s == a.originServer; yield s) > 0 then acc

1335− else union acc [a.originServer]) [] result.pendingAcks

1344+ else union acc [a.originServer]) [] result.pendingAcks

1336− sendAcksAndClean result.pendingAcks ackOriginServers

1345+ sendAcksAndClean result.pendingAcks ackOriginServers

1337− else yield {}

1346+ else yield {})

1338 yield (respond {ok: True {}})

1347 yield (respond {ok: True {}})

1339 else yield (respond {ok: False {}})

1348 else yield (respond {ok: False {}})

1340

1349