9520409 · skrepka · git.iko.soy

130-- ============================================

131

132any = \xs pred -> count (do x <- xs; where pred x; yield x) > 0

133−none = \xs pred -> count (do x <- xs; where pred x; yield x) == 0

133+none = \xs pred -> not (any xs pred)

134isJust = \m -> case m of

135 Nothing {} -> False {}

136 _ -> True {}

463

464samePubkeyServer = \a b -> a.pubkey == b.pubkey && a.server == b.server

465

466−mergeByPubkeyServer = \initial items ->

466+mergeByKey = \sameKey initial items ->

467 fold (\acc e ->

468− if any acc (\a -> samePubkeyServer a e && a.expiresAt >= e.expiresAt)

468+ if any acc (\a -> sameKey a e && a.expiresAt >= e.expiresAt)

469 then acc

470 else union

471− (do a <- acc; where not (samePubkeyServer a e); yield a)

471+ (do a <- acc; where not (sameKey a e); yield a)

472 [e]) initial items

473

474+mergeByPubkeyServer = mergeByKey samePubkeyServer

475+

474removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->

476removeOffline = \presenceSet offlineOnlyKeys trustedOffline t ->

475 diff presenceSet (do

477 diff presenceSet (do

476 p <- presenceSet

478 p <- presenceSet

582 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)

584 sendToEach peerServers (\server -> sendBatchedGossipToServer server allEvents)

583

585

584broadcastOfflineGossip = \t peerServers expiredLocal -> do

586broadcastOfflineGossip = \t peerServers expiredLocal -> do

585− let expiredEntries = fold (\acc p ->

587+ let expiredEntries = mergeByKey (\a b -> a.pubkey == b.pubkey) [] (do

586− if any acc (\a -> a.pubkey == p.pubkey && a.expiresAt >= p.expiresAt) then acc

588+ p <- expiredLocal

587− else union

588− (do a <- acc; where not (a.pubkey == p.pubkey); yield a)

589− [{pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in clampTtl (if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else rawTtl)), expiresAt: p.expiresAt}]) [] expiredLocal

589+ yield {pubkey: p.pubkey, ttl: (let rawTtl = (p.expiresAt - p.timestamp) / 1000 in clampTtl (if rawTtl < offlineGossipMinTtl then offlineGossipMinTtl else rawTtl)), expiresAt: p.expiresAt})

590 if federationEnabled && count expiredEntries > 0 && count peerServers > 0

591 then do

592 let gossipEvents = do

863 where p.expiresAt > t

864 yield p.server

865 else []

866+ forwards <- *forwards

867+ let forwardsAtCapacity = count forwards >= maxForwardsEntries

866 let status = if not authorized then "unauthorized"

868 let status = if not authorized then "unauthorized"

867 else if not shouldInsert then "rejected"

869 else if not shouldInsert then "rejected"

868 else if localOnline then "delivered"

870 else if localOnline then "delivered"

869− else if count remoteServers > 0 then "federated"

871+ else if count remoteServers > 0 && federationEnabled && not forwardsAtCapacity then "federated"

870 else "queued"

872 else "queued"

871− forwards <- *forwards

872− let forwardsAtCapacity = count forwards >= maxForwardsEntries

873 let forwardSlots = maxForwardsEntries - count forwards

874 let forwardServers = takeN remoteServers forwardSlots

875 set *forwards = if shouldInsert && not forwardsAtCapacity && count forwardServers > 0 && federationEnabled

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

877 else forwards

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

878+ yield {status: status, shouldInsert: shouldInsert, remoteServers: remoteServers, fromKey: fromKey, expiredLocal: pr.expiredLocal, peerServers: pr.peerServers, forwardsAtCapacity: forwardsAtCapacity}

879− let finalStatus = if result.status == "federated" && (not federationEnabled || result.forwardsAtCapacity) then "queued" else result.status

880− let actionable = not (result.status == "unauthorized") && not (result.status == "rejected")

881 fork (do

879 fork (do

882− if count result.remoteServers > 0 && actionable && federationEnabled

880+ if count result.remoteServers > 0 && result.shouldInsert && federationEnabled

883 then sendToEach result.remoteServers (\server -> do

881 then sendToEach result.remoteServers (\server -> do

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

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

885 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

883 toKey: to, fromKey: result.fromKey, encryptedBlob: encryptedBlob,

890 Err _ -> yield {}

888 Err _ -> yield {}

891 )

889 )

892 else yield {}

890 else yield {}

893− if result.forwardsAtCapacity && count result.remoteServers > 0 && actionable

891+ if result.forwardsAtCapacity && count result.remoteServers > 0 && result.shouldInsert

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

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

895 else yield {}

893 else yield {}

896 broadcastOfflineGossip t result.peerServers result.expiredLocal)

894 broadcastOfflineGossip t result.peerServers result.expiredLocal)

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

895+ yield (respond {status: result.status, messageId: if result.shouldInsert then msgId else ""})

898

896

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

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

900 t <- now

898 t <- now

1159 t <- now

1157 t <- now

1160 let authorized = verifyFederationAuth authorization

1158 let authorized = verifyFederationAuth authorization

1161 let dedupedIds = dedupMessageIds messageIds

1159 let dedupedIds = dedupMessageIds messageIds

1162− let validAckFields = authorized && originServer == serverHostN && not (fromServer == serverHostN)

1160+ let validAck = authorized && originServer == serverHostN && not (fromServer == serverHostN)

1163 pruneResult <- atomic do

1161 pruneResult <- atomic do

1164 pr <- maybePrune t

1162 pr <- maybePrune t

1165 forwards <- *forwards

1163 forwards <- *forwards

1166− let validAck = validAckFields

1167 let forwardsToRemove = do

1164 let forwardsToRemove = do

1168 f <- forwards

1165 f <- forwards

1169 mid <- dedupedIds

1166 mid <- dedupedIds

1173 set *forwards = if validAck

1170 set *forwards = if validAck

1174 then diff forwards forwardsToRemove

1171 then diff forwards forwardsToRemove

1175 else forwards

1172 else forwards

1176− yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal, validAck: validAck}

1173+ yield {peerServers: pr.peerServers, expiredLocal: pr.expiredLocal}

1177 forkBroadcastOffline t pruneResult

1174 forkBroadcastOffline t pruneResult

1178− yield (respond {ok: pruneResult.validAck})

1175+ yield (respond {ok: validAck})

1179

1176

1180 Prune {authorization, respond} -> do

1177 Prune {authorization, respond} -> do

1181 t <- now

1178 t <- now