zen: many updates

cmd/zen/main.go

@@ -1,8 +1,16 @@
 package main
 
-import "gitlab.bareksa.com/backend/zen/cmd/zen/serve"
+import (
+	"context"
+	"os"
+	"os/signal"
+
+	"gitlab.bareksa.com/backend/zen/cmd/zen/serve"
+)
 
 // This is the main entry point for CLI interface.
 func main() {
-	serve.Serve()
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer cancel()
+	serve.Serve(ctx)
 }

cmd/zen/serve/serve.go

@@ -1,20 +1,44 @@
 package serve
 
 import (
+	"context"
+	"errors"
 	"log/slog"
+	"net"
 	"net/http"
+	"os"
+	"time"
 
 	"connectrpc.com/grpcreflect"
+	"gitlab.bareksa.com/backend/zen"
 	"gitlab.bareksa.com/backend/zen/internal/gen/proto/notify/v1/notifyv1connect"
 	"gitlab.bareksa.com/backend/zen/internal/rpchandler"
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
+	"golang.org/x/sync/errgroup"
 
-	_ "gitlab.bareksa.com/backend/zen/core/zerr"
+	"gitlab.bareksa.com/backend/zen/core/zlog"
 )
 
-func Serve() {
+// Get preferred outbound ip of this machine
+func GetOutboundIP() net.IP {
+	conn, err := net.Dial("udp", "8.8.8.8:80")
+	if err != nil {
+		panic(err)
+	}
+	defer conn.Close()
+
+	localAddr := conn.LocalAddr().(*net.UDPAddr)
+
+	return localAddr.IP
+}
+
+func Serve(ctx context.Context) {
+	logger := zlog.New(os.Stderr, nil)
+	slog.SetDefault(slog.New(logger))
+	zen.SetDefaultLogger(logger)
+
 	mux := http.NewServeMux()
 	reflector := grpcreflect.NewStaticReflector(
 		notifyv1connect.NotifyServiceName,
@@ -24,12 +48,42 @@ func Serve() {
 	mux.Handle(grpcreflect.NewHandlerV1Alpha(reflector))
 	mux.Handle(notifyv1connect.NewNotifyServiceHandler(rpchandler.NotifyServiceHandler{}))
 
-	slog.Info("Starting server on :8080")
-	err := http.ListenAndServe(
-		":8080",
-		h2c.NewHandler(mux, &http2.Server{}),
+	grpcServer := &http.Server{
+		Handler: h2c.NewHandler(mux, &http2.Server{}),
+		Addr:    ":8080",
+	}
+
+	go func() {
+		zen.Infow(context.Background(), "starting grpc server at :8080",
+			"outbound_ip", GetOutboundIP(),
 		)
+		grpcServer.ListenAndServe()
+	}()
+
+	group, gropuCtx := errgroup.WithContext(ctx)
+	group.Go(func() error { return serve(gropuCtx, grpcServer) })
+
+	err := group.Wait()
 	if err != nil {
-		panic(err)
+		zen.Errorw(ctx, "failed to serve", "error", err)
+	}
+}
+
+func serve(ctx context.Context, server *http.Server) error {
+	errChan := make(chan error, 1)
+	go func() {
+		err := server.ListenAndServe()
+		if !errors.Is(err, http.ErrServerClosed) {
+			errChan <- err
+			close(errChan)
+		}
+	}()
+	select {
+	case err := <-errChan:
+		return err
+	case <-ctx.Done():
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		return server.Shutdown(ctx)
 	}
 }

core/zerr/err.go

@@ -27,7 +27,7 @@ type Err struct {
 	time          time.Time
 	id            string
 	logger        Logger
-	notifier      Notifier
+	notifier      NotificationHandler
 	sequence      Sequence
 }
 

core/zerr/notifier.go

@@ -0,0 +1,15 @@
+package zerr
+
+import (
+	"context"
+
+	"gitlab.bareksa.com/backend/zen/core/zoptions"
+)
+
+type NotificationHandler interface {
+	NotifyError(ctx context.Context, err Error, opts ...zoptions.NotifyOption)
+}
+
+type NullNotifier struct{}
+
+func (NullNotifier) NotifyError(ctx context.Context, err Error, opts ...zoptions.NotifyOption) {}

core/zerr/wrap.go

@@ -3,8 +3,6 @@ package zerr
 import (
 	"context"
 	"time"
-
-	"gitlab.bareksa.com/backend/zen/core/zoptions"
 )
 
 type WrapInput struct {
@@ -13,7 +11,7 @@ type WrapInput struct {
 	PC       uintptr
 	Time     time.Time
 	Logger   Logger
-	Notifier Notifier
+	Notifier NotificationHandler
 	Details  []any
 }
 
@@ -27,10 +25,6 @@ type Logger interface {
 	LogError(ctx context.Context, err Error)
 }
 
-type Notifier interface {
-	NotifyError(ctx context.Context, err Error, opts ...zoptions.NotifyOption)
-}
-
 type WrapperFunc func(input WrapInput) Error
 
 func (wr WrapperFunc) Wrap(input WrapInput) Error {

core/zlog/caller.go

@@ -1 +0,0 @@
-package zlog

core/zlog/source.go

@@ -0,0 +1,28 @@
+package zlog
+
+import (
+	"log/slog"
+	"os"
+	"strings"
+)
+
+type slogSource struct {
+	*slog.Source
+}
+
+func (sl slogSource) LogValue() slog.Value {
+	wd, _ := os.Getwd()
+	file, found := strings.CutPrefix(sl.File, wd)
+	if found {
+		file = strings.TrimPrefix(file, string(os.PathSeparator))
+	}
+
+	split := strings.Split(sl.Function, string(os.PathSeparator))
+	function := split[len(split)-1]
+
+	return slog.GroupValue(
+		slog.String("file", file),
+		slog.Int("line", sl.Line),
+		slog.String("function", function),
+	)
+}

core/zlog/zlog.go

@@ -31,9 +31,13 @@ type ZLogOptions struct {
 	// If nil, default options are used.
 	HandlerOptions *slog.HandlerOptions
 	// Pretty enables pretty-printing of log messages.
-	Pretty bool
+	//
+	// If nil, it is automatically detected based on the output writer.
+	Pretty *bool
 	// Color enables colorized output.
-	Color bool
+	//
+	// If nil, it is automatically detected based on the output writer.
+	Color *bool
 	// NotificationHandler is a handler that is called when a log message is
 	// intended to be sent to a notification service.
 	//
@@ -48,16 +52,19 @@ func New(w io.Writer, opts *ZLogOptions) *ZLog {
 	if w == nil {
 		panic(errors.New("zlog: New: w is nil"))
 	}
-	pretty := isatty.IsTerminal(os.Stderr.Fd())
 	if opts == nil {
-		opts = defaultOpts
-	} else {
+		opts = &ZLogOptions{}
+	}
+	if opts.Pretty == nil {
+		opts.Pretty = detectPretty(w)
+	}
+	if opts.Color == nil {
+		opts.Color = detectPretty(w)
+	}
 	if opts.HandlerOptions == nil {
 		opts.HandlerOptions = defaultHandlerOptions
-		} else {
-			opts.HandlerOptions.ReplaceAttr = wrapPrettyReplaceAttr(pretty, opts.HandlerOptions.ReplaceAttr)
-		}
 	}
+	opts.HandlerOptions.ReplaceAttr = wrapPrettyReplaceAttr(*opts.Pretty, opts.HandlerOptions.ReplaceAttr)
 	wl := WrapLocker(w)
 
 	return &ZLog{
@@ -101,10 +108,14 @@ func (lo *ZLog) Enabled(ctx context.Context, lvl slog.Level) bool {
 
 // Handle implements the slog.Handler interface.
 func (lo *ZLog) Handle(ctx context.Context, record slog.Record) error {
-	if !lo.opts.Pretty {
+	if !*lo.opts.Pretty {
 		j := lo.jsonPool.Get()
 		defer j.Put()
-		return j.Handle(ctx, record)
+		_ = j.Handle(ctx, record)
+		lo.writer.Lock()
+		defer lo.writer.Unlock()
+		_, err := j.buf.WriteTo(lo.writer)
+		return err
 	}
 
 	var levelColor *color.Color
@@ -144,6 +155,7 @@ func (lo *ZLog) Handle(ctx context.Context, record slog.Record) error {
 
 	if record.Message != "" {
 		buf.WriteString(record.Message)
+		// levelColor.Fprintf(buf, record.Message)
 	}
 
 	buf.WriteByte('\n')
@@ -155,7 +167,7 @@ func (lo *ZLog) Handle(ctx context.Context, record slog.Record) error {
 	if jHandler.buf.Len() > 3 { // Skip empty objects like "{}\n"
 		jsonData := jHandler.buf.Bytes()
 		jsonData = pretty.Pretty(jsonData)
-		if lo.opts.Color {
+		if *lo.opts.Color {
 			jsonData = pretty.Color(jsonData, nil)
 		}
 		buf.Write(jsonData)
@@ -189,18 +201,27 @@ func wrapPrettyReplaceAttr(pretty bool, f replaceAttrFunc) replaceAttrFunc {
 	if f == nil {
 		f = func(s []string, a slog.Attr) slog.Attr { return a }
 	}
-	return func(s []string, a slog.Attr) slog.Attr {
-		if len(s) > 0 {
-			return f(s, a)
+	return func(groups []string, a slog.Attr) slog.Attr {
+		if len(groups) > 0 {
+			return f(groups, a)
 		}
-		if pretty {
-			return f(s, a)
+		if !pretty {
+			switch a.Key {
+			case slog.MessageKey:
+				return slog.Attr{Key: "message", Value: a.Value}
+			case slog.SourceKey:
+				if s, ok := a.Value.Any().(*slog.Source); ok {
+					return slog.Attr{Key: "caller", Value: slogSource{Source: s}.LogValue()}
+				}
+				return slog.Attr{Key: "caller", Value: a.Value}
+			}
+			return f(groups, a)
 		}
 		switch a.Key {
 		case slog.TimeKey, slog.LevelKey, slog.SourceKey, slog.MessageKey:
 			return slog.Attr{}
 		}
-		return f(s, a)
+		return f(groups, a)
 	}
 }
 
@@ -212,16 +233,21 @@ func defaultLevel() slog.Leveler {
 }
 
 var defaultOpts = &ZLogOptions{
-	Pretty:         isatty.IsTerminal(os.Stderr.Fd()),
-	Color:          isatty.IsTerminal(os.Stderr.Fd()),
+	Pretty:         detectPretty(os.Stderr),
+	Color:          detectPretty(os.Stderr),
 	HandlerOptions: defaultHandlerOptions,
 }
 
 var defaultHandlerOptions = &slog.HandlerOptions{
 	AddSource:   true,
 	Level:       defaultLevel(),
-	ReplaceAttr: wrapPrettyReplaceAttr(
-		isatty.IsTerminal(os.Stderr.Fd()),
-		nil,
-	),
+	ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr { return a },
+}
+
+func detectPretty(w io.Writer) *bool {
+	var b bool
+	if fd, ok := w.(interface{ Fd() uintptr }); ok {
+		b = isatty.IsTerminal(fd.Fd())
+	}
+	return &b
 }

core/zlog/zlog_test.go

@@ -0,0 +1,46 @@
+package zlog_test
+
+import (
+	"context"
+	"log/slog"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kinbiko/jsonassert"
+	"gitlab.bareksa.com/backend/zen/core/zcaller"
+	"gitlab.bareksa.com/backend/zen/core/zlog"
+)
+
+func Test_New(t *testing.T) {
+	ja := jsonassert.New(t)
+	output := &strings.Builder{}
+	z := zlog.New(output, nil)
+
+	date := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	record := slog.NewRecord(date, slog.LevelInfo, "test", zcaller.Current())
+	if err := z.Handle(context.Background(), record); err != nil {
+		t.Fatal(err)
+	}
+
+	const want = `{
+		"level": "INFO",
+		"message": "test",
+		"time": "<<PRESENCE>>",
+		"caller": {
+			"file": "zlog_test.go",
+			"line": 22,
+			"function": "zlog_test.Test_New"
+		}
+	}`
+	got := strings.TrimSpace(output.String())
+	defer func() {
+		if t.Failed() {
+			t.Log("got:", got)
+			t.Log("want:", want)
+		}
+	}()
+
+	ja.Assertf(got, want)
+}

core/znotify/null.go

@@ -0,0 +1,15 @@
+package znotify
+
+import (
+	"context"
+	"log/slog"
+
+	"gitlab.bareksa.com/backend/zen/core/zerr"
+	"gitlab.bareksa.com/backend/zen/core/zoptions"
+)
+
+type Null struct{}
+
+func (Null) NotifyError(ctx context.Context, err zerr.Error, opts ...zoptions.NotifyOption) {}
+
+func (Null) NotifyLog(ctx context.Context, record slog.Record, options ...zoptions.NotifyOption) {}

core/znotify/znotify.go

@@ -0,0 +1,17 @@
+package znotify
+
+import (
+	"gitlab.bareksa.com/backend/zen/core/zerr"
+	"gitlab.bareksa.com/backend/zen/core/zlog"
+)
+
+type Notifier interface {
+	zerr.NotificationHandler
+	zlog.NotificationHandler
+}
+
+var DefaultNotifier Notifier = Null{}
+
+func SetDefaultNotifier(notifier Notifier) {
+	DefaultNotifier = notifier
+}

core/ztelemetry/service.go

@@ -1,3 +1,6 @@
 package ztelemetry
 
-type ServiceMetadata struct{}
+type Service struct {
+	Name string
+	Type string
+}

core/ztower/ztower.go

@@ -0,0 +1 @@
+package ztower

error.go

@@ -0,0 +1,50 @@
+package zen
+
+import (
+	"fmt"
+	"time"
+
+	"gitlab.bareksa.com/backend/zen/core/zcaller"
+	"gitlab.bareksa.com/backend/zen/core/zerr"
+	"gitlab.bareksa.com/backend/zen/core/zlog"
+	"gitlab.bareksa.com/backend/zen/core/znotify"
+)
+
+func Wrap(err error, message string) zerr.Error {
+	input := zerr.WrapInput{
+		Errors:   []error{err},
+		Message:  message,
+		PC:       zcaller.Get(3),
+		Time:     time.Now(),
+		Logger:   zlog.Logger,
+		Notifier: zerr.NullNotifier{},
+		Details:  []any{},
+	}
+	return zerr.Wrap(input)
+}
+
+func Wrapf(err error, message string, args ...any) zerr.Error {
+	input := zerr.WrapInput{
+		Errors:   []error{err},
+		Message:  fmt.Sprintf(message, args...),
+		PC:       zcaller.Get(3),
+		Time:     time.Now(),
+		Logger:   zlog.Logger,
+		Notifier: znotify.DefaultNotifier,
+		Details:  []any{},
+	}
+	return zerr.Wrap(input)
+}
+
+func Wrapw(err error, message string, fields ...any) zerr.Error {
+	input := zerr.WrapInput{
+		Errors:   []error{err},
+		Message:  message,
+		PC:       zcaller.Get(3),
+		Time:     time.Now(),
+		Logger:   zlog.Logger,
+		Notifier: znotify.DefaultNotifier,
+		Details:  fields,
+	}
+	return zerr.Wrap(input)
+}

go.mod

@@ -3,13 +3,16 @@ module gitlab.bareksa.com/backend/zen
 go 1.23
 
 require (
+	buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.34.2-20240717164558-a6c49f84cc0f.2
 	connectrpc.com/connect v1.16.2
 	connectrpc.com/grpcreflect v1.2.0
 	github.com/fatih/color v1.17.0
+	github.com/kinbiko/jsonassert v1.1.1
 	github.com/mattn/go-isatty v0.0.20
 	github.com/pborman/indent v1.2.1
 	github.com/tidwall/pretty v1.2.1
 	golang.org/x/net v0.23.0
+	golang.org/x/sync v0.8.0
 	google.golang.org/protobuf v1.34.2
 )
 

go.sum

@@ -1,3 +1,5 @@
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.34.2-20240717164558-a6c49f84cc0f.2 h1:SZRVx928rbYZ6hEKUIN+vtGDkl7uotABRWGY4OAg5gM=
+buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go v1.34.2-20240717164558-a6c49f84cc0f.2/go.mod h1:ylS4c28ACSI59oJrOdW4pHS4n0Hw4TgSPHn8rpHl4Yw=
 connectrpc.com/connect v1.16.2 h1:ybd6y+ls7GOlb7Bh5C8+ghA6SvCBajHwxssO2CGFjqE=
 connectrpc.com/connect v1.16.2/go.mod h1:n2kgwskMHXC+lVqb18wngEpF95ldBHXjZYJussz5FRc=
 connectrpc.com/grpcreflect v1.2.0 h1:Q6og1S7HinmtbEuBvARLNwYmTbhEGRpHDhqrPNlmK+U=
@@ -6,6 +8,8 @@ github.com/fatih/color v1.17.0 h1:GlRw1BRJxkpqUCBKzKOw098ed57fEsKeNjpTe3cSjK4=
 github.com/fatih/color v1.17.0/go.mod h1:YZ7TlrGPkiz6ku9fK3TLD/pl3CpsiFyu8N92HLgmosI=
 github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/kinbiko/jsonassert v1.1.1 h1:DB12divY+YB+cVpHULLuKePSi6+ui4M/shHSzJISkSE=
+github.com/kinbiko/jsonassert v1.1.1/go.mod h1:NO4lzrogohtIdNUNzx8sdzB55M4R4Q1bsrWVdqQ7C+A=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
 github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
@@ -17,6 +21,8 @@ github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 golang.org/x/net v0.23.0 h1:7EYJ93RZ9vYSZAIb2x3lnuvqO5zneoD6IvWjuhfxjTs=
 golang.org/x/net v0.23.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg=
+golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
+golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=

internal/bufferpool/bufferpool.go

@@ -59,6 +59,7 @@ func New(sharedCapacity uint64, bufferCapacity int) *Pool {
 	b := &Pool{
 		maxSharedCapacity: sharedCapacity,
 		maxBufferCapacity: bufferCapacity,
+		pool:              &sync.Pool{},
 	}
 	b.pool.New = func() any {
 		return &Buffer{&bytes.Buffer{}, b}