Retry clops (#3800)
* Add convo retries, sketch out tests * Only append nextMessage to messages * Remove debug codezio/stable
parent
333ccdad39
commit
fc0eab2d03
|
@ -7,6 +7,7 @@ import {ConvoItem, ConvoStatus} from '#/state/messages/convo'
|
||||||
import {isWeb} from 'platform/detection'
|
import {isWeb} from 'platform/detection'
|
||||||
import {MessageInput} from '#/screens/Messages/Conversation/MessageInput'
|
import {MessageInput} from '#/screens/Messages/Conversation/MessageInput'
|
||||||
import {MessageItem} from '#/screens/Messages/Conversation/MessageItem'
|
import {MessageItem} from '#/screens/Messages/Conversation/MessageItem'
|
||||||
|
import {Button, ButtonText} from '#/components/Button'
|
||||||
import {Loader} from '#/components/Loader'
|
import {Loader} from '#/components/Loader'
|
||||||
import {Text} from '#/components/Typography'
|
import {Text} from '#/components/Typography'
|
||||||
|
|
||||||
|
@ -31,6 +32,14 @@ function renderItem({item}: {item: ConvoItem}) {
|
||||||
return <Text>Deleted message</Text>
|
return <Text>Deleted message</Text>
|
||||||
} else if (item.type === 'pending-message') {
|
} else if (item.type === 'pending-message') {
|
||||||
return <Text>{item.message.text}</Text>
|
return <Text>{item.message.text}</Text>
|
||||||
|
} else if (item.type === 'pending-retry') {
|
||||||
|
return (
|
||||||
|
<View>
|
||||||
|
<Button label="Retry" onPress={item.retry}>
|
||||||
|
<ButtonText>Retry</ButtonText>
|
||||||
|
</Button>
|
||||||
|
</View>
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
return null
|
return null
|
||||||
|
|
|
@ -1,38 +0,0 @@
|
||||||
import {describe, it} from '@jest/globals'
|
|
||||||
|
|
||||||
describe(`#/state/dms/client`, () => {
|
|
||||||
describe(`ChatsService`, () => {
|
|
||||||
describe(`unread count`, () => {
|
|
||||||
it.todo(`marks a chat as read, decrements total unread count`)
|
|
||||||
})
|
|
||||||
|
|
||||||
describe(`log processing`, () => {
|
|
||||||
/*
|
|
||||||
* We receive a new chat log AND messages for it in the same batch. We
|
|
||||||
* need to first initialize the chat, then process the received logs.
|
|
||||||
*/
|
|
||||||
describe(`handles new chats and subsequent messages received in same log batch`, () => {
|
|
||||||
it.todo(`receives new chat and messages`)
|
|
||||||
it.todo(
|
|
||||||
`receives new chat, new messages come in while still initializing new chat`,
|
|
||||||
)
|
|
||||||
})
|
|
||||||
})
|
|
||||||
|
|
||||||
describe(`reset state`, () => {
|
|
||||||
it.todo(`after period of inactivity, rehydrates entirely fresh state`)
|
|
||||||
})
|
|
||||||
})
|
|
||||||
|
|
||||||
describe(`ChatService`, () => {
|
|
||||||
describe(`history fetching`, () => {
|
|
||||||
it.todo(`fetches initial chat history`)
|
|
||||||
it.todo(`fetches additional chat history`)
|
|
||||||
it.todo(`handles history fetch failure`)
|
|
||||||
})
|
|
||||||
|
|
||||||
describe(`optimistic updates`, () => {
|
|
||||||
it.todo(`adds sending messages`)
|
|
||||||
})
|
|
||||||
})
|
|
||||||
})
|
|
|
@ -0,0 +1,57 @@
|
||||||
|
import {describe, it} from '@jest/globals'
|
||||||
|
|
||||||
|
describe(`#/state/messages/convo`, () => {
|
||||||
|
describe(`status states`, () => {
|
||||||
|
it.todo(`cannot re-initialize from a non-unintialized state`)
|
||||||
|
it.todo(`can re-initialize from a failed state`)
|
||||||
|
|
||||||
|
describe(`destroy`, () => {
|
||||||
|
it.todo(`cannot be interacted with when destroyed`)
|
||||||
|
it.todo(`polling is stopped when destroyed`)
|
||||||
|
it.todo(`events are cleaned up when destroyed`)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`history fetching`, () => {
|
||||||
|
it.todo(`fetches initial chat history`)
|
||||||
|
it.todo(`fetches additional chat history`)
|
||||||
|
it.todo(`handles history fetch failure`)
|
||||||
|
it.todo(`does not insert deleted messages`)
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`sending messages`, () => {
|
||||||
|
it.todo(`optimistically adds sending messages`)
|
||||||
|
it.todo(`sends messages in order`)
|
||||||
|
it.todo(`failed message send fails all sending messages`)
|
||||||
|
it.todo(`can retry all failed messages via retry ConvoItem`)
|
||||||
|
it.todo(
|
||||||
|
`successfully sent messages are re-ordered, if needed, by events received from server`,
|
||||||
|
)
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`deleting messages`, () => {
|
||||||
|
it.todo(`messages are optimistically deleted from the chat`)
|
||||||
|
it.todo(`messages are confirmed deleted via events from the server`)
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`log handling`, () => {
|
||||||
|
it.todo(`updates rev to latest message received`)
|
||||||
|
it.todo(`only handles log events for this convoId`)
|
||||||
|
it.todo(`does not insert deleted messages`)
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`item ordering`, () => {
|
||||||
|
it.todo(`pending items are first, and in order`)
|
||||||
|
it.todo(`new message items are next, and in order`)
|
||||||
|
it.todo(`past message items are next, and in order`)
|
||||||
|
})
|
||||||
|
|
||||||
|
describe(`inactivity`, () => {
|
||||||
|
it.todo(
|
||||||
|
`below a certain threshold of inactivity, restore entirely from log`,
|
||||||
|
)
|
||||||
|
it.todo(
|
||||||
|
`above a certain threshold of inactivity, rehydrate entirely fresh state`,
|
||||||
|
)
|
||||||
|
})
|
||||||
|
})
|
|
@ -6,6 +6,8 @@ import {
|
||||||
import {EventEmitter} from 'eventemitter3'
|
import {EventEmitter} from 'eventemitter3'
|
||||||
import {nanoid} from 'nanoid/non-secure'
|
import {nanoid} from 'nanoid/non-secure'
|
||||||
|
|
||||||
|
import {isNative} from '#/platform/detection'
|
||||||
|
|
||||||
export type ConvoParams = {
|
export type ConvoParams = {
|
||||||
convoId: string
|
convoId: string
|
||||||
agent: BskyAgent
|
agent: BskyAgent
|
||||||
|
@ -44,6 +46,11 @@ export type ConvoItem =
|
||||||
key: string
|
key: string
|
||||||
message: ChatBskyConvoSendMessage.InputSchema['message']
|
message: ChatBskyConvoSendMessage.InputSchema['message']
|
||||||
}
|
}
|
||||||
|
| {
|
||||||
|
type: 'pending-retry'
|
||||||
|
key: string
|
||||||
|
retry: () => void
|
||||||
|
}
|
||||||
|
|
||||||
export type ConvoState =
|
export type ConvoState =
|
||||||
| {
|
| {
|
||||||
|
@ -66,6 +73,17 @@ export type ConvoState =
|
||||||
status: ConvoStatus.Destroyed
|
status: ConvoStatus.Destroyed
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function isConvoItemMessage(
|
||||||
|
item: ConvoItem,
|
||||||
|
): item is ConvoItem & {type: 'message'} {
|
||||||
|
if (!item) return false
|
||||||
|
return (
|
||||||
|
item.type === 'message' ||
|
||||||
|
item.type === 'deleted-message' ||
|
||||||
|
item.type === 'pending-message'
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
export class Convo {
|
export class Convo {
|
||||||
private convoId: string
|
private convoId: string
|
||||||
private agent: BskyAgent
|
private agent: BskyAgent
|
||||||
|
@ -90,8 +108,10 @@ export class Convo {
|
||||||
string,
|
string,
|
||||||
{id: string; message: ChatBskyConvoSendMessage.InputSchema['message']}
|
{id: string; message: ChatBskyConvoSendMessage.InputSchema['message']}
|
||||||
> = new Map()
|
> = new Map()
|
||||||
|
private footerItems: Map<string, ConvoItem> = new Map()
|
||||||
|
|
||||||
private pendingEventIngestion: Promise<void> | undefined
|
private pendingEventIngestion: Promise<void> | undefined
|
||||||
|
private isProcessingPendingMessages = false
|
||||||
|
|
||||||
constructor(params: ConvoParams) {
|
constructor(params: ConvoParams) {
|
||||||
this.convoId = params.convoId
|
this.convoId = params.convoId
|
||||||
|
@ -165,7 +185,7 @@ export class Convo {
|
||||||
{
|
{
|
||||||
cursor: this.historyCursor,
|
cursor: this.historyCursor,
|
||||||
convoId: this.convoId,
|
convoId: this.convoId,
|
||||||
limit: 20,
|
limit: isNative ? 25 : 50,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
headers: {
|
headers: {
|
||||||
|
@ -230,8 +250,6 @@ export class Convo {
|
||||||
/*
|
/*
|
||||||
* This is VERY important. We don't want to insert any messages from
|
* This is VERY important. We don't want to insert any messages from
|
||||||
* your other chats.
|
* your other chats.
|
||||||
*
|
|
||||||
* TODO there may be a better way to handle this
|
|
||||||
*/
|
*/
|
||||||
if (log.convoId !== this.convoId) continue
|
if (log.convoId !== this.convoId) continue
|
||||||
|
|
||||||
|
@ -241,7 +259,6 @@ export class Convo {
|
||||||
) {
|
) {
|
||||||
if (this.newMessages.has(log.message.id)) {
|
if (this.newMessages.has(log.message.id)) {
|
||||||
// Trust the log as the source of truth on ordering
|
// Trust the log as the source of truth on ordering
|
||||||
// TODO test this
|
|
||||||
this.newMessages.delete(log.message.id)
|
this.newMessages.delete(log.message.id)
|
||||||
}
|
}
|
||||||
this.newMessages.set(log.message.id, log.message)
|
this.newMessages.set(log.message.id, log.message)
|
||||||
|
@ -269,10 +286,116 @@ export class Convo {
|
||||||
this.commit()
|
this.commit()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async processPendingMessages() {
|
||||||
|
const pendingMessage = Array.from(this.pendingMessages.values()).shift()
|
||||||
|
|
||||||
|
/*
|
||||||
|
* If there are no pending messages, we're done.
|
||||||
|
*/
|
||||||
|
if (!pendingMessage) {
|
||||||
|
this.isProcessingPendingMessages = false
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
this.isProcessingPendingMessages = true
|
||||||
|
|
||||||
|
// throw new Error('UNCOMMENT TO TEST RETRY')
|
||||||
|
const {id, message} = pendingMessage
|
||||||
|
|
||||||
|
const response = await this.agent.api.chat.bsky.convo.sendMessage(
|
||||||
|
{
|
||||||
|
convoId: this.convoId,
|
||||||
|
message,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
encoding: 'application/json',
|
||||||
|
headers: {
|
||||||
|
Authorization: this.__tempFromUserDid,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
const res = response.data
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Insert into `newMessages` as soon as we have a real ID. That way, when
|
||||||
|
* we get an event log back, we can replace in situ.
|
||||||
|
*/
|
||||||
|
this.newMessages.set(res.id, {
|
||||||
|
...res,
|
||||||
|
$type: 'chat.bsky.convo.defs#messageView',
|
||||||
|
sender: this.convo?.members.find(m => m.did === this.__tempFromUserDid),
|
||||||
|
})
|
||||||
|
this.pendingMessages.delete(id)
|
||||||
|
|
||||||
|
await this.processPendingMessages()
|
||||||
|
|
||||||
|
this.commit()
|
||||||
|
} catch (e) {
|
||||||
|
this.footerItems.set('pending-retry', {
|
||||||
|
type: 'pending-retry',
|
||||||
|
key: 'pending-retry',
|
||||||
|
retry: this.batchRetryPendingMessages.bind(this),
|
||||||
|
})
|
||||||
|
this.commit()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async batchRetryPendingMessages() {
|
||||||
|
this.footerItems.delete('pending-retry')
|
||||||
|
this.commit()
|
||||||
|
|
||||||
|
try {
|
||||||
|
const messageArray = Array.from(this.pendingMessages.values())
|
||||||
|
const {data} = await this.agent.api.chat.bsky.convo.sendMessageBatch(
|
||||||
|
{
|
||||||
|
items: messageArray.map(({message}) => ({
|
||||||
|
convoId: this.convoId,
|
||||||
|
message,
|
||||||
|
})),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
encoding: 'application/json',
|
||||||
|
headers: {
|
||||||
|
Authorization: this.__tempFromUserDid,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
const {items} = data
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Insert into `newMessages` as soon as we have a real ID. That way, when
|
||||||
|
* we get an event log back, we can replace in situ.
|
||||||
|
*/
|
||||||
|
for (const item of items) {
|
||||||
|
this.newMessages.set(item.id, {
|
||||||
|
...item,
|
||||||
|
$type: 'chat.bsky.convo.defs#messageView',
|
||||||
|
sender: this.convo?.members.find(
|
||||||
|
m => m.did === this.__tempFromUserDid,
|
||||||
|
),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
for (const pendingMessage of messageArray) {
|
||||||
|
this.pendingMessages.delete(pendingMessage.id)
|
||||||
|
}
|
||||||
|
|
||||||
|
this.commit()
|
||||||
|
} catch (e) {
|
||||||
|
this.footerItems.set('pending-retry', {
|
||||||
|
type: 'pending-retry',
|
||||||
|
key: 'pending-retry',
|
||||||
|
retry: this.batchRetryPendingMessages.bind(this),
|
||||||
|
})
|
||||||
|
this.commit()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async sendMessage(message: ChatBskyConvoSendMessage.InputSchema['message']) {
|
async sendMessage(message: ChatBskyConvoSendMessage.InputSchema['message']) {
|
||||||
if (this.status === ConvoStatus.Destroyed) return
|
if (this.status === ConvoStatus.Destroyed) return
|
||||||
// Ignore empty messages for now since they have no other purpose atm
|
// Ignore empty messages for now since they have no other purpose atm
|
||||||
if (!message.text) return
|
if (!message.text.trim()) return
|
||||||
|
|
||||||
const tempId = nanoid()
|
const tempId = nanoid()
|
||||||
|
|
||||||
|
@ -282,33 +405,9 @@ export class Convo {
|
||||||
})
|
})
|
||||||
this.commit()
|
this.commit()
|
||||||
|
|
||||||
await new Promise(y => setTimeout(y, 500))
|
if (!this.isProcessingPendingMessages) {
|
||||||
const response = await this.agent.api.chat.bsky.convo.sendMessage(
|
this.processPendingMessages()
|
||||||
{
|
}
|
||||||
convoId: this.convoId,
|
|
||||||
message,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
encoding: 'application/json',
|
|
||||||
headers: {
|
|
||||||
Authorization: this.__tempFromUserDid,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
)
|
|
||||||
const res = response.data
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Insert into `newMessages` as soon as we have a real ID. That way, when
|
|
||||||
* we get an event log back, we can replace in situ.
|
|
||||||
*/
|
|
||||||
this.newMessages.set(res.id, {
|
|
||||||
...res,
|
|
||||||
$type: 'chat.bsky.convo.defs#messageView',
|
|
||||||
sender: this.convo?.members.find(m => m.did === this.__tempFromUserDid),
|
|
||||||
})
|
|
||||||
this.pendingMessages.delete(tempId)
|
|
||||||
|
|
||||||
this.commit()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/*
|
/*
|
||||||
|
@ -345,6 +444,10 @@ export class Convo {
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
this.footerItems.forEach(item => {
|
||||||
|
items.unshift(item)
|
||||||
|
})
|
||||||
|
|
||||||
this.pastMessages.forEach(m => {
|
this.pastMessages.forEach(m => {
|
||||||
if (ChatBskyConvoDefs.isMessageView(m)) {
|
if (ChatBskyConvoDefs.isMessageView(m)) {
|
||||||
items.push({
|
items.push({
|
||||||
|
@ -365,25 +468,33 @@ export class Convo {
|
||||||
|
|
||||||
return items.map((item, i) => {
|
return items.map((item, i) => {
|
||||||
let nextMessage = null
|
let nextMessage = null
|
||||||
|
const isMessage = isConvoItemMessage(item)
|
||||||
|
|
||||||
if (
|
if (isMessage) {
|
||||||
ChatBskyConvoDefs.isMessageView(item.message) ||
|
|
||||||
ChatBskyConvoDefs.isDeletedMessageView(item.message)
|
|
||||||
) {
|
|
||||||
const next = items[i - 1]
|
|
||||||
if (
|
if (
|
||||||
next &&
|
isMessage &&
|
||||||
(ChatBskyConvoDefs.isMessageView(next.message) ||
|
(ChatBskyConvoDefs.isMessageView(item.message) ||
|
||||||
ChatBskyConvoDefs.isDeletedMessageView(next.message))
|
ChatBskyConvoDefs.isDeletedMessageView(item.message))
|
||||||
) {
|
) {
|
||||||
nextMessage = next.message
|
const next = items[i - 1]
|
||||||
|
|
||||||
|
if (
|
||||||
|
isConvoItemMessage(next) &&
|
||||||
|
next &&
|
||||||
|
(ChatBskyConvoDefs.isMessageView(next.message) ||
|
||||||
|
ChatBskyConvoDefs.isDeletedMessageView(next.message))
|
||||||
|
) {
|
||||||
|
nextMessage = next.message
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
...item,
|
||||||
|
nextMessage,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return {
|
return item
|
||||||
...item,
|
|
||||||
nextMessage,
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue