Merge pull request #93 from mutablelogic/v5

Various updates

Author: djthorpe

go.mod

@@ -5,7 +5,7 @@ go 1.23.5
 require (
 	github.com/alecthomas/kong v1.10.0
 	github.com/djthorpe/go-marshaler v1.0.0
-	github.com/djthorpe/go-pg v1.0.5
+	github.com/djthorpe/go-pg v1.0.6
 	github.com/golang-jwt/jwt/v5 v5.2.2
 	github.com/mutablelogic/go-client v1.0.12
 	github.com/stretchr/testify v1.10.0
@@ -32,8 +32,6 @@ require (
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/hashicorp/errwrap v1.1.0 // indirect
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/jackc/pgx/v5 v5.7.3 // indirect

go.sum

@@ -32,10 +32,10 @@ github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5Qvfr
 github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/djthorpe/go-errors v1.0.3 h1:GZeMPkC1mx2vteXLI/gvxZS0Ee9zxzwD1mcYyKU5jD0=
 github.com/djthorpe/go-errors v1.0.3/go.mod h1:HtfrZnMd6HsX75Mtbv9Qcnn0BqOrrFArvCaj3RMnZhY=
-github.com/djthorpe/go-marshaler v0.0.15 h1:ZXq5YHCsbREbbYJtc0ie9hz7HJ7vIeeDlMbe7cGh0C0=
-github.com/djthorpe/go-marshaler v0.0.15/go.mod h1:xCXhTzj52UL3YStRsqUSfrKses7ofmfTXYQfVedn8Lw=
-github.com/djthorpe/go-pg v1.0.5 h1:UYCV5fSXOJEFTafem1wB57RK2J0V7Nr9nCquC5sO+ZE=
-github.com/djthorpe/go-pg v1.0.5/go.mod h1:XHl/w8+66Hs746nOYd+gdjqPImNuLVZ5UsXLI47rb4c=
+github.com/djthorpe/go-marshaler v1.0.0 h1:dLIjr2vXXUi8VF93LIQPHfBjW3USbtYjqKt+b2WTf2M=
+github.com/djthorpe/go-marshaler v1.0.0/go.mod h1:xCXhTzj52UL3YStRsqUSfrKses7ofmfTXYQfVedn8Lw=
+github.com/djthorpe/go-pg v1.0.6 h1:v/ZcMhtgULa301LPYyUEo7fJJwuseOaAdpCyKid0qbU=
+github.com/djthorpe/go-pg v1.0.6/go.mod h1:XHl/w8+66Hs746nOYd+gdjqPImNuLVZ5UsXLI47rb4c=
 github.com/docker/docker v27.1.1+incompatible h1:hO/M4MtV36kzKldqnA37IWhebRA+LnqqcqDja6kVaKY=
 github.com/docker/docker v27.1.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
@@ -64,9 +64,6 @@ github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
-github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
-github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
-github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
 github.com/hexops/gotextdiff v1.0.3/go.mod h1:pSWU5MAI3yDq+fZBTazCSJysOMbxWL1BSow5/V2vxeg=

pkg/httprouter/httprouter.go

@@ -63,18 +63,20 @@ func (r *router) Run(ctx context.Context) error {
 // HTTP ROUTER
 
 // Register a function to handle a URL path
-func (r *router) HandleFunc(ctx context.Context, prefix string, fn http.HandlerFunc) {
+func (r *router) HandleFunc(parent context.Context, prefix string, fn http.HandlerFunc) {
 	// Wrap the function with middleware
 	for _, middleware := range r.middleware {
 		fn = middleware.HandleFunc(fn)
 	}
 
 	// Apply middleware, set context
-	ref.Log(ctx).Debug(ctx, "Register route: ", types.JoinPath(r.prefix, prefix))
+	ref.Log(parent).Debug(parent, "Register route: ", types.JoinPath(r.prefix, prefix))
 	r.ServeMux.HandleFunc(types.JoinPath(r.prefix, prefix), func(w http.ResponseWriter, r *http.Request) {
-		r = r.WithContext(ref.WithLog(r.Context(), ref.Log(ctx)))
-		// TODO: Add Log into the r context, but don't replace the original
-		fn(w, r)
+		// Add provider to context
+		ctx := ref.WithProvider(r.Context(), ref.Provider(parent))
+
+		// Serve the request
+		fn(w, r.WithContext(ctx))
 	})
 }
 
@@ -84,7 +86,7 @@ func (r *router) Origin() string {
 }
 
 // Register serving of static files from a filesystem
-func (r *router) HandleFS(ctx context.Context, prefix string, fs fs.FS) {
+func (r *router) HandleFS(parent context.Context, prefix string, fs fs.FS) {
 	// Create the file server
 	fn := http.StripPrefix(types.JoinPath(r.prefix, prefix), http.FileServer(http.FS(fs))).ServeHTTP
 
@@ -94,12 +96,12 @@ func (r *router) HandleFS(ctx context.Context, prefix string, fs fs.FS) {
 	}
 
 	// Apply middleware
-	ref.Log(ctx).Debug(ctx, "Register route: ", types.JoinPath(r.prefix, prefix))
+	ref.Log(parent).Debug(parent, "Register static: ", types.JoinPath(r.prefix, prefix))
 	r.ServeMux.HandleFunc(types.JoinPath(r.prefix, prefix), func(w http.ResponseWriter, req *http.Request) {
 		// Set CORS headers
 		httpresponse.Cors(w, req, r.origin, http.MethodGet)
 
 		// Call the file server
-		fn(w, req.WithContext(ref.WithLog(ctx, ref.Log(ctx))))
+		fn(w, req.WithContext(ref.WithProvider(parent, ref.Provider(parent))))
 	})
 }

pkg/pgqueue/config/task.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"reflect"
+	"regexp"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -42,6 +44,8 @@ func NewTask(manager *pgqueue.Manager, threads uint) (server.Task, error) {
 	self.callbacks = make(map[string]server.PGCallback, 100)
 	self.decoder = marshaler.NewDecoder("json",
 		convertPtr,
+		convertPGTime,
+		convertPGDuration,
 		convertFloatToIntUint,
 		marshaler.ConvertTime,
 		marshaler.ConvertDuration,
@@ -61,7 +65,7 @@ func (task *task) Run(parent context.Context) error {
 
 	// Create a cancelable context
 	ctx, cancel := context.WithCancel(context.Background())
-	ctx = ref.WithPath(ref.WithLog(ctx, ref.Log(parent)), ref.Path(parent)...)
+	ctx = ref.WithPath(ref.WithProvider(ctx, ref.Provider(parent)), ref.Path(parent)...)
 
 	// Ticker loop
 	tickerch := make(chan *schema.Ticker)
@@ -147,7 +151,9 @@ FOR_LOOP:
 					}
 					n += len(tasks)
 				}
-				ref.Log(ctx).With("ticker", evt).Debug(parent, "removed ", n, " tasks from queue")
+				if n > 0 {
+					ref.Log(ctx).With("ticker", evt).Debug(parent, "removed ", n, " tasks from queue")
+				}
 			}
 		}
 	}
@@ -167,6 +173,11 @@ func (t *task) Conn() pg.PoolConn {
 	return t.manager.Conn()
 }
 
+// Namespace returns the namespace of the queue.
+func (t *task) Namespace() string {
+	return t.manager.Namespace()
+}
+
 // RegisterTicker registers a periodic task (ticker) with a callback function.
 // It returns the metadata of the registered ticker.
 func (t *task) RegisterTicker(ctx context.Context, meta schema.TickerMeta, fn server.PGCallback) (*schema.Ticker, error) {
@@ -255,7 +266,8 @@ func (t *task) tryTask(ctx context.Context, taskpool *pgqueue.TaskPool, task *sc
 	taskpool.RunTask(ctx, task, t.getTaskCallback(task), func(err error) {
 		var status string
 		delta := time.Since(now).Truncate(time.Millisecond)
-		if _, err_ := t.manager.ReleaseTask(context.TODO(), task.Id, err == nil, err, &status); err_ != nil {
+		child := ref.WithPath(ref.WithProvider(context.TODO(), ref.Provider(ctx)), ref.Path(ctx)...)
+		if _, err_ := t.manager.ReleaseTask(child, task.Id, err == nil, err, &status); err_ != nil {
 			err = errors.Join(err, err_)
 		}
 		switch {
@@ -293,16 +305,75 @@ func joinName(parts ...string) string {
 	return strings.Join(parts, namespaceSeparator)
 }
 
-func splitName(name string, n int) []string {
-	return strings.SplitN(name, namespaceSeparator, n)
-}
-
 // //////////////////////////////////////////////////////////////////////////////
 // PRIVATE METHODS
+
 var (
-	nilValue = reflect.ValueOf(nil)
+	nilValue           = reflect.ValueOf(nil)
+	timeType           = reflect.TypeOf(time.Time{})
+	durationType       = reflect.TypeOf(time.Duration(0))
+	rePostgresDuration = regexp.MustCompile(`^(\d+):(\d+):(\d+)$`)
 )
 
+// convertPGTime returns time from postgres format
+func convertPGTime(src reflect.Value, dest reflect.Type) (reflect.Value, error) {
+	// Pass value through
+	if src.Type() == dest {
+		return src, nil
+	}
+
+	if dest == timeType || dest.Kind() == reflect.Ptr && dest.Elem() == timeType {
+		var v reflect.Value
+
+		// Convert time 2025-05-03T17:29:32.329803 => time.Time
+		if t, err := time.Parse("2006-01-02T15:04:05.999999999", src.String()); err == nil {
+			v = reflect.ValueOf(t)
+		} else if t, err := time.Parse("2006-01-02T15:04:05.999999999Z", src.String()); err == nil {
+			v = reflect.ValueOf(t)
+		}
+
+		// Return value
+		if v.IsValid() {
+			if dest.Kind() == reflect.Ptr {
+				value := reflect.New(dest.Elem())
+				value.Elem().Set(v)
+				return value, nil
+			} else {
+				return v, nil
+			}
+		}
+	}
+
+	// Skip
+	return nilValue, nil
+}
+
+// convertPGDuration returns duration from postgres format
+func convertPGDuration(src reflect.Value, dest reflect.Type) (reflect.Value, error) {
+	// Pass value through
+	if src.Type() == dest {
+		return src, nil
+	}
+
+	if dest == durationType {
+		// Convert 00:00:00 => time.Duration
+		if parts := rePostgresDuration.FindStringSubmatch(src.String()); len(parts) == 4 {
+			if hours, err := strconv.ParseUint(parts[1], 10, 64); err != nil {
+				return nilValue, err
+			} else if minutes, err := strconv.ParseUint(parts[2], 10, 64); err != nil {
+				return nilValue, err
+			} else if seconds, err := strconv.ParseUint(parts[3], 10, 64); err != nil {
+				return nilValue, err
+			} else {
+				return reflect.ValueOf(time.Duration(hours)*time.Hour + time.Duration(minutes)*time.Minute + time.Duration(seconds)*time.Second), nil
+			}
+		}
+	}
+
+	// Skip
+	return nilValue, nil
+}
+
 // convertPtr returns value if pointer
 func convertPtr(src reflect.Value, dest reflect.Type) (reflect.Value, error) {
 	// Pass value through

pkg/pgqueue/manager.go

@@ -13,6 +13,7 @@ import (
 	pg "github.com/djthorpe/go-pg"
 	httpresponse "github.com/mutablelogic/go-server/pkg/httpresponse"
 	schema "github.com/mutablelogic/go-server/pkg/pgqueue/schema"
+	ref "github.com/mutablelogic/go-server/pkg/ref"
 	types "github.com/mutablelogic/go-server/pkg/types"
 )
 
@@ -462,14 +463,15 @@ func (manager *Manager) RunTaskLoop(ctx context.Context, ch chan<- *schema.Task)
 
 // RunNotificationLoop runs a loop to process database notifications, until the context is cancelled
 // or an error occurs.
-func (manager *Manager) RunNotificationLoop(ctx context.Context, ch chan<- *pg.Notification) error {
+func (manager *Manager) RunNotificationLoop(parent context.Context, ch chan<- *pg.Notification) error {
 	// Subscribe to topics
 	for _, topic := range manager.topics {
-		if err := manager.listener.Listen(ctx, topic); err != nil {
+		if err := manager.listener.Listen(parent, topic); err != nil {
 			return err
 		}
 	}
 	defer func() {
+		ctx := ref.WithProvider(context.Background(), ref.Provider(parent))
 		for _, topic := range manager.topics {
 			manager.listener.Unlisten(ctx, topic)
 		}
@@ -478,10 +480,10 @@ func (manager *Manager) RunNotificationLoop(ctx context.Context, ch chan<- *pg.N
 	// Loop until context is cancelled
 	for {
 		select {
-		case <-ctx.Done():
+		case <-parent.Done():
 			return nil
 		default:
-			if notification, err := manager.listener.WaitForNotification(ctx); err != nil {
+			if notification, err := manager.listener.WaitForNotification(parent); err != nil {
 				if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
 					return err
 				}

pkg/ref/context.go

@@ -54,10 +54,10 @@ func Label(ctx context.Context) string {
 }
 
 func Log(ctx context.Context) server.Logger {
-	if value := ctx.Value(ctxLogger); value == nil {
+	if provider := ctx.Value(ctxProvider); provider == nil {
 		return nil
 	} else {
-		return value.(server.Logger)
+		return provider.(server.Logger)
 	}
 }
 
@@ -93,10 +93,6 @@ func Task(ctx context.Context) *pgqueue.Task {
 	}
 }
 
-func WithLog(ctx context.Context, logger server.Logger) context.Context {
-	return context.WithValue(ctx, ctxLogger, logger)
-}
-
 func WithAuth(ctx context.Context, auth server.Auth) context.Context {
 	return context.WithValue(ctx, ctxAuth, auth)
 }

pkg/types/hash.go

@@ -0,0 +1,16 @@
+package types
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+)
+
+///////////////////////////////////////////////////////////////////////////////
+// PUBLIC METHODS
+
+func Hash(data []byte) string {
+	// Return the SHA256 hash of the data
+	h := sha256.New()
+	h.Write(data)
+	return hex.EncodeToString(h.Sum(nil))
+}

plugin.go

@@ -119,6 +119,9 @@ type PGQueue interface {
 	// Conn returns the underlying connection pool object.
 	Conn() pg.PoolConn
 
+	// Namespace returns the namespace of the queue.
+	Namespace() string
+
 	// RegisterTicker registers a periodic task (ticker) with a callback function.
 	// It returns the metadata of the registered ticker.
 	RegisterTicker(context.Context, pgschema.TickerMeta, PGCallback) (*pgschema.Ticker, error)