| 5686 | } |
| 5687 | |
| 5688 | func (p *Server) publishMessage(chatID uuid.UUID, message database.ChatMessage) { |
| 5689 | sdkMessage := db2sdk.ChatMessage(message) |
| 5690 | event := codersdk.ChatStreamEvent{ |
| 5691 | Type: codersdk.ChatStreamEventTypeMessage, |
| 5692 | ChatID: chatID, |
| 5693 | Message: &sdkMessage, |
| 5694 | } |
| 5695 | p.cacheDurableMessage(chatID, event) |
| 5696 | // Claim every still-in-progress buffered message_part for this |
| 5697 | // durable assistant message BEFORE publishing it, so any new |
| 5698 | // subscriber that races publishEvent below takes a buffer |
| 5699 | // snapshot in which the parts for this turn are already |
| 5700 | // suppressed. Existing subscribers already received the |
| 5701 | // constituent parts on the live channel; the frontend |
| 5702 | // dedupes those against the durable message via |
| 5703 | // clearStreamState in the same batch. |
| 5704 | p.claimCommittedParts(chatID, message) |
| 5705 | p.publishEvent(chatID, event) |
| 5706 | p.publishChatStreamNotify(chatID, coderdpubsub.ChatStreamNotifyMessage{ |
| 5707 | AfterMessageID: message.ID - 1, |
| 5708 | }) |
| 5709 | } |
| 5710 | |
| 5711 | // claimCommittedParts walks the chat's buffered message_part events |
| 5712 | // and assigns every in-progress part (committedMessageID == 0) to |