You've already forked agentic-coding-workflow
84 lines
2.0 KiB
Go
84 lines
2.0 KiB
Go
package chat
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
|
|
"github.com/go-chi/chi/v5"
|
|
|
|
"github.com/yan1h/agent-coding-workflow/internal/infra/errs"
|
|
)
|
|
|
|
// streamMessages implements GET /api/v1/conversations/{id}/stream.
|
|
// It upgrades the connection to a Server-Sent Events stream and forwards
|
|
// SSEEvents produced by MessageService.Stream until the channel closes or
|
|
// the client disconnects.
|
|
func (h *Handler) streamMessages(w http.ResponseWriter, r *http.Request) {
|
|
uid, err := h.userID(r)
|
|
if err != nil {
|
|
writeErr(w, r, err)
|
|
return
|
|
}
|
|
convID, err := parseUUID(chi.URLParam(r, "id"))
|
|
if err != nil {
|
|
writeErr(w, r, err)
|
|
return
|
|
}
|
|
var since int64
|
|
if s := r.URL.Query().Get("since"); s != "" {
|
|
since, _ = strconv.ParseInt(s, 10, 64)
|
|
}
|
|
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
http.Error(w, "streaming unsupported", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
w.Header().Set("Content-Type", "text/event-stream")
|
|
w.Header().Set("Cache-Control", "no-cache")
|
|
w.Header().Set("Connection", "keep-alive")
|
|
w.Header().Set("X-Accel-Buffering", "no")
|
|
w.WriteHeader(http.StatusOK)
|
|
|
|
ch, err := h.msgSvc.Stream(r.Context(), uid, convID, since)
|
|
if err != nil {
|
|
writeSSEError(w, flusher, err)
|
|
return
|
|
}
|
|
for {
|
|
select {
|
|
case <-r.Context().Done():
|
|
return
|
|
case ev, ok := <-ch:
|
|
if !ok {
|
|
return
|
|
}
|
|
writeSSE(w, flusher, ev)
|
|
}
|
|
}
|
|
}
|
|
|
|
// writeSSE writes a single SSE frame to w and flushes.
|
|
func writeSSE(w http.ResponseWriter, f http.Flusher, ev SSEEvent) {
|
|
data, _ := json.Marshal(ev.Data)
|
|
fmt.Fprintf(w, "id: %d\nevent: %s\ndata: %s\n\n", ev.ID, ev.Event, data)
|
|
f.Flush()
|
|
}
|
|
|
|
// writeSSEError writes an error event to the SSE stream and flushes.
|
|
func writeSSEError(w http.ResponseWriter, f http.Flusher, err error) {
|
|
code := errs.CodeInternal
|
|
msg := err.Error()
|
|
if ae, ok := errs.As(err); ok {
|
|
code = ae.Code
|
|
msg = ae.Message
|
|
}
|
|
payload := map[string]string{"code": string(code), "message": msg}
|
|
data, _ := json.Marshal(payload)
|
|
fmt.Fprintf(w, "event: error\ndata: %s\n\n", data)
|
|
f.Flush()
|
|
}
|