feat: support streaming of rule logs

Author: danya

go.mod

@@ -3,6 +3,7 @@ module github.com/enapter/enapter-cli
 go 1.21
 
 require (
+	github.com/gorilla/websocket v1.5.3
 	github.com/stretchr/testify v1.9.0
 	github.com/urfave/cli/v2 v2.27.4
 )

go.sum

@@ -2,6 +2,8 @@ github.com/cpuguy83/go-md2man/v2 v2.0.5 h1:ZtcqGrnekaHpVLArFSe4HK5DoKx1T0rq2DwVB
 github.com/cpuguy83/go-md2man/v2 v2.0.5/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=

internal/app/enaptercli/cmd_base.go

@@ -12,6 +12,7 @@ import (
 	"net/url"
 	"strings"
 
+	"github.com/gorilla/websocket"
 	"github.com/urfave/cli/v2"
 )
 
@@ -122,6 +123,31 @@ func (c *cmdBase) doHTTPRequest(ctx context.Context, p doHTTPRequestParams) erro
 	return p.RespProcessor(resp)
 }
 
+func (c *cmdBase) dialWebSocket(ctx context.Context, path string) (*websocket.Conn, error) {
+	url, err := url.Parse(c.apiHost + "/v3" + path)
+	if err != nil {
+		return nil, fmt.Errorf("parse url: %w", err)
+	}
+	if url.Scheme == "https" {
+		url.Scheme = "wss"
+	} else {
+		url.Scheme = "ws"
+	}
+
+	headers := make(http.Header)
+	headers.Add("X-Enapter-Auth-Token", c.token)
+
+	conn, resp, err := websocket.DefaultDialer.DialContext(ctx, url.String(), headers)
+	if err != nil {
+		return nil, fmt.Errorf("dial: %w", err)
+	}
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		return nil, cli.Exit(parseRespErrorMessage(resp), 1)
+	}
+
+	return conn, nil
+}
+
 func (c *cmdBase) defaultRespProcessor(resp *http.Response) error {
 	if resp.StatusCode != http.StatusOK {
 		return cli.Exit(parseRespErrorMessage(resp), 1)

internal/app/enaptercli/cmd_rule_engine.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 
+	"github.com/gorilla/websocket"
 	"github.com/urfave/cli/v2"
 )
 
@@ -33,3 +34,11 @@ func (c *cmdRuleEngine) doHTTPRequest(ctx context.Context, p doHTTPRequestParams
 	p.Path = path
 	return c.cmdBase.doHTTPRequest(ctx, p)
 }
+
+func (c *cmdRuleEngine) dialWebSocket(ctx context.Context, path string) (*websocket.Conn, error) {
+	path, err := url.JoinPath("/site/rule_engine", path)
+	if err != nil {
+		return nil, fmt.Errorf("join path: %w", err)
+	}
+	return c.cmdBase.dialWebSocket(ctx, path)
+}

internal/app/enaptercli/cmd_rule_engine_rule.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 
+	"github.com/gorilla/websocket"
 	"github.com/urfave/cli/v2"
 )
 
@@ -30,6 +31,7 @@ func buildCmdRuleEngineRule() *cli.Command {
 			buildCmdRuleEngineRuleList(),
 			buildCmdRuleEngineRuleUpdate(),
 			buildCmdRuleEngineRuleUpdateScript(),
+			buildCmdRuleEngineRuleLogsf(),
 		},
 	}
 }
@@ -42,3 +44,11 @@ func (c *cmdRuleEngineRule) doHTTPRequest(ctx context.Context, p doHTTPRequestPa
 	p.Path = path
 	return c.cmdRuleEngine.doHTTPRequest(ctx, p)
 }
+
+func (c *cmdRuleEngineRule) dialWebSocket(ctx context.Context, path string) (*websocket.Conn, error) {
+	path, err := url.JoinPath("/rules", path)
+	if err != nil {
+		return nil, fmt.Errorf("join path: %w", err)
+	}
+	return c.cmdRuleEngine.dialWebSocket(ctx, path)
+}

internal/app/enaptercli/cmd_rule_engine_rule_logsf.go

@@ -0,0 +1,69 @@
+package enaptercli
+
+import (
+	"fmt"
+
+	"github.com/urfave/cli/v2"
+)
+
+type cmdRuleEngineRuleLogsf struct {
+	cmdRuleEngineRule
+	ruleID string
+}
+
+func buildCmdRuleEngineRuleLogsf() *cli.Command {
+	cmd := &cmdRuleEngineRuleLogsf{}
+	return &cli.Command{
+		Name:               "logsf",
+		Usage:              "Stream rule logs",
+		CustomHelpTemplate: cmd.HelpTemplate(),
+		Flags:              cmd.Flags(),
+		Before:             cmd.Before,
+		Action: func(cliCtx *cli.Context) error {
+			return cmd.do(cliCtx)
+		},
+	}
+}
+
+func (c *cmdRuleEngineRuleLogsf) Flags() []cli.Flag {
+	return append(c.cmdRuleEngineRule.Flags(),
+		&cli.StringFlag{
+			Name:        "rule-id",
+			Usage:       "Rule ID or slug to update",
+			Destination: &c.ruleID,
+			Required:    true,
+		},
+	)
+}
+
+func (c *cmdRuleEngineRuleLogsf) do(cliCtx *cli.Context) error {
+	conn, err := c.dialWebSocket(cliCtx.Context, c.ruleID+"/logs/ws")
+	if err != nil {
+		return fmt.Errorf("dial ws: %w", err)
+	}
+	defer conn.Close()
+
+	for {
+		select {
+		case <-cliCtx.Done():
+			return nil
+		default:
+		}
+
+		_, payload, err := conn.ReadMessage()
+		if err != nil {
+			return fmt.Errorf("read: %w", err)
+		}
+
+		select {
+		case <-cliCtx.Done():
+			return nil
+		default:
+		}
+
+		payload = append(payload, '\n')
+		if _, err = c.writer.Write(payload); err != nil {
+			return fmt.Errorf("write: %w", err)
+		}
+	}
+}

internal/app/enaptercli/testdata/helps/enapter rule-engine rule

@@ -13,6 +13,7 @@ COMMANDS:
    list           List rules
    update         Update a rule
    update-script  Update the script of a rule
+   logsf          Stream rule logs
    help, h        Shows a list of commands or help for one command
 
 OPTIONS: