fix sse

core/clients/common.go

@@ -17,7 +17,6 @@ package clients
 import (
 	"context"
 	"errors"
-	"fmt"
 	"io"
 	"net/http"
 	"net/url"
@@ -48,11 +47,14 @@ func handleClientError(resp *resty.Response, err error) error {
 		return err
 	}
 	if resp.StatusCode() >= http.StatusBadRequest {
-		b, err := io.ReadAll(resp.RawBody())
-		if err != nil {
-			return fmt.Errorf("read body error failed: %v", err)
+		body := resp.String()
+		if body == "" {
+			b, err := io.ReadAll(resp.RawBody())
+			if err == nil {
+				body = strings.TrimSpace(string(b))
+			}
 		}
-		body := strings.TrimSpace(string(b))
+
 		errStr, err := strconv.Unquote(body)
 		if err != nil {
 			errStr = body

core/clients/server.go

@@ -17,8 +17,6 @@ package clients
 import (
 	"context"
 	"encoding/json"
-	"errors"
-	"io"
 	"strconv"
 	"strings"
 	"time"
@@ -343,47 +341,6 @@ func (c *serverV1) LogWatch(ctx context.Context, namespace, name string, number,
 		return nil, nil, err
 	}
 
-	logCh := make(chan *livelog.Line)
-	errCh := make(chan error)
-	go func() {
-		body := resp.RawBody()
-		defer body.Close()
-
-		sseParser := sse.NewParser(body)
-		err = sseParser.ReadEventLoop(func(message *sse.Message, err error) error {
-			select {
-			case <-ctx.Done():
-				return context.Canceled
-			default:
-			}
-			if errors.Is(err, io.EOF) {
-				return nil
-			}
-			if err != nil {
-				return err
-			}
-
-			if message.Event == "error" {
-				if strings.ToLower(message.Data) == "eof" {
-					return io.EOF
-				}
-				errCh <- errors.New(message.Data)
-				return nil
-			}
-			if message.Event != "data" {
-				return nil
-			}
-
-			var line livelog.Line
-			if err := json.Unmarshal([]byte(message.Data), &line); err != nil {
-				// TODO log
-				return nil
-			}
-			logCh <- &line
-			return nil
-		})
-		errCh <- err
-		close(errCh)
-	}()
-	return logCh, errCh, nil
+	dataCh, errCh := sse.ReceiveLoop[*livelog.Line](ctx, resp.RawBody(), nil)
+	return dataCh, errCh, nil
 }

core/handler/handler.go

@@ -55,6 +55,8 @@ var corsOpts = cors.Options{
 
 func New(log *wslog.Logger, db *gorm.DB, ll livelog.Interface, sched scheduler.Interface) http.Handler {
 	apiMiddlewares := chi.Middlewares{
+		middleware.Logger,
+		middleware.Recoverer,
 		cors.New(corsOpts).Handler,
 		serviceMiddleware(log, ll, sched, db),
 		timeoutMiddleware,

go.mod

@@ -3,7 +3,7 @@ module github.com/zc2638/ink
 go 1.21.0
 
 require (
-	github.com/99nil/gopkg v0.0.0-20240114071414-7dc8bfdd19e5
+	github.com/99nil/gopkg v0.0.0-20240115141111-91403b7f0c2a
 	github.com/docker/distribution v2.8.2+incompatible
 	github.com/docker/docker v24.0.5+incompatible
 	github.com/docker/go-units v0.5.0

go.sum

@@ -53,8 +53,8 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 gioui.org v0.0.0-20210308172011-57750fc8a0a6/go.mod h1:RSH6KIUZ0p2xy5zHDxgAM4zumjgTw83q2ge/PI+yyw8=
-github.com/99nil/gopkg v0.0.0-20240114071414-7dc8bfdd19e5 h1:sYpx9nb+ihnkXed2Ow+0bHo0B4OkoyM3P835MarqQ2I=
-github.com/99nil/gopkg v0.0.0-20240114071414-7dc8bfdd19e5/go.mod h1:yQy7Xa2r1VHYfgqfCuDE6VGDmBt8xJgDVTZ2Bw3xTW0=
+github.com/99nil/gopkg v0.0.0-20240115141111-91403b7f0c2a h1:96Hm1P0h2y7163cFDbM6NUFYlKPudkahnpqJW49G4PE=
+github.com/99nil/gopkg v0.0.0-20240115141111-91403b7f0c2a/go.mod h1:yQy7Xa2r1VHYfgqfCuDE6VGDmBt8xJgDVTZ2Bw3xTW0=
 github.com/AdaLogics/go-fuzz-headers v0.0.0-20210715213245-6c3934b029d8/go.mod h1:CzsSbkDixRphAF5hS6wbMKq0eI6ccJRb7/A0M6JBnwg=
 github.com/Azure/azure-pipeline-go v0.2.3/go.mod h1:x841ezTBIMG6O3lAcl8ATHnsOPVl2bqk7S3ta6S6u4k=
 github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=