Method watchChats

coderd/exp_chats.go:173–224

EXPERIMENTAL: this endpoint is experimental and is subject to change. @Summary Watch chat events for a user via WebSockets @ID watch-chat-events-for-a-user-via-websockets @Security CoderSessionToken @Tags Chats @Produce json @Success 200 {object} codersdk.ChatWatchEvent @Router /api/experimental/ch

(rw http.ResponseWriter, r *http.Request)

// @Router /api/experimental/chats/watch [get]
// @Description Experimental: this endpoint is subject to change.
func (api *API) watchChats(rw http.ResponseWriter, r *http.Request) {
	ctx := r.Context()
	apiKey := httpmw.APIKey(r)
	logger := api.Logger.Named("chat_watcher")

	conn, err := websocket.Accept(rw, r, nil)
	if err != nil {
		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
			Message: "Failed to open chat watch stream.",
			Detail:  err.Error(),
		})
		return
	}

	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	_ = conn.CloseRead(context.Background())

	ctx, wsNetConn := codersdk.WebsocketNetConn(ctx, conn, websocket.MessageText)
	defer wsNetConn.Close()

	go httpapi.HeartbeatClose(ctx, logger, cancel, conn)

	// The encoder is only written from the SubscribeWithErr callback,
	// which delivers serially per subscription. Do not add a second
	// write path without introducing synchronization.
	encoder := json.NewEncoder(wsNetConn)

	cancelSubscribe, err := api.Pubsub.SubscribeWithErr(pubsub.ChatWatchEventChannel(apiKey.UserID),
		pubsub.HandleChatWatchEvent(
			func(ctx context.Context, payload codersdk.ChatWatchEvent, err error) {
				if err != nil {
					logger.Error(ctx, "chat watch event subscription error", slog.Error(err))
					return
				}
				if err := encoder.Encode(payload); err != nil {
					logger.Debug(ctx, "failed to send chat watch event", slog.Error(err))
					cancel()
					return
				}
			},
		))
	if err != nil {
		logger.Error(ctx, "failed to subscribe to chat watch events", slog.Error(err))
		_ = conn.Close(websocket.StatusInternalError, "Failed to subscribe to chat events.")
		return
	}
	defer cancelSubscribe()

	<-ctx.Done()
}

// EXPERIMENTAL: chatsByWorkspace returns a mapping of workspace ID to
// the latest non-archived chat ID for each requested workspace.