reanimator/core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Finalize history admin tools and semantic UI navigation

Browse Source
This commit is contained in:
Michael Chus
2026-02-23 16:59:09 +03:00
parent 4e8554f5f0
commit 8aa8b26184
43 changed files with 5543 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
internal/history/worker.go
View File

@@ -17,33 +17,41 @@ func (s *Service) StartWorker(ctx context.Context) {
}
func (s *Service) workerLoop(ctx context.Context) {
ticker := time.NewTicker(750 * time.Millisecond)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
job, ok, err := s.claimNextJob(ctx)
if err != nil {
log.Printf("history worker claim failed err=%v", err)
continue
default:
}
job, ok, err := s.claimNextJob(ctx)
if err != nil {
log.Printf("history worker claim failed err=%v", err)
select {
case <-ctx.Done():
return
case <-time.After(300 * time.Millisecond):
}
if !ok {
continue
}
log.Printf("history worker claimed job_id=%s type=%s entity=%s:%s", job.ID, job.JobType, job.EntityType, job.EntityID)
started := time.Now()
result, procErr := s.processJob(ctx, job)
duration := time.Since(started)
if procErr != nil {
log.Printf("history worker job failed job_id=%s type=%s entity=%s:%s duration_ms=%d err=%v", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds(), procErr)
} else {
log.Printf("history worker job completed job_id=%s type=%s entity=%s:%s duration_ms=%d", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds())
}
if completeErr := s.completeJob(ctx, job.ID, result, procErr); completeErr != nil {
log.Printf("history worker complete failed job_id=%s err=%v", job.ID, completeErr)
continue
}
if !ok {
select {
case <-ctx.Done():
return
case <-time.After(150 * time.Millisecond):
}
continue
}
log.Printf("history worker claimed job_id=%s type=%s entity=%s:%s", job.ID, job.JobType, job.EntityType, job.EntityID)
started := time.Now()
result, procErr := s.processJob(ctx, job)
duration := time.Since(started)
if procErr != nil {
log.Printf("history worker job failed job_id=%s type=%s entity=%s:%s duration_ms=%d err=%v", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds(), procErr)
} else {
log.Printf("history worker job completed job_id=%s type=%s entity=%s:%s duration_ms=%d", job.ID, job.JobType, job.EntityType, job.EntityID, duration.Milliseconds())
}
if completeErr := s.completeJob(ctx, job.ID, result, procErr); completeErr != nil {
log.Printf("history worker complete failed job_id=%s err=%v", job.ID, completeErr)
}
}
}
@@ -118,6 +126,8 @@ func (s *Service) completeJob(ctx context.Context, jobID string, result map[stri
func (s *Service) processJob(ctx context.Context, job JobRecord) (map[string]any, error) {
switch job.JobType {
case "recompute":
return s.processRecomputeJob(ctx, job)
case "delete_event_recompute":
return s.processDeleteEventRecompute(ctx, job)
case "rollback":