257a8ff · skrepka · git.iko.soy

325 POST {pubkey: PubkeyHex, challenge: Text, signature: SignatureHex, revokeOthers: Bool} /auth/verify headers {clientIp: Text} -> {token: Text, expiresAt: Timestamp} = VerifyAuth

326

327route ClientApi where

328− POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int, hasMore: Bool} = Poll

328+ POST {cursor: Timestamp, generation: Int} /poll headers {authorization: Text, clientIp: Text} -> {events: [PollEvent], cursor: Timestamp, generation: Int} = Poll

329 POST {messages: [{to: PubkeyHex, encryptedBlob: BlobHex}]} /messages headers {authorization: Text, clientIp: Text} -> {results: [{error: Text}]} = SendMessages

330

331route FederationApi where

923-- generation mismatch (server restart) to force a cursor reset, then hands

924-- off to `pollOnce` for the actual long-poll body. Clients detect a server

925-- restart by comparing the `generation` they sent to the one returned.

926−handlePoll : Int -> Int -> Text -> Text -> IO {r *pollWakeup, r *serverGeneration, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int, hasMore: Bool})

926+handlePoll : Int -> Int -> Text -> Text -> IO {r *pollWakeup, r *serverGeneration, r *sessions, rw *forwards, rw *globalRateCount, rw *globalRateWindowStart, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int})

927handlePoll = \cursor generation authorization clientIp -> do

928 t <- now

929 serverGen <- *serverGeneration

937-- Body of a single poll. Treats the cursor advance as an implicit ack

938-- (`ackPolledMessages`), then runs `doPoll` to collect events (or block

939-- until wakeup) and assembles the response.

940−pollOnce : Text -> Int -> Text -> Int -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int, hasMore: Bool})

940+pollOnce : Text -> Int -> Text -> Int -> Int<Ms> -> IO {r *pollWakeup, r *sessions, rw *forwards, rw *messages, clock} (Result HttpError {cursor: Timestamp, events: [PollEvent], generation: Int})

941pollOnce = \authorization cursor clientIp serverGen t -> do

942 setup <- atomic withSessionAuth authorization t clientIp (yield (Nothing {})) (\pubkey -> do

943 let clampedCursor = clamp 0 t cursor

949 Just {value: s} -> do

950 pollResult <- doPoll s t

951 yield (Ok {

952− value: {pollResult.events, pollResult.cursor, generation: serverGen, pollResult.hasMore}

952+ value: {pollResult.events, pollResult.cursor, generation: serverGen}

953 })

954

955-- One pass: collect events at `t`. If empty, retry-wait on `pollWakeup`

958-- some other transaction commits a change to `*pollWakeup`. Because

959-- `pollHeartbeat` bumps `*pollWakeup` every `pollTimeoutMs`, no caller waits

960-- longer than one heartbeat period before getting at least an empty response.

961−doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}], hasMore: Bool}

961+doPoll : {clampedCursor: Int, pubkey: PubkeyHex, startWakeup: Timestamp} -> Int<Ms> -> IO {r *pollWakeup, clock} {cursor: Int<Ms>, events: [{encryptedBlob: BlobHex}]}

962doPoll = \s t -> do

963 let collectAt = \tnow -> atomic (do

964 msgs <- *messages

972 collectAt t2

973

974-- Pull this pubkey's unexpired messages newer than `cursor`, sort by

975−-- receivedAt, take up to `maxPollEvents`, and report `hasMore` if truncated.

975+-- receivedAt, take up to `maxPollEvents`. The returned cursor is the max

976−-- The returned cursor is the max `receivedAt` in the page (or the input

976+-- `receivedAt` in the page (or the input cursor when the page is empty),

977−-- cursor when the page is empty), which the client echoes on the next /poll

977+-- which the client echoes on the next /poll to advance — and which the

978−-- to advance — and which the server treats as an implicit ack.

978+-- server treats as an implicit ack.

979−collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp, hasMore: Bool}

979+collectPollEvents : [Message] -> PubkeyHex -> Timestamp -> Timestamp -> {events: [PollEvent], cursor: Timestamp}

980−collectPollEvents = \messages pubkey cursor t -> (\myMsgs -> (\limited -> {events: pluck limited (\m -> {m.encryptedBlob}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited, hasMore: count myMsgs > count limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) myMsgs))) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))

980+collectPollEvents = \messages pubkey cursor t -> (\limited -> {events: pluck limited (\m -> {m.encryptedBlob}), cursor: fold (\acc m -> max acc m.receivedAt) cursor limited}) (takeRelation maxPollEvents (sortBy (\m -> m.receivedAt) (filter messages (\m -> m.toKey == pubkey && t - m.receivedAt < messageMaxAge && m.receivedAt > cursor))))

981

982-- Treat the supplied cursor as an ack: drop all messages with

983-- receivedAt <= cursor, plus their pending forwards.