package jobqueue import ( "context" "fmt" "time" "github.com/redis/go-redis/v9" ) type JobStatus string const ( StatusPending JobStatus = "pending" StatusProcessing JobStatus = "processing" StatusCompleted JobStatus = "completed" StatusFailed JobStatus = "failed" ) type Job struct { ID string `json:"id"` Status JobStatus `json:"status"` Total int `json:"total"` Imported int `json:"imported"` Skipped int `json:"skipped"` Progress int `json:"progress"` Errors []string `json:"errors,omitempty"` CreatedAt time.Time `json:"created_at"` FinishedAt *time.Time `json:"finished_at,omitempty"` Metadata map[string]string `json:"metadata,omitempty"` } type JobQueue struct { Redis *redis.Client } func NewJobQueue(redis *redis.Client) *JobQueue { return &JobQueue{Redis: redis} } func (jq *JobQueue) CreateJob(ctx context.Context, jobID string, total int, metadata map[string]string) error { key := fmt.Sprintf("import_job:%s", jobID) now := time.Now() job := map[string]interface{}{ "id": jobID, "status": string(StatusPending), "total": total, "imported": 0, "skipped": 0, "progress": 0, "created_at": now.Format(time.RFC3339), } if metadata != nil { for k, v := range metadata { job["meta_"+k] = v } } if err := jq.Redis.HSet(ctx, key, job).Err(); err != nil { return err } jq.Redis.Expire(ctx, key, 24*time.Hour) return nil } func (jq *JobQueue) UpdateProgress(ctx context.Context, jobID string, imported int, skipped int, errors []string) error { key := fmt.Sprintf("import_job:%s", jobID) job, err := jq.Redis.HGetAll(ctx, key).Result() if err != nil { return err } total := parseInt(job["total"]) progress := 0 if total > 0 { progress = ((imported + skipped) * 100) / total } update := map[string]interface{}{ "status": string(StatusProcessing), "imported": imported, "skipped": skipped, "progress": progress, } if len(errors) > 0 { update["errors"] = joinErrors(errors) } return jq.Redis.HSet(ctx, key, update).Err() } func (jq *JobQueue) CompleteJob(ctx context.Context, jobID string, imported int, skipped int, errors []string) error { key := fmt.Sprintf("import_job:%s", jobID) now := time.Now() update := map[string]interface{}{ "status": string(StatusCompleted), "imported": imported, "skipped": skipped, "progress": 100, "finished_at": now.Format(time.RFC3339), } if len(errors) > 0 { update["errors"] = joinErrors(errors) } return jq.Redis.HSet(ctx, key, update).Err() } func (jq *JobQueue) FailJob(ctx context.Context, jobID string, errMsg string) error { key := fmt.Sprintf("import_job:%s", jobID) now := time.Now() return jq.Redis.HSet(ctx, key, map[string]interface{}{ "status": string(StatusFailed), "finished_at": now.Format(time.RFC3339), "errors": errMsg, }).Err() } func (jq *JobQueue) GetJob(ctx context.Context, jobID string) (*Job, error) { key := fmt.Sprintf("import_job:%s", jobID) data, err := jq.Redis.HGetAll(ctx, key).Result() if err != nil { return nil, err } if len(data) == 0 { return nil, fmt.Errorf("job not found") } job := &Job{ ID: data["id"], Status: JobStatus(data["status"]), Total: parseInt(data["total"]), Imported: parseInt(data["imported"]), Skipped: parseInt(data["skipped"]), Progress: parseInt(data["progress"]), } if data["errors"] != "" { job.Errors = splitErrors(data["errors"]) } if t, err := time.Parse(time.RFC3339, data["created_at"]); err == nil { job.CreatedAt = t } if data["finished_at"] != "" { if t, err := time.Parse(time.RFC3339, data["finished_at"]); err == nil { job.FinishedAt = &t } } return job, nil } func (jq *JobQueue) DeleteJob(ctx context.Context, jobID string) error { key := fmt.Sprintf("import_job:%s", jobID) return jq.Redis.Del(ctx, key).Err() } func parseInt(s string) int { n := 0 for _, c := range s { if c >= '0' && c <= '9' { n = n*10 + int(c-'0') } } return n } func joinErrors(errors []string) string { result := "" for i, e := range errors { if i > 0 { result += "||" } result += e } return result } func splitErrors(s string) []string { result := []string{} current := "" for _, c := range s { if c == '|' { if current != "" { result = append(result, current) } current = "" } else { current += string(c) } } if current != "" { result = append(result, current) } return result }