# Workflow Layer Agent Instructions

## Overview

Workflows handle **async operations** using the fastflow engine. They execute as DAGs (Directed Acyclic Graphs) with retry and recovery support.

## Directory Structure

```
workflow/
├── interface.go           # Workflow interface
├── constants.go           # DAG names, action names, keys
├── impl/                  # Fastflow implementation
├── dag/                   # DAG definitions
│   ├── register.go        # Central registration
│   ├── backup/
│   │   └── snapshot/
│   │       ├── create.go  # Create backup DAG
│   │       └── delete.go  # Delete backup DAG
│   └── recovery_group/
└── action/                # Action implementations
    ├── base_action.go     # Wrapper and retry logic
    ├── retryer.go         # Custom retry
    └── backup/
        ├── base/
        │   └── base.go    # Shared backup dependencies
        └── snapshot/
            └── create/
                ├── create_infra.go
                └── sync_status.go
```

## DAG Definition

**Create in `dag/<domain>/<feature>/<action>.go`:**
```go
func GetCreateBackupDag() (*entity.Dag, []run.Action) {
    base := backup_base.NewBaseAction(...)
    
    actions := []run.Action{
        create.NewValidateAction(base),
        create.NewCreateInfraAction(base),
        create.NewSyncStatusAction(base),
    }
    
    dag := &entity.Dag{
        BaseInfo: entity.BaseInfo{ID: workflow.DagCreateBackup},
        Name:     workflow.DagCreateBackup,
        Vars: entity.DagVars{
            workflow.KeyBackupID: {},  // Required input variable
        },
        Status: entity.DagStatusNormal,
        Tasks: []entity.Task{
            {
                ID:          workflow.ActionValidate,
                ActionName:  workflow.ActionValidate,
                TimeoutSecs: 60,
            },
            {
                ID:          workflow.ActionCreateInfra,
                ActionName:  workflow.ActionCreateInfra,
                DependOn:    []string{workflow.ActionValidate},  // Dependencies
                TimeoutSecs: 300,
            },
            {
                ID:          workflow.ActionSyncStatus,
                ActionName:  workflow.ActionSyncStatus,
                DependOn:    []string{workflow.ActionCreateInfra},
                TimeoutSecs: 600,
            },
        },
    }
    return dag, actions
}
```

**Task parameters with variables:**
```go
{
    ID:          workflow.ActionSomeAction,
    ActionName:  workflow.ActionSomeAction,
    Params:      map[string]string{
        "backup_id": fmt.Sprintf("{{%s}}", workflow.KeyBackupID),
    },
}
```

## Action Implementation

**Create in `action/<domain>/<feature>/<action>.go`:**
```go
type CreateInfraAction struct {
    base.BaseAction
}

func NewCreateInfraAction(base base.BaseAction) *action.Wrapper {
    a := &CreateInfraAction{BaseAction: base}
    return action.MustNewActionWrapper(
        workflow.ActionCreateInfra,
        a,
        action.SetRunRetryFun(action.InvokeRunActionWithRetry),
    )
}

// Params struct for type-safe parameter access
type CreateInfraParams struct {
    BackupID uint64 `json:"backup_id"`
}

func (a *CreateInfraAction) Params() interface{} {
    return &CreateInfraParams{}
}

// Run executes the action (short operations)
func (a *CreateInfraAction) Run(ctx run.ExecuteContext, params interface{}) error {
    p := params.(*CreateInfraParams)
    logger := action.GetLogger(ctx.Context()).Named(a.Name())
    
    // Setup tracing
    ctx, span := action.Tracer.Start(ctx.Context(), action.GenActionSpanName(a.Name(), "Run"))
    defer span.End()
    
    // Action logic
    logger.Info("creating infra", zap.Uint64("backupID", p.BackupID))
    
    return nil
}
```

## Run vs RunAfter

```go
// Run: Quick setup, called once
func (a *Action) Run(ctx run.ExecuteContext, params interface{}) error {
    // Fast operations only
    return nil
}

// RunAfter: Long-polling, auto-retried if worker restarts
// MUST be idempotent!
func (a *Action) RunAfter(ctx run.ExecuteContext, params interface{}) error {
    // Poll until complete
    return a.pollUntilDone(ctx)
}
```

## Retry Patterns

