package sync import ( "context" "encoding/json" "fmt" "github.com/hibiken/asynq" ) // unmarshalPayload is a generic function to unmarshal task payloads func unmarshalPayload[T any](task *asynq.Task) (*T, error) { var payload T if err := json.Unmarshal(task.Payload(), &payload); err != nil { return nil, fmt.Errorf("failed to unmarshal payload: %w", err) } return &payload, nil } // HandleFullSync handles the full sync task func (s *SyncJob) HandleFullSync(ctx context.Context, t *asynq.Task) error { _, err := unmarshalPayload[SyncPayload](t) if err != nil { return err } return s.RunFullSync(ctx) } // HandleEntitySync handles the entity sync task func (s *SyncJob) HandleEntitySync(ctx context.Context, t *asynq.Task) error { payload, err := unmarshalPayload[EntitySyncPayload](t) if err != nil { return err } return s.syncEntities(payload.ClassName, ctx) } // HandleEdgeSync handles the edge sync task func (s *SyncJob) HandleEdgeSync(ctx context.Context, t *asynq.Task) error { payload, err := unmarshalPayload[EdgeSyncPayload](t) if err != nil { return err } batchSize := payload.BatchSize if batchSize <= 0 { batchSize = DefaultBatchSize } return s.SyncEdgesBatch(ctx, batchSize, payload.Offset) }