[🐴] Integrate event bus (#3915) · jeanmachine.dev/witchsky.app@3bac018

+103 -125

src/state/messages/convo/agent.ts

··· 10 10 import {logger} from '#/logger' 11 11 import {isNative} from '#/platform/detection' 12 12 import { 13 + ACTIVE_POLL_INTERVAL, 14 + BACKGROUND_POLL_INTERVAL, 15 + } from '#/state/messages/convo/const' 16 + import { 13 17 ConvoDispatch, 14 18 ConvoDispatchEvent, 15 19 ConvoErrorCode, ··· 19 23 ConvoState, 20 24 ConvoStatus, 21 25 } from '#/state/messages/convo/types' 22 - 23 - const ACTIVE_POLL_INTERVAL = 1e3 24 - const BACKGROUND_POLL_INTERVAL = 10e3 26 + import {MessagesEventBus} from '#/state/messages/events/agent' 27 + import {MessagesEventBusError} from '#/state/messages/events/types' 25 28 26 29 // TODO temporary 27 30 let DEBUG_ACTIVE_CHAT: string | undefined ··· 41 44 private id: string 42 45 43 46 private agent: BskyAgent 47 + private events: MessagesEventBus 44 48 private __tempFromUserDid: string 45 49 46 50 private status: ConvoStatus = ConvoStatus.Uninitialized 47 - private pollInterval = ACTIVE_POLL_INTERVAL 48 51 private error: 49 52 | { 50 53 code: ConvoErrorCode ··· 52 55 retry: () => void 53 56 } 54 57 | undefined 55 - private historyCursor: string | undefined | null = undefined 58 + private oldestRev: string | undefined | null = undefined 56 59 private isFetchingHistory = false 57 - private eventsCursor: string | undefined = undefined 60 + private latestRev: string | undefined = undefined 58 61 59 62 private pastMessages: Map< 60 63 string, ··· 73 76 private headerItems: Map<string, ConvoItem> = new Map() 74 77 75 78 private isProcessingPendingMessages = false 76 - private nextPoll: NodeJS.Timeout | undefined 77 79 78 80 convoId: string 79 81 convo: ChatBskyConvoDefs.ConvoView | undefined ··· 85 87 this.id = nanoid(3) 86 88 this.convoId = params.convoId 87 89 this.agent = params.agent 90 + this.events = params.events 88 91 this.__tempFromUserDid = params.__tempFromUserDid 89 92 90 93 this.subscribe = this.subscribe.bind(this) ··· 92 95 this.sendMessage = this.sendMessage.bind(this) 93 96 this.deleteMessage = this.deleteMessage.bind(this) 94 97 this.fetchMessageHistory = this.fetchMessageHistory.bind(this) 98 + this.ingestFirehose = this.ingestFirehose.bind(this) 99 + this.onFirehoseConnect = this.onFirehoseConnect.bind(this) 100 + this.onFirehoseError = this.onFirehoseError.bind(this) 95 101 96 102 if (DEBUG_ACTIVE_CHAT) { 97 103 logger.error(`Convo: another chat was already active`, { ··· 100 106 } else { 101 107 DEBUG_ACTIVE_CHAT = this.convoId 102 108 } 109 + 110 + this.events.trailConvo(this.convoId, events => { 111 + this.ingestFirehose(events) 112 + }) 113 + this.events.onConnect(this.onFirehoseConnect) 114 + this.events.onError(this.onFirehoseError) 103 115 } 104 116 105 117 private commit() { ··· 198 210 case ConvoDispatchEvent.Init: { 199 211 this.status = ConvoStatus.Initializing 200 212 this.setup() 213 + this.requestPollInterval(ACTIVE_POLL_INTERVAL) 201 214 break 202 215 } 203 216 } ··· 207 220 switch (action.event) { 208 221 case ConvoDispatchEvent.Ready: { 209 222 this.status = ConvoStatus.Ready 210 - this.pollInterval = ACTIVE_POLL_INTERVAL 211 - this.fetchMessageHistory().then(() => { 212 - this.restartPoll() 213 - }) 223 + this.fetchMessageHistory() 214 224 break 215 225 } 216 226 case ConvoDispatchEvent.Background: { 217 227 this.status = ConvoStatus.Backgrounded 218 - this.pollInterval = BACKGROUND_POLL_INTERVAL 219 - this.fetchMessageHistory().then(() => { 220 - this.restartPoll() 221 - }) 228 + this.fetchMessageHistory() 229 + this.requestPollInterval(BACKGROUND_POLL_INTERVAL) 222 230 break 223 231 } 224 232 case ConvoDispatchEvent.Suspend: { 225 233 this.status = ConvoStatus.Suspended 234 + this.withdrawRequestedPollInterval() 226 235 break 227 236 } 228 237 case ConvoDispatchEvent.Error: { 229 238 this.status = ConvoStatus.Error 230 239 this.error = action.payload 240 + this.withdrawRequestedPollInterval() 231 241 break 232 242 } 233 243 } ··· 237 247 switch (action.event) { 238 248 case ConvoDispatchEvent.Resume: { 239 249 this.refreshConvo() 240 - this.restartPoll() 250 + this.requestPollInterval(ACTIVE_POLL_INTERVAL) 241 251 break 242 252 } 243 253 case ConvoDispatchEvent.Background: { 244 254 this.status = ConvoStatus.Backgrounded 245 - this.pollInterval = BACKGROUND_POLL_INTERVAL 246 - this.restartPoll() 255 + this.requestPollInterval(BACKGROUND_POLL_INTERVAL) 247 256 break 248 257 } 249 258 case ConvoDispatchEvent.Suspend: { 250 259 this.status = ConvoStatus.Suspended 251 - this.cancelNextPoll() 260 + this.withdrawRequestedPollInterval() 252 261 break 253 262 } 254 263 case ConvoDispatchEvent.Error: { 255 264 this.status = ConvoStatus.Error 256 265 this.error = action.payload 257 - this.cancelNextPoll() 266 + this.withdrawRequestedPollInterval() 258 267 break 259 268 } 260 269 } ··· 262 271 } 263 272 case ConvoStatus.Backgrounded: { 264 273 switch (action.event) { 274 + // TODO truncate history if needed 265 275 case ConvoDispatchEvent.Resume: { 266 - this.status = ConvoStatus.Ready 267 - this.pollInterval = ACTIVE_POLL_INTERVAL 268 - this.refreshConvo() 269 - // TODO truncate history if needed 270 - this.restartPoll() 276 + if (this.convo) { 277 + this.status = ConvoStatus.Ready 278 + this.refreshConvo() 279 + } else { 280 + this.status = ConvoStatus.Initializing 281 + this.setup() 282 + } 283 + this.requestPollInterval(ACTIVE_POLL_INTERVAL) 271 284 break 272 285 } 273 286 case ConvoDispatchEvent.Suspend: { 274 287 this.status = ConvoStatus.Suspended 275 - this.cancelNextPoll() 288 + this.withdrawRequestedPollInterval() 276 289 break 277 290 } 278 291 case ConvoDispatchEvent.Error: { 279 292 this.status = ConvoStatus.Error 280 293 this.error = action.payload 281 - this.cancelNextPoll() 294 + this.withdrawRequestedPollInterval() 282 295 break 283 296 } 284 297 } ··· 287 300 case ConvoStatus.Suspended: { 288 301 switch (action.event) { 289 302 case ConvoDispatchEvent.Init: { 290 - this.status = ConvoStatus.Ready 291 - this.pollInterval = ACTIVE_POLL_INTERVAL 292 - this.refreshConvo() 293 - // TODO truncate history if needed 294 - this.restartPoll() 303 + this.reset() 295 304 break 296 305 } 297 306 case ConvoDispatchEvent.Resume: { 298 - this.status = ConvoStatus.Ready 299 - this.pollInterval = ACTIVE_POLL_INTERVAL 300 - this.refreshConvo() 301 - this.restartPoll() 307 + this.reset() 302 308 break 303 309 } 304 310 case ConvoDispatchEvent.Error: { ··· 356 362 357 363 this.status = ConvoStatus.Uninitialized 358 364 this.error = undefined 359 - this.historyCursor = undefined 360 - this.eventsCursor = undefined 365 + this.oldestRev = undefined 366 + this.latestRev = undefined 361 367 362 368 this.pastMessages = new Map() 363 369 this.newMessages = new Map() ··· 426 432 DEBUG_ACTIVE_CHAT = undefined 427 433 } 428 434 435 + private requestedPollInterval: (() => void) | undefined 436 + private requestPollInterval(interval: number) { 437 + this.withdrawRequestedPollInterval() 438 + this.requestedPollInterval = this.events.requestPollInterval(interval) 439 + } 440 + private withdrawRequestedPollInterval() { 441 + if (this.requestedPollInterval) { 442 + this.requestedPollInterval() 443 + } 444 + } 445 + 429 446 private pendingFetchConvo: 430 447 | Promise<{ 431 448 convo: ChatBskyConvoDefs.ConvoView ··· 499 516 logger.debug('Convo: fetch message history', {}, logger.DebugContext.convo) 500 517 501 518 /* 502 - * If historyCursor is null, we've fetched all history. 519 + * If oldestRev is null, we've fetched all history. 503 520 */ 504 - if (this.historyCursor === null) return 521 + if (this.oldestRev === null) return 505 522 506 523 /* 507 524 * Don't fetch again if a fetch is already in progress ··· 529 546 530 547 const response = await this.agent.api.chat.bsky.convo.getMessages( 531 548 { 532 - cursor: this.historyCursor, 549 + cursor: this.oldestRev, 533 550 convoId: this.convoId, 534 551 limit: isNative ? 25 : 50, 535 552 }, ··· 541 558 ) 542 559 const {cursor, messages} = response.data 543 560 544 - this.historyCursor = cursor ?? null 561 + this.oldestRev = cursor ?? null 545 562 546 563 for (const message of messages) { 547 564 if ( 548 565 ChatBskyConvoDefs.isMessageView(message) || 549 566 ChatBskyConvoDefs.isDeletedMessageView(message) 550 567 ) { 551 - this.pastMessages.set(message.id, message) 552 - 553 - // set to latest rev 554 - if ( 555 - message.rev > (this.eventsCursor = this.eventsCursor || message.rev) 556 - ) { 557 - this.eventsCursor = message.rev 568 + /* 569 + * If this message is already in new messages, it was added by the 570 + * firehose ingestion, and we can safely overwrite it. This trusts 571 + * the server on ordering, and keeps it in sync. 572 + */ 573 + if (this.newMessages.has(message.id)) { 574 + this.newMessages.delete(message.id) 558 575 } 576 + this.pastMessages.set(message.id, message) 559 577 } 560 578 } 561 579 } catch (e: any) { ··· 576 594 } 577 595 } 578 596 579 - private restartPoll() { 580 - this.cancelNextPoll() 581 - this.pollLatestEvents() 597 + onFirehoseConnect() { 598 + this.footerItems.delete(ConvoItemError.PollFailed) 599 + this.commit() 582 600 } 583 601 584 - private cancelNextPoll() { 585 - if (this.nextPoll) clearTimeout(this.nextPoll) 586 - } 587 - 588 - private pollLatestEvents() { 589 - /* 590 - * Uncomment to view poll events 591 - */ 592 - logger.debug('Convo: poll events', {id: this.id}, logger.DebugContext.convo) 593 - 594 - try { 595 - this.fetchLatestEvents().then(({events}) => { 596 - this.applyLatestEvents(events) 597 - }) 598 - this.nextPoll = setTimeout(() => { 599 - this.pollLatestEvents() 600 - }, this.pollInterval) 601 - } catch (e: any) { 602 - logger.error('Convo: poll events failed') 603 - 604 - this.cancelNextPoll() 605 - 606 - this.footerItems.set(ConvoItemError.PollFailed, { 607 - type: 'error-recoverable', 608 - key: ConvoItemError.PollFailed, 609 - code: ConvoItemError.PollFailed, 610 - retry: () => { 611 - this.footerItems.delete(ConvoItemError.PollFailed) 612 - this.commit() 613 - this.pollLatestEvents() 614 - }, 615 - }) 616 - 617 - this.commit() 618 - } 619 - } 620 - 621 - private pendingFetchLatestEvents: 622 - | Promise<{ 623 - events: ChatBskyConvoGetLog.OutputSchema['logs'] 624 - }> 625 - | undefined 626 - async fetchLatestEvents() { 627 - if (this.pendingFetchLatestEvents) return this.pendingFetchLatestEvents 628 - 629 - this.pendingFetchLatestEvents = new Promise<{ 630 - events: ChatBskyConvoGetLog.OutputSchema['logs'] 631 - }>(async (resolve, reject) => { 632 - try { 633 - // throw new Error('UNCOMMENT TO TEST POLL FAILURE') 634 - const response = await this.agent.api.chat.bsky.convo.getLog( 635 - { 636 - cursor: this.eventsCursor, 637 - }, 638 - { 639 - headers: { 640 - Authorization: this.__tempFromUserDid, 641 - }, 642 - }, 643 - ) 644 - const {logs} = response.data 645 - resolve({events: logs}) 646 - } catch (e) { 647 - reject(e) 648 - } finally { 649 - this.pendingFetchLatestEvents = undefined 650 - } 602 + onFirehoseError(error?: MessagesEventBusError) { 603 + this.footerItems.set(ConvoItemError.PollFailed, { 604 + type: 'error-recoverable', 605 + key: ConvoItemError.PollFailed, 606 + code: ConvoItemError.PollFailed, 607 + retry: () => { 608 + this.footerItems.delete(ConvoItemError.PollFailed) 609 + this.commit() 610 + error?.retry() 611 + }, 651 612 }) 652 - 653 - return this.pendingFetchLatestEvents 613 + this.commit() 654 614 } 655 615 656 - private applyLatestEvents(events: ChatBskyConvoGetLog.OutputSchema['logs']) { 616 + ingestFirehose(events: ChatBskyConvoGetLog.OutputSchema['logs']) { 657 617 let needsCommit = false 658 618 659 619 for (const ev of events) { ··· 662 622 * know what it is. 663 623 */ 664 624 if (typeof ev.rev === 'string') { 625 + const isUninitialized = !this.latestRev 626 + const isNewEvent = this.latestRev && ev.rev > this.latestRev 627 + 628 + /* 629 + * We received an event prior to fetching any history, so we can safely 630 + * use this as the initial history cursor 631 + */ 632 + if (this.oldestRev === undefined && isUninitialized) { 633 + this.oldestRev = ev.rev 634 + } 635 + 665 636 /* 666 637 * We only care about new events 667 638 */ 668 - if (ev.rev > (this.eventsCursor = this.eventsCursor || ev.rev)) { 639 + if (isNewEvent || isUninitialized) { 669 640 /* 670 641 * Update rev regardless of if it's a ev type we care about or not 671 642 */ 672 - this.eventsCursor = ev.rev 643 + this.latestRev = ev.rev 673 644 674 645 /* 675 646 * This is VERY important. We don't want to insert any messages from ··· 681 652 ChatBskyConvoDefs.isLogCreateMessage(ev) && 682 653 ChatBskyConvoDefs.isMessageView(ev.message) 683 654 ) { 655 + /** 656 + * If this message is already in new messages, it was added by our 657 + * sending logic, and is based on client-ordering. When we receive 658 + * the "commited" event from the log, we should replace this 659 + * reference and re-insert in order to respect the order we receied 660 + * from the log. 661 + */ 684 662 if (this.newMessages.has(ev.message.id)) { 685 - // Trust the ev as the source of truth on ordering 686 663 this.newMessages.delete(ev.message.id) 687 664 } 688 665 this.newMessages.set(ev.message.id, ev.message) ··· 694 671 /* 695 672 * Update if we have this in state. If we don't, don't worry about it. 696 673 */ 674 + // TODO check for other storage spots 697 675 if (this.pastMessages.has(ev.message.id)) { 698 676 /* 699 677 * For now, we remove deleted messages from the thread, if we receive one.

+2

src/state/messages/convo/const.ts

··· 1 + export const ACTIVE_POLL_INTERVAL = 1e3 2 + export const BACKGROUND_POLL_INTERVAL = 5e3

+3

src/state/messages/convo/index.tsx

··· 5 5 6 6 import {Convo} from '#/state/messages/convo/agent' 7 7 import {ConvoParams, ConvoState} from '#/state/messages/convo/types' 8 + import {useMessagesEventBus} from '#/state/messages/events' 8 9 import {useMarkAsReadMutation} from '#/state/queries/messages/conversation' 9 10 import {useAgent} from '#/state/session' 10 11 import {useDmServiceUrlStorage} from '#/screens/Messages/Temp/useDmServiceUrlStorage' ··· 26 27 const isScreenFocused = useIsFocused() 27 28 const {serviceUrl} = useDmServiceUrlStorage() 28 29 const {getAgent} = useAgent() 30 + const events = useMessagesEventBus() 29 31 const [convo] = useState( 30 32 () => 31 33 new Convo({ ··· 33 35 agent: new BskyAgent({ 34 36 service: serviceUrl, 35 37 }), 38 + events, 36 39 __tempFromUserDid: getAgent().session?.did!, 37 40 }), 38 41 )

+3

src/state/messages/convo/types.ts

··· 5 5 ChatBskyConvoSendMessage, 6 6 } from '@atproto-labs/api' 7 7 8 + import {MessagesEventBus} from '#/state/messages/events/agent' 9 + 8 10 export type ConvoParams = { 9 11 convoId: string 10 12 agent: BskyAgent 13 + events: MessagesEventBus 11 14 __tempFromUserDid: string 12 15 } 13 16

+9 -1

src/state/messages/events/agent.ts

··· 347 347 348 348 this.isPolling = true 349 349 350 - logger.debug(`${LOGGER_CONTEXT}: poll`, {}, logger.DebugContext.convo) 350 + // logger.debug( 351 + // `${LOGGER_CONTEXT}: poll`, 352 + // { 353 + // requestedPollIntervals: Array.from( 354 + // this.requestedPollIntervals.values(), 355 + // ), 356 + // }, 357 + // logger.DebugContext.convo, 358 + // ) 351 359 352 360 try { 353 361 const response = await this.agent.api.chat.bsky.convo.getLog(

Configure Feed

Configure Feed