Updated

Author: djthorpe

cmd/server/service.go

@@ -92,6 +92,13 @@ func (cmd *ServiceRunCommand) Run(app server.Cmd) error {
 				config.Pool = pool
 			}
 
+			// Set the queue
+			if queue, ok := ref.Provider(ctx).Task(ctx, "pgqueue").(server.PGQueue); !ok || queue == nil {
+				return nil, httpresponse.ErrInternalError.Withf("Invalid connection pool %q", "pgqueue")
+			} else {
+				config.Queue = queue
+			}
+
 			return config, nil
 
 		case "httpserver":

pkg/cert/config/config.go

@@ -10,6 +10,9 @@ import (
 	certhandler "github.com/mutablelogic/go-server/pkg/cert/handler"
 	schema "github.com/mutablelogic/go-server/pkg/cert/schema"
 	httpresponse "github.com/mutablelogic/go-server/pkg/httpresponse"
+	pgqueue "github.com/mutablelogic/go-server/pkg/pgqueue/schema"
+	"github.com/mutablelogic/go-server/pkg/ref"
+	"github.com/mutablelogic/go-server/pkg/types"
 )
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -28,6 +31,7 @@ type Config struct {
 		StreetAddress string `name:"street" help:"Street address"`
 		PostalCode    string `name:"postal" help:"Postal code"`
 	} `embed:"" prefix:"root."`
+	Queue server.PGQueue `kong:"-"` // Connection to queue
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -67,6 +71,17 @@ func (c Config) New(ctx context.Context) (server.Task, error) {
 		certhandler.RegisterCert(ctx, c.Router, c.Prefix, certmanager)
 	}
 
+	// Queue task to check for SSL expiry
+	if c.Queue != nil {
+		c.Queue.RegisterTicker(ctx, pgqueue.TickerMeta{
+			Ticker:   c.Name(),
+			Interval: types.DurationPtr(time.Minute),
+		}, func(ctx context.Context, _ any) error {
+			ref.Log(ctx).Print(ctx, "Checking for SSL expiry...")
+			return nil
+		})
+	}
+
 	// Return the task
 	return newTaskWith(certmanager), nil
 }

pkg/pgqueue/config/task.go

@@ -130,7 +130,11 @@ func (t *task) RegisterTicker(ctx context.Context, meta schema.TickerMeta, fn se
 	if err != nil {
 		return nil, err
 	}
-	// TODO: Register the ticker callback
+
+	// Register the ticker callback
+	t.registerCallback(ticker.Namespace, ticker.Ticker, fn)
+
+	// Return the ticker metadata
 	return ticker, nil
 }
 
@@ -141,14 +145,29 @@ func (t *task) RegisterQueue(ctx context.Context, meta schema.QueueMeta, fn serv
 	if err != nil {
 		return nil, err
 	}
+
 	// Register a queue cleanup timer
+	// TODO: queue name should include the namespace
 	if _, err := t.manager.RegisterTickerNs(ctx, schema.CleanupNamespace, schema.TickerMeta{
 		Ticker:   queue.Queue,
 		Interval: queue.TTL,
 	}); err != nil {
 		_, err_ := t.manager.DeleteQueue(ctx, meta.Queue)
 		return nil, errors.Join(err, err_)
+	} else {
+		t.registerCallback(schema.CleanupNamespace, queue.Queue, func(ctx context.Context, _ any) error {
+			// Cleanup the queue
+			if _, err := t.manager.CleanQueue(ctx, queue.Queue); err != nil {
+				return err
+			}
+			return nil
+		})
 	}
+
+	// Register the task callback
+	t.registerCallback(queue.Namespace, queue.Queue, fn)
+
+	// Return the queue metadata
 	return queue, nil
 }