package sync 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 { log.Printf("Failed to start asynq server: %v", err) } }