# Worker Agent Instructions

## Overview

Workers are **scheduled background jobs** using `gocron`. They handle periodic tasks like cleanup, sync, and maintenance.

## Directory Structure

```
worker/
├── server.go              # Central registration
├── common/
│   └── metrics.go         # Wrapper with metrics/recovery
├── backup/
│   ├── jobs.go            # Scheduler definition
│   └── backup.go          # Worker functions
├── changefeed/
│   ├── job.go
│   └── changefeed.go
└── <domain>/
    ├── job.go             # New*WorkerScheduler()
    └── <domain>.go        # Worker functions
```

## Scheduler Pattern

**Define in `<domain>/jobs.go`:**
```go
package backup

import (
    "context"
    "time"
    "github.com/go-co-op/gocron"
    "github.com/tidbcloud/dataflow-service/app/worker/common"
)

const (
    scheduleAutoBackupIntervalSeconds = 60
)

func NewWorkerScheduler(ctx context.Context, loc *time.Location) (*gocron.Scheduler, error) {
    s := gocron.NewScheduler(loc)
    s.SingletonModeAll()  // Prevent overlapping executions

    // Interval-based scheduling
    if _, err := s.Every(scheduleAutoBackupIntervalSeconds).Seconds().
        Do(common.WorkerFunWrapper("ScheduleAutoBackup", ScheduleAutoBackup)); err != nil {
        return nil, err
    }
    
    // Cron-based scheduling (run at 3 AM daily)
    if _, err := s.Cron("0 3 * * *").
        Do(common.WorkerFunWrapper("CleanupOldBackups", CleanupOldBackups)); err != nil {
        return nil, err
    }
    
    return s, nil
}
```

## Worker Function Pattern

**Define in `<domain>/<domain>.go`:**
```go
package backup

import (
    "context"
    "github.com/tidbcloud/dataflow-service/app/container"
    "github.com/tidbcloud/dataflow-service/pkg/log"
    "go.uber.org/zap"
)

func ScheduleAutoBackup(ctx context.Context) error {
    logger := log.GetLogger(ctx)
    logger.Info("=== ScheduleAutoBackup worker started ===")
    
    // Delegate to model layer
    if err := container.Global().AutoBackupScheduler.ScheduleAutoBackup(ctx); err != nil {
        logger.Warn("schedule auto backup error", zap.Error(err))
    }
    
    logger.Info("=== ScheduleAutoBackup worker completed ===")
    return nil  // Always return nil - don't break scheduler
}

func CleanupOldBackups(ctx context.Context) error {
    logger := log.GetLogger(ctx)
    logger.Info("=== CleanupOldBackups worker started ===")
    
    if err := container.Global().BackupCleanupModel.Cleanup(ctx); err != nil {
        logger.Warn("cleanup old backups error", zap.Error(err))
    }
    
    logger.Info("=== CleanupOldBackups worker completed ===")
    return nil
}
```

## Worker Registration

**In `server.go`:**
```go
func MustNewWorker(ctx context.Context, loc *time.Location) *Worker {
    backupScheduler, err := backup.NewWorkerScheduler(ctx, loc)
    if err != nil {
        panic(err)
    }
    
    changefeedScheduler, err := changefeed.NewWorkerScheduler(ctx, loc)
    if err != nil {
        panic(err)
    }
    
    w := &Worker{}
    w.schedulers = append(w.schedulers,
        backupScheduler,
        changefeedScheduler,
        // ... add new schedulers
    )
    return w
}

func (w *Worker) StartWorker() {
    for _, scheduler := range w.schedulers {
        scheduler.StartAsync()
    }
}
```

## WorkerFunWrapper

The wrapper adds metrics, recovery, and tracing:

```go
// common/metrics.go
func WorkerFunWrapper(workerName string, f Handler) WorkerFun {
    handler := WithRecover(workerName, WithMetrics(workerName, f))
    return func() error {
        ctx := context.Background()
        ctx, span := Tracer.Start(ctx, genWorkerSpanName(workerName))
        defer span.End()
        return handler(ctx)
    }
}
```

**Automatically provides:**
- Panic recovery (logs stack trace)
- Metrics (duration, success/error status)
- Distributed tracing (OpenTelemetry span)

## Key Rules

1. **Always return nil** - Log errors, don't fail the scheduler
2. **Delegate to models** - Workers are thin, business logic in models
3. **Use container.Global()** - Access dependencies via DI
4. **Log start/end** - Help with debugging periodic jobs
5. **Idempotency** - Workers may run multiple times, handle gracefully
6. **SingletonModeAll()** - Prevent job overlap

## Scheduling Options

```go
// Interval-based
s.Every(60).Seconds().Do(fn)
s.Every(5).Minutes().Do(fn)
s.Every(1).Hour().Do(fn)
s.Every(1).Day().At("03:00").Do(fn)

// Cron expression
s.Cron("*/5 * * * *").Do(fn)   // Every 5 minutes
s.Cron("0 3 * * *").Do(fn)     // Daily at 3 AM
s.Cron("0 0 * * 0").Do(fn)     // Weekly on Sunday
```

## Adding New Worker

1. Create `worker/<domain>/job.go` with `NewWorkerScheduler()`
2. Create `worker/<domain>/<domain>.go` with worker functions
3. Add scheduler to `worker/server.go`
4. Define constants in `constants.go` if needed