feat(query): before fixing bugs, first you need tooling lol

cmd/humanlog/query.go

@@ -3,25 +3,30 @@ package main
 import (
 	"context"
 	"fmt"
-	"log"
 	"log/slog"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"connectrpc.com/connect"
 	"github.com/NimbleMarkets/ntcharts/canvas/runes"
 	"github.com/NimbleMarkets/ntcharts/linechart"
 	"github.com/NimbleMarkets/ntcharts/linechart/timeserieslinechart"
+	"github.com/charmbracelet/lipgloss"
 	"github.com/charmbracelet/x/term"
+	"github.com/crazy3lf/colorconv"
 	"github.com/humanlogio/api/go/svc/account/v1/accountv1connect"
 	"github.com/humanlogio/api/go/svc/organization/v1/organizationv1connect"
 	queryv1 "github.com/humanlogio/api/go/svc/query/v1"
 	"github.com/humanlogio/api/go/svc/query/v1/queryv1connect"
 	"github.com/humanlogio/api/go/svc/user/v1/userv1connect"
+	typesv1 "github.com/humanlogio/api/go/types/v1"
 	"github.com/humanlogio/humanlog/internal/pkg/config"
 	"github.com/humanlogio/humanlog/internal/pkg/state"
 	"github.com/humanlogio/humanlog/pkg/auth"
+	"github.com/humanlogio/humanlog/pkg/sink/stdiosink"
 	"github.com/humanlogio/humanlog/pkg/tui"
 	"github.com/urfave/cli"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -117,19 +122,36 @@ func queryApiSummarizeCmd(
 ) cli.Command {
 	fromFlag := cli.DurationFlag{Name: "since", Value: 365 * 24 * time.Hour}
 	toFlag := cli.DurationFlag{Name: "to", Value: 0}
+	localhost := cli.BoolFlag{Name: "localhost"}
 	return cli.Command{
 		Name:  "summarize",
-		Flags: []cli.Flag{fromFlag, toFlag},
+		Flags: []cli.Flag{localhost, fromFlag, toFlag},
 		Action: func(cctx *cli.Context) error {
 			ctx := getCtx(cctx)
 			state := getState(cctx)
-			ll := getLogger(cctx)
-			tokenSource := getTokenSource(cctx)
-			apiURL := getAPIUrl(cctx)
-			httpClient := getHTTPClient(cctx)
-			_, err := ensureLoggedIn(ctx, cctx, state, tokenSource, apiURL, httpClient)
-			if err != nil {
-				return err
+
+			var queryClient queryv1connect.QueryServiceClient
+			if !cctx.Bool(localhost.Name) {
+				ll := getLogger(cctx)
+				tokenSource := getTokenSource(cctx)
+				apiURL := getAPIUrl(cctx)
+				httpClient := getHTTPClient(cctx)
+				_, err := ensureLoggedIn(ctx, cctx, state, tokenSource, apiURL, httpClient)
+				if err != nil {
+					return err
+				}
+				clOpts := connect.WithInterceptors(
+					auth.Interceptors(ll, tokenSource)...,
+				)
+				queryClient = queryv1connect.NewQueryServiceClient(httpClient, apiURL, clOpts)
+			} else {
+				httpClient := getHTTPClient(cctx)
+				cfg := getCfg(cctx)
+				if cfg.ExperimentalFeatures == nil || cfg.ExperimentalFeatures.ServeLocalhostOnPort == nil {
+					return fmt.Errorf("localhost feature is not enabled or not configured, can't dial localhost")
+				}
+				addr := fmt.Sprintf("http://localhost:%d", *cfg.ExperimentalFeatures.ServeLocalhostOnPort)
+				queryClient = queryv1connect.NewQueryServiceClient(httpClient, addr)
 			}
 
 			termWidth, termHeight, err := term.GetSize(os.Stdout.Fd())
@@ -140,13 +162,8 @@ func queryApiSummarizeCmd(
 			from := now.Add(-cctx.Duration(fromFlag.Name))
 			to := now.Add(-cctx.Duration(toFlag.Name))
 
-			clOpts := connect.WithInterceptors(
-				auth.Interceptors(ll, tokenSource)...,
-			)
-			queryClient := queryv1connect.NewQueryServiceClient(httpClient, apiURL, clOpts)
-
 			res, err := queryClient.SummarizeEvents(ctx, connect.NewRequest(&queryv1.SummarizeEventsRequest{
-				AccountId:   *state.AccountID,
+				AccountId:   *state.CurrentAccountID,
 				BucketCount: 20,
 				From:        timestamppb.New(from),
 				To:          timestamppb.New(to),
@@ -215,11 +232,11 @@ func queryApiSummarizeCmd(
 				} else {
 					ts = t.Format(stepTimeFormat)
 				}
-				log.Printf("label: ts=%v", ts)
+				loginfo("label: ts=%v", ts)
 				return ts
 			})
 			for _, bucket := range buckets {
-				log.Printf("ts=%v   ev=%d", bucket.Ts.AsTime().Format(time.RFC3339Nano), bucket.GetEventCount())
+				loginfo("ts=%v   ev=%d", bucket.Ts.AsTime().Format(time.RFC3339Nano), bucket.GetEventCount())
 				tslc.Push(timeserieslinechart.TimePoint{
 					Time:  bucket.Ts.AsTime(),
 					Value: float64(bucket.GetEventCount()),
@@ -244,25 +261,162 @@ func queryApiWatchCmd(
 	getAPIUrl func(cctx *cli.Context) string,
 	getHTTPClient func(*cli.Context) *http.Client,
 ) cli.Command {
+	fromFlag := cli.DurationFlag{Name: "since", Value: 365 * 24 * time.Hour}
+	toFlag := cli.DurationFlag{Name: "to", Value: 0}
+	localhost := cli.BoolFlag{Name: "localhost"}
 	return cli.Command{
-		Name: "watch",
+		Name:  "watch",
+		Flags: []cli.Flag{localhost, fromFlag, toFlag},
 		Action: func(cctx *cli.Context) error {
 			ctx := getCtx(cctx)
+			cfg := getCfg(cctx)
 			state := getState(cctx)
-			tokenSource := getTokenSource(cctx)
-			apiURL := getAPIUrl(cctx)
-			httpClient := getHTTPClient(cctx)
-			_, err := ensureLoggedIn(ctx, cctx, state, tokenSource, apiURL, httpClient)
+			var queryClient queryv1connect.QueryServiceClient
+			if !cctx.Bool(localhost.Name) {
+				ll := getLogger(cctx)
+				tokenSource := getTokenSource(cctx)
+				apiURL := getAPIUrl(cctx)
+				httpClient := getHTTPClient(cctx)
+				_, err := ensureLoggedIn(ctx, cctx, state, tokenSource, apiURL, httpClient)
+				if err != nil {
+					return err
+				}
+				clOpts := connect.WithInterceptors(
+					auth.Interceptors(ll, tokenSource)...,
+				)
+				queryClient = queryv1connect.NewQueryServiceClient(httpClient, apiURL, clOpts)
+			} else {
+				httpClient := getHTTPClient(cctx)
+
+				if cfg.ExperimentalFeatures == nil || cfg.ExperimentalFeatures.ServeLocalhostOnPort == nil {
+					return fmt.Errorf("localhost feature is not enabled or not configured, can't dial localhost")
+				}
+				addr := fmt.Sprintf("http://localhost:%d", *cfg.ExperimentalFeatures.ServeLocalhostOnPort)
+				queryClient = queryv1connect.NewQueryServiceClient(httpClient, addr)
+			}
+			now := time.Now()
+			from := now.Add(-cctx.Duration(fromFlag.Name))
+			to := now.Add(-cctx.Duration(toFlag.Name))
+			sinkOpts, errs := stdiosink.StdioOptsFrom(*cfg)
+			if len(errs) > 0 {
+				for _, err := range errs {
+					logerror("config error: %v", err)
+				}
+			}
+
+			loginfo("from=%s", from)
+			loginfo("to=%s", to)
+			loginfo("query=%s", strings.Join(cctx.Args(), " "))
+
+			req := &queryv1.WatchQueryRequest{
+				AccountId: *state.CurrentAccountID,
+				Query: &typesv1.LogQuery{
+					From: timestamppb.New(from),
+					To:   timestamppb.New(to),
+				},
+			}
+			res, err := queryClient.WatchQuery(ctx, connect.NewRequest(req))
 			if err != nil {
-				return err
+				return fmt.Errorf("calling WatchQuery: %v", err)
+			}
+			defer res.Close()
+
+			sink := stdiosink.NewStdio(os.Stdout, sinkOpts)
+
+			for res.Receive() {
+				events := res.Msg().Events
+				for _, leg := range events {
+					prefix := getPrefix(leg.MachineId, leg.SessionId)
+					postProcess := func(pattern string) string {
+						return prefix + pattern
+					}
+					for _, ev := range leg.Logs {
+						if err := sink.ReceiveWithPostProcess(ctx, ev, postProcess); err != nil {
+							return fmt.Errorf("printing log: %v", err)
+						}
+					}
+				}
+			}
+			if err := res.Err(); err != nil {
+				return fmt.Errorf("querying: %v", err)
 			}
-			ll := getLogger(cctx)
-			clOpts := connect.WithInterceptors(
-				auth.Interceptors(ll, tokenSource)...,
-			)
-			queryClient := queryv1connect.NewQueryServiceClient(httpClient, apiURL, clOpts)
-			_ = queryClient
 			return nil
 		},
 	}
 }
+
+type tuple struct{ m, s int64 }
+
+var colorPrefixes = map[tuple]string{}
+
+func getPrefix(machine, session int64) string {
+	prefix, ok := colorPrefixes[tuple{m: machine, s: session}]
+	if ok {
+		return prefix
+	}
+	s := lipgloss.NewStyle().
+		BorderStyle(lipgloss.DoubleBorder()).BorderRight(true)
+
+	mPrefix := s.Background(lipgloss.AdaptiveColor{
+		Light: int64toLightRGB(machine),
+		Dark:  int64toDarkRGB(machine),
+	}).Render(strconv.FormatInt(machine, 10))
+	sPrefix := s.Background(lipgloss.AdaptiveColor{
+		Light: int64toLightRGB(session),
+		Dark:  int64toDarkRGB(session),
+	}).Render(strconv.FormatInt(session, 10))
+
+	prefix = lipgloss.JoinHorizontal(lipgloss.Left, mPrefix, sPrefix)
+	colorPrefixes[tuple{m: machine, s: session}] = prefix
+	return prefix
+}
+
+func int64toDarkRGB(n int64) string {
+	// modified from https://stackoverflow.com/a/52746259
+	n = (374761397 + n*3266489917) & 0xffffffff
+	n = ((n ^ n>>15) * 2246822519) & 0xffffffff
+	n = ((n ^ n>>13) * 3266489917) & 0xffffffff
+	n = (n ^ n>>16) >> 8
+
+	hex := fmt.Sprintf("#%06x", n)
+
+	// clamp the brightness
+	r, g, b, err := colorconv.HexToRGB(hex)
+	if err != nil {
+		panic(err)
+	}
+	h, s, v := colorconv.RGBToHSV(r, g, b)
+	if v > 0.5 {
+		v -= 0.5
+	}
+	r, g, b, err = colorconv.HSVToRGB(h, s, v)
+	if err != nil {
+		panic(err)
+	}
+	return fmt.Sprintf("#%02x%02x%02x", r, g, b)
+}
+
+func int64toLightRGB(n int64) string {
+	// modified from https://stackoverflow.com/a/52746259
+	n = (374761397 + n*3266489917) & 0xffffffff
+	n = ((n ^ n>>15) * 2246822519) & 0xffffffff
+	n = ((n ^ n>>13) * 3266489917) & 0xffffffff
+	n = (n ^ n>>16) >> 8
+
+	hex := fmt.Sprintf("#%06x", n)
+
+	// clamp the brightness
+	r, g, b, err := colorconv.HexToRGB(hex)
+	if err != nil {
+		panic(err)
+	}
+	h, s, v := colorconv.RGBToHSV(r, g, b)
+	if v < 0.5 {
+		v += 0.5
+	}
+	r, g, b, err = colorconv.HSVToRGB(h, s, v)
+	if err != nil {
+		panic(err)
+	}
+	return fmt.Sprintf("#%02x%02x%02x", r, g, b)
+}

cmd/humanlog/query_test.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func Test_int64toLightRGB(t *testing.T) {
+	tests := []struct {
+		name string
+		in   int64
+		want string
+	}{
+		{
+			in:   62,
+			want: "#3aef45",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := int64toLightRGB(tt.in)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func Test_int64toDarkRGB(t *testing.T) {
+	tests := []struct {
+		name string
+		in   int64
+		want string
+	}{
+		{
+			in:   62,
+			want: "#1b6f20",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := int64toDarkRGB(tt.in)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func Test_getPrefix(t *testing.T) {
+	type args struct {
+		machine int64
+		session int64
+	}
+	tests := []struct {
+		name string
+		args args
+		want string
+	}{
+		{
+			args: args{machine: 1, session: 2},
+			want: "1║2║",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getPrefix(tt.args.machine, tt.args.session); got != tt.want {
+				t.Errorf("getPrefix() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/charmbracelet/lipgloss v0.13.0
 	github.com/charmbracelet/x/term v0.2.0
 	github.com/cli/safeexec v1.0.1
+	github.com/crazy3lf/colorconv v1.2.0
 	github.com/fatih/color v1.16.0
 	github.com/gen2brain/beeep v0.0.0-20240516210008-9c006672e7f4
 	github.com/go-logfmt/logfmt v0.5.1

go.sum

@@ -43,6 +43,8 @@ github.com/cli/safeexec v1.0.1 h1:e/C79PbXF4yYTN/wauC4tviMxEV13BwljGj0N9j+N00=
 github.com/cli/safeexec v1.0.1/go.mod h1:Z/D4tTN8Vs5gXYHDCbaM1S/anmEDnJb1iW0+EJ5zx3Q=
 github.com/cpuguy83/go-md2man/v2 v2.0.2 h1:p1EgwI/C7NhT0JmVkwCD2ZBK8j4aeHQX2pMHHBfMQ6w=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/crazy3lf/colorconv v1.2.0 h1:UM7kSZWnwFMGiC+PpYrjxQSOd6sEyWb+dRKKTd3KslA=
+github.com/crazy3lf/colorconv v1.2.0/go.mod h1:2jTJ7QCWCj2sSLOhF4Gzi0J5/hoX8/VY8VzNvXAlD1I=
 github.com/danieljoos/wincred v1.2.0 h1:ozqKHaLK0W/ii4KVbbvluM91W2H3Sh0BncbUNPS7jLE=
 github.com/danieljoos/wincred v1.2.0/go.mod h1:FzQLLMKBFdvu+osBrnFODiv32YGwCfx0SkRa/eYHgec=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=