How to determine if jobs are running and debug job implementation code?

[lovgrandma]

For some reason my job queue only goes so far as my ProvisionSmsJob function which is being called after receiving gRPC message from another server. It goes so far as to complete line

log.Printf("Enqueued Sms Delivery task: %v %v %v %v %v %v %v", info.ID, info.Queue, info.State, info.Timeout, info.LastErr, info.Type, info)

and saves the job on my redis db in a way that looks like this "asynq:{default}:t:903e04db-51fe-43fe-bae1-c431dce005c0"

But I never see my HandleSmsDeliveryTask function run based on this log never occurs log.Printf("Beginning Sms Blast Delivery for From: %v, Content: %v", p.From, p.Content)

Is this because the worker will not log to the console I am viewing? I kind of use this way to code as Im not keen on guessing how to loop through each record and run a twilio function

import (
	"tycoon.systems/tycoon-services/s3credentials"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"context"
	"tycoon.systems/tycoon-services/structs"
	"github.com/hibiken/asynq"
	"reflect"
	"encoding/json"
	"fmt"
	"log"
)

var (
	uri = s3credentials.GetS3Data("mongo", "addressAuth", "")
	credential = options.Credential{
		AuthSource: "admin",
		Username: s3credentials.GetS3Data("mongo", "u", ""),
		Password: s3credentials.GetS3Data("mongo", "p", ""),
	}
	clientOpts = options.Client().ApplyURI(uri).
   		SetAuth(credential)
	client, err = mongo.Connect(context.TODO(), clientOpts)
	jobQueueAddr = s3credentials.GetS3Data("redis", "redishost", "") + ":" + s3credentials.GetS3Data("redis", "tycoon_systems_queue_port", "")
	srv = asynq.NewServer(
		asynq.RedisClientOpt{Addr: jobQueueAddr},
		asynq.Config{
			Concurrency: 10, // total concurrent workers
			Queues: map[string]int{
				"critical": 6,
				"default":  3,
				"low":      1,
			},
		},
	)
)

const (
	TypeSmsDelivery		= "sms:deliver"
)

func main() {
	mux := asynq.NewServeMux()
	mux.HandleFunc(TypeSmsDelivery, HandleSmsDeliveryTask)
	if err := srv.Run(mux); err != nil {
		log.Printf("Could not run Job Queue Server: %v", err)
	}
}

func ProvisionSmsJob(msg structs.Msg) string {
	if reflect.TypeOf(msg.Content).Kind() == reflect.String &&
	reflect.TypeOf(msg.From).Kind() == reflect.String &&
	reflect.TypeOf(msg.JobId).Kind() == reflect.String {
		jobClient := asynq.NewClient(asynq.RedisClientOpt{Addr: jobQueueAddr})
		defer jobClient.Close()
		task, err := NewSmsDeliveryTask(msg)
		if err != nil {
			log.Printf("Could not create Sms Delivery task: %v", err)
		}
		info, err := jobClient.Enqueue(task)
		if err != nil {
			log.Printf("Could not create Sms Delivery task: %v", err)
		}
		log.Printf("Enqueued Sms Delivery task: %v %v %v %v %v %v %v", info.ID, info.Queue, info.State, info.Timeout, info.LastErr, info.Type, info)
		return info.ID
	}
	return "failed"
}

func GetConnection() *mongo.Client {
	return client
}

// Build new delivery to be consumed by queue
func NewSmsDeliveryTask(msg structs.Msg) (*asynq.Task, error) {
	payload, err := json.Marshal(msg)
	if err != nil {
		return nil, err
	}
	return asynq.NewTask(TypeSmsDelivery, payload), nil
}

// Unmarshal queued delivery task to determine if in correct format
func HandleSmsDeliveryTask(ctx context.Context, t *asynq.Task) error {
	var p structs.Msg
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
	}
	log.Printf("Beginning Sms Blast Delivery for From: %v, Content: %v", p.From, p.Content)
	err := PerformSmsDelivery(p)
	if err != nil {
		return fmt.Errorf("Perform Sms Delivery failed: %v: %w", err, asynq.SkipRetry)
	}
	return nil
}

func PerformSmsDelivery(msg structs.Msg) error {
	if client == nil {
		return fmt.Errorf("Database client connection unavailable %v", err)
	}
	coll := client.Database("mpst").Collection("numbers")
	record := &structs.Number{}
	err := coll.FindOne(context.TODO(), bson.D{{"from", msg.From}}).Decode(&record)
	if err != nil {
		return fmt.Errorf("failure %v", err)
	}
	fmt.Printf("Record for Job %v", record)
	return nil
}```

[lovgrandma]

I have since ran the worker functionality to create the server in a separate file with the code below and simply ran it as a go routine from the main() function from the main server application file. Seems to work.

srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: jobQueueAddr},
		asynq.Config{
			Concurrency: 10, // total concurrent workers
			Queues: map[string]int{
				"critical": 6,
				"default":  3,
				"low":      1,
			},
		},
	)
	mux := asynq.NewServeMux()
	mux.HandleFunc(sms_queue.TypeSmsDelivery, sms_queue.HandleSmsDeliveryTask)
	if err := srv.Run(mux); err != nil {
		log.Printf("Could not run Job Queue Server: %v", err)
	}