**Built-in retry functions:**
| Function | Behavior | Use Case |
|----------|----------|----------|
| `InvokeRunActionWithRetry` | 3 attempts, 1s delay | Quick ops, low failure rate |
| `InvokeRunActionWithExponentialRetry` | 8 attempts, 500ms-10s backoff | **Default for Run()** |
| `InvokeRunActionWithInfinitelyRetry` | Infinite, 15s delay | **Idempotent ops (delete, cleanup)** |
| `InvokeRunActionWithInfinitelyExponentialRetry` | Infinite, exponential | **Default for RunAfter()** |

**How to choose retry strategy:**
- **Create/Update operations**: Use `InvokeRunActionWithExponentialRetry` (fail fast if something is wrong)
- **Delete/Cleanup operations**: Use `InvokeRunActionWithInfinitelyRetry` (idempotent, must eventually succeed)
- **Long-polling in RunAfter**: Use `InvokeRunActionWithInfinitelyExponentialRetry` (default)

**Set retry in constructor:**
```go
return action.MustNewActionWrapper(
    name,
    actionImpl,
    action.SetRunRetryFun(action.InvokeRunActionWithInfinitelyRetry),      // For Run()
    action.SetRunAfterRetryFun(action.InvokeRunActionWithInfinitelyRetry), // For RunAfter()
)
```

**Stop retry signal:**
```go
if shouldStop {
    return action.StopRetry  // Stops infinite retry immediately
}
```

**Custom retry for specific errors:**
```go
type MyAction struct {
    base.BaseAction
    errorWaiter action.Retryer
}

func NewMyAction(base base.BaseAction) *MyAction {
    return &MyAction{
        BaseAction:  base,
        errorWaiter: action.NewRetryer([]error{ErrNotReady}),  // Only retry this error
    }
}

func (a *MyAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
    return a.errorWaiter.Execute(ctx.Context(), logger, func() error {
        return a.checkStatus()
    })
}
```

## ShareData (Pass Data Between Actions)

```go
// Set data
ctx.ShareData().Set(workflow.KeyBackupID, strconv.FormatUint(id, 10))

// Get data
idStr, ok := ctx.ShareData().Get(workflow.KeyBackupID)

// Helper functions
id, err := action.GetUint64FromShareData(ctx.ShareData(), workflow.KeyBackupID)
err := action.SetSliceUint64IntoShareData(ctx.ShareData(), key, []uint64{1, 2, 3})
```

## Base Action Pattern

**Shared dependencies in `action/<domain>/base/base.go`:**
```go
type BaseAction struct {
    TransProvider  db.TransProvider
    BackupRepo     backup_repo.BackupRepo
    CoreCli        grpcli.CoreGrpcClient
    // ... shared deps
}

func (b *BaseAction) LoadBackup(ctx context.Context, id uint64) (*entity.Backup, error) {
    return b.BackupRepo.Load(ctx, id)
}

func (b *BaseAction) EnhanceLogger(logger *zap.Logger, backup *entity.Backup) *zap.Logger {
    return logger.With(
        zap.Uint64("backupID", backup.ID.Uint64()),
        zap.Uint64("clusterID", backup.ClusterID.Uint64()),
    )
}
```

## DAG Registration

**In `dag/register.go`:**
```go
func RegisterDags() error {
    wf := container.Global().WorkflowModel
    
    if err := wf.RegisterDag(dag_backup.GetCreateBackupDag()); err != nil {
        return err
    }
    if err := wf.RegisterDag(dag_backup.GetDeleteBackupDag()); err != nil {
        return err
    }
    // ... register all DAGs
    return nil
}
```

## Running Workflows

```go
// From model layer
dagIns, err := m.workflow.RunDag(workflow.DagCreateBackup, map[string]string{
    workflow.KeyBackupID: strconv.FormatUint(backupID, 10),
})

// With tags
dagIns, err := m.workflow.RunDagWithTags(dagID, vars, map[string]string{
    "cluster_id": strconv.FormatUint(clusterID, 10),
})

// Retry/Cancel
err := m.workflow.RetryDagIns(dagInsID)
err := m.workflow.CancelDagIns(dagInsID)
```

## Idempotency Requirements

Actions (especially RunAfter) **MUST be idempotent**:

```go
func (a *SyncStatusAction) RunAfter(ctx run.ExecuteContext, params interface{}) error {
    backup, _ := a.LoadBackup(ctx.Context(), p.BackupID)
    
    // Already complete? Return success
    if backup.InFinalStatus() {
        logger.Info("backup already finished")
        return nil
    }
    
    // Continue polling...
    return a.pollStatus(ctx)
}
```

**Idempotency checklist:**
1. Check current state before modifying
2. "Already done" = success, not error
3. Use unique IDs to prevent duplicates
4. Prefer upsert/conditional updates