Updated

Author: djthorpe

pkg/types/ptr.go

@@ -67,6 +67,14 @@ func PtrBool(v *bool) bool {
 	return *v
 }
 
+// PtrTime returns a pointer to a time.Time
+func TimePtr(t time.Time) *time.Time {
+	if t.IsZero() {
+		return nil
+	}
+	return &t
+}
+
 // PtrTime returns a time.Time from a pointer
 func PtrTime(t *time.Time) time.Time {
 	if t == nil {

plugin.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"net/http"
+	"time"
 
 	// Packages
 	pg "github.com/djthorpe/go-pg"
@@ -103,6 +104,9 @@ type PGQueue interface {
 	// Register a queue with a callback, and return the registered queue
 	RegisterQueue(context.Context, schema.Queue, PGCallback) (*schema.Queue, error)
 
+	// Create a task for a queue with a payload and optional delay, and return it
+	CreateTask(context.Context, string, any, time.Duration) (*schema.Task, error)
+
 	// Delete a ticker by name
 	DeleteTicker(context.Context, string) error
 }

plugin/pgqueue/task.go

@@ -82,6 +82,7 @@ FOR_LOOP:
 		case ticker := <-tickerch:
 			t.execTicker(ctx, ticker, errch)
 		case err := <-errch:
+			// TODO: Handle errors, for now, just log them
 			log.Println("ERROR:", err)
 		}
 	}
@@ -140,6 +141,18 @@ func (t *task) RegisterQueue(ctx context.Context, meta schema.Queue, fn server.P
 	return queue, nil
 }
 
+// Push a task onto a queue, and return the task
+func (t *task) CreateTask(ctx context.Context, queue string, payload any, delay time.Duration) (*schema.Task, error) {
+	var delayedAt *time.Time
+	if delay > 0 {
+		delayedAt = types.TimePtr(time.Now().Add(delay))
+	}
+	return t.client.CreateTask(ctx, queue, schema.TaskMeta{
+		Payload:   payload,
+		DelayedAt: delayedAt,
+	})
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // PRIVATE METHODS