feat: add job maintenance dashboard and automated stale job expiration system

internal/worker/pool.go

@@ -32,7 +32,6 @@ type WorkerPool struct {
 }
 
 func NewWorkerPool(database *db.DB, analyzer analyzer.LLMAnalyzer, numWorkers int, filterCfg *config.FilterConfig) *WorkerPool {
-	// Initialize filters using the provided config
 	filterPipe := filter.NewPipeline(
 		filter.NewHardKeywordFilter(filterCfg.HardKeywords),
 		filter.NewScoringFilter(filterCfg.ScoringWeights),
@@ -54,33 +53,37 @@ func (p *WorkerPool) Toggle() bool {
 	if p.IsRunning() {
 		p.Stop()
 		return false
+	} else {
+		p.Start()
+		return true
 	}
-	p.Start()
-	return true
 }
 
 func (p *WorkerPool) Start() {
-	if p.isRunning.Load() {
-		return
+	if p.isRunning.Swap(true) {
+		return // Already running
 	}
+
 	p.ctx, p.cancel = context.WithCancel(context.Background())
-	p.isRunning.Store(true)
+	logger.Log.Infow("Started workers", "component", "WorkerPool", "count", p.workerCount)
 
 	for i := 0; i < p.workerCount; i++ {
 		p.wg.Add(1)
 		go p.workerLoop(i)
 	}
-	logger.Log.Infow("Started workers", "component", "WorkerPool", "count", p.workerCount)
 }
 
 func (p *WorkerPool) Stop() {
-	if !p.isRunning.Load() {
-		return
+	if !p.isRunning.Swap(false) {
+		return // Already stopped
+	}
+
+	logger.Log.Infow("Stopping worker pool...", "component", "WorkerPool")
+	if p.cancel != nil {
+		p.cancel()
 	}
-	p.cancel()
-	p.isRunning.Store(false)
 	p.wg.Wait()
-	logger.Log.Infow("All workers stopped", "component", "WorkerPool")
+	logger.Log.Infow("Worker pool stopped", "component", "WorkerPool")
 }
 
 func (p *WorkerPool) workerLoop(workerID int) {
@@ -95,7 +98,7 @@ func (p *WorkerPool) workerLoop(workerID int) {
 		default:
 			task, err := p.db.ClaimNextTask()
 			if err != nil {
-				logger.Log.Errorw("Error claiming task", "component", "Worker", "worker_id", workerID, "error", err)
+				logger.Log.Errorw("Failed to claim task", "component", "Worker", "worker_id", workerID, "error", err)
 				time.Sleep(1 * time.Second)
 				continue
 			}
@@ -119,6 +122,8 @@ func (p *WorkerPool) workerLoop(workerID int) {
 				processErr = p.handleProcess(p.ctx, task)
 			case "ANALYZE":
 				processErr = p.handleAnalyze(p.ctx, task)
+			case "EXPIRE_JOB":
+				processErr = p.handleExpireJob(p.ctx, task)
 			default:
 				processErr = fmt.Errorf("unknown task type: %s", task.TaskType)
 			}
@@ -159,6 +164,11 @@ type AnalyzePayload struct {
 	RawJobID int64 `json:"raw_job_id"`
 }
 
+type ExpireJobPayload struct {
+	RawJobID int64  `json:"raw_job_id"`
+	Reason   string `json:"reason"`
+}
+
 // ---------------------------------------------------------
 // Task Handlers
 // ---------------------------------------------------------
@@ -179,6 +189,8 @@ func (p *WorkerPool) handleDiscover(ctx context.Context, task *db.Task) error {
 		// Check deduplication (already fetched/inserted jobs)
 		exists, err := p.db.CanonicalIDExists(disc.CanonicalID)
 		if err == nil && exists {
+			// Update last seen timestamp to confirm job is still active on target board
+			_ = p.db.TouchJobLastSeen(disc.CanonicalID)
 			continue // Skip existing job
 		}
 
@@ -289,6 +301,15 @@ func (p *WorkerPool) handleAnalyze(ctx context.Context, task *db.Task) error {
 	return err
 }
 
+func (p *WorkerPool) handleExpireJob(ctx context.Context, task *db.Task) error {
+	var payload ExpireJobPayload
+	if err := json.Unmarshal([]byte(task.Payload), &payload); err != nil {
+		return err
+	}
+	logger.Log.Infow("Expiring job", "component", "Worker", "raw_job_id", payload.RawJobID, "reason", payload.Reason)
+	return p.db.ExpireJob(payload.RawJobID, payload.Reason)
+}
+
 func (p *WorkerPool) getCrawlerForPlatform(platform string) crawler.SiteCrawler {
 	plat := strings.ToLower(platform)
 	if strings.Contains(plat, "google") {