tercul-backend/syncjob/queue.go

package syncjob

import (
	"encoding/json"
	"log"
	"time"

	"github.com/hibiken/asynq"
)

// enqueueTask is a generic function to enqueue tasks with payload
func enqueueTask[T any](client *asynq.Client, taskType string, payload T, delay time.Duration) error {
	data, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	task := asynq.NewTask(taskType, data)
	_, err = client.Enqueue(task, asynq.ProcessIn(delay))
	if err != nil {
		return err
	}

	return nil
}

// EnqueueFullSync enqueues a full sync task
func EnqueueFullSync(client *asynq.Client) error {
	payload := SyncPayload{Full: true}
	if err := enqueueTask(client, TaskFullSync, payload, DefaultFullSyncDelay); err != nil {
		return err
	}
	log.Println("Enqueued full sync task.")
	return nil
}

// EnqueueEntitySync enqueues an entity sync task
func EnqueueEntitySync(client *asynq.Client, className string) error {
	payload := EntitySyncPayload{ClassName: className}
	if err := enqueueTask(client, TaskEntitySync, payload, DefaultEntityDelay); err != nil {
		return err
	}
	log.Printf("Enqueued entity sync task for class %s.", className)
	return nil
}

// EnqueueEdgeSync enqueues an edge sync task
func EnqueueEdgeSync(client *asynq.Client, batchSize, offset int) error {
	payload := EdgeSyncPayload{
		BatchSize: batchSize,
		Offset:    offset,
	}
	if err := enqueueTask(client, TaskEdgeSync, payload, DefaultEdgeDelay); err != nil {
		return err
	}
	log.Printf("Enqueued edge sync task (offset %d, batch size %d).", offset, batchSize)
	return nil
}

// RegisterQueueHandlers registers all sync job handlers with the Asynq server
func RegisterQueueHandlers(srv *asynq.Server, syncJob *SyncJob) {
	mux := asynq.NewServeMux()
	mux.HandleFunc(TaskFullSync, syncJob.HandleFullSync)
	mux.HandleFunc(TaskEntitySync, syncJob.HandleEntitySync)
	mux.HandleFunc(TaskEdgeSync, syncJob.HandleEdgeSync)
	if err := srv.Run(mux); err != nil {
		panic(err)
	}
}