Method SubscribeAuthorized

coderd/x/chatd/chatd.go:4959–5463

SubscribeAuthorized subscribes an already-authorized chat to merged stream updates. The passed chat row proves authorization, but SubscribeAuthorized still reloads the chat after the stream subscriptions are armed so the initial status and relay setup use fresh state.

(
	ctx context.Context,
	chat database.Chat,
	requestHeader http.Header,
	afterMessageID int64,
)

// still reloads the chat after the stream subscriptions are armed so the
// initial status and relay setup use fresh state.
func (p *Server) SubscribeAuthorized(
	ctx context.Context,
	chat database.Chat,
	requestHeader http.Header,
	afterMessageID int64,
) (
	[]codersdk.ChatStreamEvent,
	<-chan codersdk.ChatStreamEvent,
	func(),
	bool,
) {
	if p == nil {
		return nil, nil, nil, false
	}
	chatID := chat.ID

	// Subscribe to the local stream for message_parts and same-replica
	// persisted messages. Capture the current retry phase under the same
	// lock so the transient snapshot and subscriber registration reflect
	// a single moment in time.
	localSnapshot, localRetry, localParts, localCancel := p.subscribeToStream(chatID)

	// Merge all event sources.
	mergedCtx, mergedCancel := context.WithCancel(ctx)
	mergedEvents := make(chan codersdk.ChatStreamEvent, 128)

	var allCancels []func()
	allCancels = append(allCancels, localCancel)

	// Subscribe to pubsub for durable and structured control
	// events (status, messages, queue updates, retry, errors).
	// When pubsub is nil (e.g. in-memory
	// single-instance) we skip this and deliver all local events.
	//
	// This MUST happen before the DB queries below so that any
	// notification published between the query and the subscription
	// is not lost (subscribe-first-then-query pattern).
	var notifications <-chan coderdpubsub.ChatStreamNotifyMessage
	var errCh <-chan error
	if p.pubsub != nil {
		notifyCh := make(chan coderdpubsub.ChatStreamNotifyMessage, 10)
		errNotifyCh := make(chan error, 1)
		notifications = notifyCh
		errCh = errNotifyCh

		listener := func(_ context.Context, message []byte, listenErr error) {
			if listenErr != nil {
				select {
				case <-mergedCtx.Done():
				case errNotifyCh <- listenErr:
				}
				return
			}
			var notify coderdpubsub.ChatStreamNotifyMessage
			if unmarshalErr := json.Unmarshal(message, &notify); unmarshalErr != nil {
				select {
				case <-mergedCtx.Done():
				case errNotifyCh <- xerrors.Errorf("unmarshal chat stream notify: %w", unmarshalErr):