Janady13/seo-sidecar / megamind/seo/schema_job_queue.go

1package seo 2 3import ( 4 "container/heap" 5 "encoding/json" 6 "os" 7 "sync" 8 "time" 9 10 "github.com/google/uuid" 11) 12 13// JobQueue manages schema processing jobs. 14type JobQueue struct { 15 mu sync.Mutex 16 jobs jobHeap 17 pending map[string]*SchemaJob 18 dedupMap map[string]string // dedup key -> job ID 19 filePath string 20} 21 22// NewJobQueue creates a job queue. 23func NewJobQueue(filePath string) *JobQueue { 24 q := &JobQueue{ 25 jobs: make(jobHeap, 0), 26 pending: make(map[string]*SchemaJob), 27 dedupMap: make(map[string]string), 28 filePath: filePath, 29 } 30 31 heap.Init(&q.jobs) 32 q.load() 33 34 return q 35} 36 37// dedupKey generates a deduplication key for a job. 38func (q *JobQueue) dedupKey(job *SchemaJob) string { 39 schemaType := "" 40 if job.Payload != nil { 41 if st, ok := job.Payload["schemaType"].(string); ok { 42 schemaType = st 43 } 44 } 45 return job.SiteKey + ":" + job.URL + ":" + job.JobType + ":" + schemaType 46} 47 48// Enqueue adds a job to the queue. 49// If a job with the same dedup key exists, it merges by bumping priority. 50func (q *JobQueue) Enqueue(job *SchemaJob) string { 51 q.mu.Lock() 52 defer q.mu.Unlock() 53 54 if job.JobID == "" { 55 job.JobID = uuid.New().String() 56 } 57 if job.ScheduledAt.IsZero() { 58 job.ScheduledAt = time.Now() 59 } 60 if job.MaxAttempts == 0 { 61 job.MaxAttempts = 3 62 } 63 64 // Check for duplicate 65 dedupKey := q.dedupKey(job) 66 if existingID, ok := q.dedupMap[dedupKey]; ok { 67 if existing, found := q.pending[existingID]; found { 68 // Bump priority if new job has higher priority 69 if job.Priority > existing.Priority { 70 existing.Priority = job.Priority 71 } 72 // Reset schedule to now if requested sooner 73 if job.ScheduledAt.Before(existing.ScheduledAt) { 74 existing.ScheduledAt = job.ScheduledAt 75 } 76 return existingID // Return existing job ID 77 } 78 } 79 80 heap.Push(&q.jobs, job) 81 q.pending[job.JobID] = job 82 q.dedupMap[dedupKey] = job.JobID 83 84 return job.JobID 85} 86 87// EnqueueChange creates and enqueues a change detection job. 88func (q *JobQueue) EnqueueChange(siteKey, url string, priority int) string { 89 return q.Enqueue(&SchemaJob{ 90 SiteKey: siteKey, 91 URL: url, 92 JobType: "detect-change", 93 Priority: priority, 94 }) 95} 96 97// EnqueueClassify creates and enqueues a classification job. 98func (q *JobQueue) EnqueueClassify(siteKey, url string, priority int) string { 99 return q.Enqueue(&SchemaJob{ 100 SiteKey: siteKey, 101 URL: url, 102 JobType: "classify-page", 103 Priority: priority, 104 }) 105} 106 107// EnqueueGenerate creates and enqueues a generation job. 108func (q *JobQueue) EnqueueGenerate(siteKey, url, schemaType string, priority int) string { 109 return q.Enqueue(&SchemaJob{ 110 SiteKey: siteKey, 111 URL: url, 112 JobType: "generate-schema", 113 Priority: priority, 114 Payload: map[string]any{"schemaType": schemaType}, 115 }) 116} 117 118// EnqueuePush creates and enqueues a push job. 119func (q *JobQueue) EnqueuePush(siteKey, schemaKey, jsonld string, priority int) string { 120 return q.Enqueue(&SchemaJob{ 121 SiteKey: siteKey, 122 JobType: "push-schema", 123 Priority: priority, 124 Payload: map[string]any{"schemaKey": schemaKey, "jsonld": jsonld}, 125 }) 126} 127 128// Dequeue removes and returns the highest priority job. 129func (q *JobQueue) Dequeue() *SchemaJob { 130 q.mu.Lock() 131 defer q.mu.Unlock() 132 133 if q.jobs.Len() == 0 { 134 return nil 135 } 136 137 job := heap.Pop(&q.jobs).(*SchemaJob) 138 delete(q.pending, job.JobID) 139 delete(q.dedupMap, q.dedupKey(job)) 140 141 return job 142} 143 144// Peek returns the next job without removing it. 145func (q *JobQueue) Peek() *SchemaJob { 146 q.mu.Lock() 147 defer q.mu.Unlock() 148 149 if q.jobs.Len() == 0 { 150 return nil 151 } 152 153 return q.jobs[0] 154} 155 156// Requeue puts a failed job back in the queue. 157func (q *JobQueue) Requeue(job *SchemaJob) bool { 158 q.mu.Lock() 159 defer q.mu.Unlock() 160 161 job.Attempts++ 162 163 if job.Attempts >= job.MaxAttempts { 164 return false // Max retries exceeded 165 } 166 167 // Exponential backoff 168 delay := time.Duration(job.Attempts*job.Attempts) * time.Second 169 job.ScheduledAt = time.Now().Add(delay) 170 171 heap.Push(&q.jobs, job) 172 q.pending[job.JobID] = job 173 174 return true 175} 176 177// Get retrieves a job by ID. 178func (q *JobQueue) Get(jobID string) *SchemaJob { 179 q.mu.Lock() 180 defer q.mu.Unlock() 181 182 return q.pending[jobID] 183} 184 185// Cancel removes a job from the queue. 186func (q *JobQueue) Cancel(jobID string) bool { 187 q.mu.Lock() 188 defer q.mu.Unlock() 189 190 job, ok := q.pending[jobID] 191 if !ok { 192 return false 193 } 194 195 // Mark as cancelled (will be skipped when dequeued) 196 job.MaxAttempts = 0 197 delete(q.pending, jobID) 198 199 return true 200} 201 202// Len returns the queue length. 203func (q *JobQueue) Len() int { 204 q.mu.Lock() 205 defer q.mu.Unlock() 206 207 return q.jobs.Len() 208} 209 210// Ready returns jobs ready to be processed. 211func (q *JobQueue) Ready() []*SchemaJob { 212 q.mu.Lock() 213 defer q.mu.Unlock() 214 215 now := time.Now() 216 var ready []*SchemaJob 217 218 for _, job := range q.jobs { 219 if job.ScheduledAt.Before(now) || job.ScheduledAt.Equal(now) { 220 ready = append(ready, job) 221 } 222 } 223 224 return ready 225} 226 227// BySite returns jobs for a specific site. 228func (q *JobQueue) BySite(siteKey string) []*SchemaJob { 229 q.mu.Lock() 230 defer q.mu.Unlock() 231 232 var result []*SchemaJob 233 for _, job := range q.pending { 234 if job.SiteKey == siteKey { 235 result = append(result, job) 236 } 237 } 238 239 return result 240} 241 242// ByType returns jobs of a specific type. 243func (q *JobQueue) ByType(jobType string) []*SchemaJob { 244 q.mu.Lock() 245 defer q.mu.Unlock() 246 247 var result []*SchemaJob 248 for _, job := range q.pending { 249 if job.JobType == jobType { 250 result = append(result, job) 251 } 252 } 253 254 return result 255} 256 257// Clear removes all jobs. 258func (q *JobQueue) Clear() { 259 q.mu.Lock() 260 defer q.mu.Unlock() 261 262 q.jobs = make(jobHeap, 0) 263 q.pending = make(map[string]*SchemaJob) 264 q.dedupMap = make(map[string]string) 265 heap.Init(&q.jobs) 266} 267 268// Save persists the queue to disk. 269func (q *JobQueue) Save() error { 270 q.mu.Lock() 271 defer q.mu.Unlock() 272 273 if q.filePath == "" { 274 return nil 275 } 276 277 jobs := make([]*SchemaJob, 0, len(q.pending)) 278 for _, job := range q.pending { 279 jobs = append(jobs, job) 280 } 281 282 data, err := json.MarshalIndent(jobs, "", " ") 283 if err != nil { 284 return err 285 } 286 287 return os.WriteFile(q.filePath, data, 0644) 288} 289 290func (q *JobQueue) load() error { 291 if q.filePath == "" { 292 return nil 293 } 294 295 data, err := os.ReadFile(q.filePath) 296 if err != nil { 297 if os.IsNotExist(err) { 298 return nil 299 } 300 return err 301 } 302 303 var jobs []*SchemaJob 304 if err := json.Unmarshal(data, &jobs); err != nil { 305 return err 306 } 307 308 for _, job := range jobs { 309 heap.Push(&q.jobs, job) 310 q.pending[job.JobID] = job 311 } 312 313 return nil 314} 315 316// --- Priority heap implementation --- 317 318type jobHeap []*SchemaJob 319 320func (h jobHeap) Len() int { return len(h) } 321 322func (h jobHeap) Less(i, j int) bool { 323 // Higher priority first 324 if h[i].Priority != h[j].Priority { 325 return h[i].Priority > h[j].Priority 326 } 327 // Earlier scheduled first 328 return h[i].ScheduledAt.Before(h[j].ScheduledAt) 329} 330 331func (h jobHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } 332 333func (h *jobHeap) Push(x any) { 334 *h = append(*h, x.(*SchemaJob)) 335} 336 337func (h *jobHeap) Pop() any { 338 old := *h 339 n := len(old) 340 job := old[n-1] 341 *h = old[0 : n-1] 342 return job 343} 344