base/plugins/tasks/durable.go

package tasks

import (
	"context"
	"fmt"
	"os"
	"regexp"
	"sync"
	"time"

	"github.com/hanzoai/base/tools/logger"
	"github.com/hanzoai/tasks/pkg/sdk/client"
	"github.com/hanzoai/tasks/pkg/sdk/temporal"
)

// validIDPattern restricts space/org IDs to safe characters, preventing
// injection into Temporal visibility queries.
var validIDPattern = regexp.MustCompile(`^[a-zA-Z0-9_.-]{1,128}$`)

// DurableConfig holds connection settings for durable task execution.
// When enabled, tasks are submitted as durable workflows for crash-safe execution
// against cloud's embedded, identity-gated Hanzo Tasks engine.
type DurableConfig struct {
	// Enabled activates durable execution. Default false (SQLite-only mode).
	Enabled bool `json:"enabled" yaml:"enabled"`

	// Address is the tasks ZAP frontend address. Default "cloud.hanzo.svc:9999"
	// (cloud's Embedded.ServeGated listener; the standalone tasksd is retired).
	Address string `json:"address" yaml:"address"`

	// Namespace is the Temporal namespace. For per-org, use org ID.
	// Default "default".
	Namespace string `json:"namespace" yaml:"namespace"`

	// DefaultQueue is the task queue name. Default "default".
	DefaultQueue string `json:"default_queue" yaml:"default_queue"`

	// RunWorker starts an embedded Temporal worker in this process.
	// If false, tasks are submitted but executed by external workers.
	// Default true when enabled.
	RunWorker bool `json:"run_worker" yaml:"run_worker"`
}

// DefaultDurableConfig returns production defaults with env var overrides.
//
// Env vars:
//   - TASKS_ENABLED (or HANZO_TASKS_ENABLED): "true" to enable
//   - TASKS_ADDRESS (or HANZO_TASKS_ADDRESS): Temporal frontend address
//   - TASKS_NAMESPACE (or HANZO_TASKS_NAMESPACE): Temporal namespace (org ID for per-org)
//   - TASKS_QUEUE: default task queue name
//   - TASKS_WORKER: "false" to disable embedded worker
func DefaultDurableConfig() DurableConfig {
	cfg := DurableConfig{
		Enabled:      false,
		Address:      "cloud.hanzo.svc:9999",
		Namespace:    "default",
		DefaultQueue: "default",
		RunWorker:    true,
	}

	for _, key := range []string{"TASKS_ENABLED", "HANZO_TASKS_ENABLED"} {
		if v := os.Getenv(key); v == "true" || v == "1" {
			cfg.Enabled = true
			break
		}
	}
	for _, key := range []string{"TASKS_ADDRESS", "HANZO_TASKS_ADDRESS"} {
		if v := os.Getenv(key); v != "" {
			cfg.Address = v
			break
		}
	}
	for _, key := range []string{"TASKS_NAMESPACE", "HANZO_TASKS_NAMESPACE"} {
		if v := os.Getenv(key); v != "" {
			cfg.Namespace = v
			break
		}
	}
	if v := os.Getenv("TASKS_QUEUE"); v != "" {
		cfg.DefaultQueue = v
	}
	if v := os.Getenv("TASKS_WORKER"); v == "false" || v == "0" {
		cfg.RunWorker = false
	}

	return cfg
}

// DurableStore implements durable task execution via Temporal.
// Supports per-org org isolation via per-namespace client connections.
type DurableStore struct {
	Client    client.Client // default namespace client
	addr      string
	namespace string
	connected bool

	// token authenticates every dial to cloud's identity-gated tasks engine.
	// nil dials ungated (dev/loopback); see iamTokenSource.
	token client.TokenSource

	// orgClients caches per-org client connections. Key is org ID (= namespace).
	mu         sync.RWMutex
	orgClients map[string]client.Client
}

// NewDurableStore connects to the tasks engine and registers its default
// namespace. cloud's gated engine org-prefixes the store by the token owner,
// so a namespace is invisible until registered — ExecuteWorkflow 400s otherwise.
func NewDurableStore(addr, namespace string) (*DurableStore, error) {
	token := iamTokenSource()
	c, err := client.Dial(client.Options{
		Address:   addr,
		Namespace: namespace,
		Token:     token,
	})
	if err != nil {
		return nil, fmt.Errorf("tasks: failed to connect to %s: %w", addr, err)
	}
	if err := registerNamespace(c, namespace); err != nil {
		c.Close()
		return nil, err
	}
	return &DurableStore{
		Client:     c,
		addr:       addr,
		namespace:  namespace,
		token:      token,
		connected:  true,
		orgClients: make(map[string]client.Client),
	}, nil
}

// registerNamespace ensures name exists on the engine before any ExecuteWorkflow
// targets it. The engine upserts, so this is idempotent across restarts.
func registerNamespace(c client.Client, name string) error {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	if err := c.RegisterNamespace(ctx, &client.RegisterNamespaceRequest{Name: name}); err != nil {
		return fmt.Errorf("tasks: register namespace %q: %w", name, err)
	}
	return nil
}

// ClientForOrg returns a Temporal client scoped to the given org namespace.
// Connections are cached and protected by a mutex. Falls back to the default
// client if org is empty or matches the default namespace.
func (ds *DurableStore) ClientForOrg(org string) (client.Client, error) {
	if org == "" || org == ds.namespace {
		return ds.Client, nil
	}
	if !validIDPattern.MatchString(org) {
		return nil, fmt.Errorf("tasks: invalid org ID: %q", org)
	}

	// Fast path: read lock.
	ds.mu.RLock()
	if c, ok := ds.orgClients[org]; ok {
		ds.mu.RUnlock()
		return c, nil
	}
	ds.mu.RUnlock()

	// Slow path: write lock, double-check, dial.
	ds.mu.Lock()
	defer ds.mu.Unlock()

	if c, ok := ds.orgClients[org]; ok {
		return c, nil
	}

	c, err := client.Dial(client.Options{
		Address:   ds.addr,
		Namespace: org,
		Token:     ds.token,
	})
	if err != nil {
		return nil, fmt.Errorf("tasks: failed to connect to namespace %s: %w", org, err)
	}
	if err := registerNamespace(c, org); err != nil {
		c.Close()
		return nil, err
	}
	ds.orgClients[org] = c
	return c, nil
}

// Close shuts down all client connections.
func (ds *DurableStore) Close() {
	ds.mu.Lock()
	for _, c := range ds.orgClients {
		c.Close()
	}
	ds.orgClients = nil
	ds.mu.Unlock()

	if ds.Client != nil {
		ds.Client.Close()
		ds.connected = false
	}
}

// IsConnected reports whether the durable store has an active connection.
func (ds *DurableStore) IsConnected() bool {
	return ds != nil && ds.connected
}

// SubmitTask starts a durable workflow execution for a task.
// Uses task.OrgID as the Temporal namespace for per-org isolation.
// Falls back to default namespace if OrgID is empty.
func (ds *DurableStore) SubmitTask(ctx context.Context, task *Task) error {
	c, err := ds.ClientForOrg(task.OrgID)
	if err != nil {
		return err
	}

	queue := task.SpaceID
	if queue == "" {
		queue = "default"
	}

	opts := client.StartWorkflowOptions{
		ID:        task.ID,
		TaskQueue: queue,
	}
	if task.Timeout > 0 {
		opts.WorkflowExecutionTimeout = task.Timeout
	}
	if task.MaxRetries > 0 {
		opts.RetryPolicy = &temporal.RetryPolicy{
			MaximumAttempts:    int32(task.MaxRetries),
			InitialInterval:    2 * time.Second,
			MaximumInterval:    time.Minute,
			BackoffCoefficient: 2.0,
		}
	}

	_, err = c.ExecuteWorkflow(ctx, opts, AgentTaskWorkflow, task)
	if err != nil {
		return fmt.Errorf("tasks: failed to submit workflow: %w", err)
	}
	task.State = TaskRunning
	return nil
}

// SubmitWorkflow starts a pipeline or fan-out workflow.
func (ds *DurableStore) SubmitWorkflow(ctx context.Context, wf *Workflow, tasks []*Task, parallel bool) error {
	c, err := ds.ClientForOrg(wf.OrgID)
	if err != nil {
		return err
	}

	queue := wf.SpaceID
	if queue == "" {
		queue = "default"
	}

	opts := client.StartWorkflowOptions{
		ID:        wf.ID,
		TaskQueue: queue,
	}

	var wfFunc interface{}
	if parallel {
		wfFunc = FanOutWorkflow
	} else {
		wfFunc = PipelineWorkflow
	}

	_, err = c.ExecuteWorkflow(ctx, opts, wfFunc, wf, tasks)
	if err != nil {
		return fmt.Errorf("tasks: failed to submit workflow: %w", err)
	}
	wf.State = TaskRunning
	return nil
}

// GetTaskStatus queries a running workflow for its current state.
// org selects the Temporal namespace. Empty string uses the default.
func (ds *DurableStore) GetTaskStatus(ctx context.Context, taskID string, org string) (TaskState, string, error) {
	c, err := ds.clientFor(org)
	if err != nil {
		return TaskPending, "", err
	}
	info, err := c.DescribeWorkflow(ctx, taskID, "")
	if err != nil {
		return TaskPending, "", err
	}

	switch info.Status {
	case client.WorkflowStatusRunning:
		return TaskRunning, "", nil
	case client.WorkflowStatusCompleted:
		return TaskCompleted, "", nil
	case client.WorkflowStatusFailed:
		return TaskFailed, "workflow failed", nil
	case client.WorkflowStatusCanceled, client.WorkflowStatusTerminated:
		return TaskCancelled, "", nil
	case client.WorkflowStatusTimedOut:
		return TaskFailed, "timed out", nil
	default:
		return TaskPending, "", nil
	}
}

// CancelTask cancels a running workflow.
func (ds *DurableStore) CancelTask(ctx context.Context, taskID string, org string) error {
	c, err := ds.clientFor(org)
	if err != nil {
		return err
	}
	return c.CancelWorkflow(ctx, taskID, "")
}

// SignalTask sends a signal to a running workflow.
func (ds *DurableStore) SignalTask(ctx context.Context, taskID, signalName string, data interface{}, org string) error {
	c, err := ds.clientFor(org)
	if err != nil {
		return err
	}
	return c.SignalWorkflow(ctx, taskID, "", signalName, data)
}

// clientFor returns the org's client, or the shared one when unscoped.
//
// Was clientForOrgVar — a name whose suffix described the CALLING CONVENTION of
// its argument rather than anything about the client it returns. With the
// variadic gone there is nothing left for "Var" to mean.
func (ds *DurableStore) clientFor(org string) (client.Client, error) {
	if org == "" {
		return ds.Client, nil
	}
	return ds.ClientForOrg(org)
}

// ListTasks returns tasks in a space by querying workflow visibility.
func (ds *DurableStore) ListTasks(ctx context.Context, spaceID string, org string) ([]*Task, error) {
	if spaceID != "" && !validIDPattern.MatchString(spaceID) {
		return nil, fmt.Errorf("tasks: invalid space ID: %q", spaceID)
	}
	c, err := ds.clientFor(org)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(`TaskQueue = "%s"`, spaceID)
	resp, err := c.ListWorkflows(ctx, query, 0, nil)
	if err != nil {
		return nil, fmt.Errorf("tasks: list failed: %w", err)
	}

	var result []*Task
	for _, exec := range resp.Executions {
		task := &Task{
			ID:      exec.WorkflowID,
			SpaceID: spaceID,
		}
		switch exec.Status {
		case client.WorkflowStatusRunning:
			task.State = TaskRunning
		case client.WorkflowStatusCompleted:
			task.State = TaskCompleted
		case client.WorkflowStatusFailed:
			task.State = TaskFailed
		case client.WorkflowStatusCanceled, client.WorkflowStatusTerminated:
			task.State = TaskCancelled
		default:
			task.State = TaskPending
		}
		if !exec.StartTime.IsZero() {
			t := exec.StartTime
			task.StartedAt = &t
		}
		result = append(result, task)
	}
	return result, nil
}

// GetNextTask finds the next pending task in a space and claims it for the agent.
func (ds *DurableStore) GetNextTask(ctx context.Context, spaceID, agentID string, org string) (*Task, error) {
	tasks, err := ds.ListTasks(ctx, spaceID, org)
	if err != nil {
		return nil, err
	}
	for _, t := range tasks {
		if t.State == TaskRunning || t.State == TaskPending {
			// Same org the task was listed under. It was silently unscoped
			// while the argument was variadic — invisible, because omitting a
			// variadic looks like nothing at all.
			_ = ds.SignalTask(ctx, t.ID, "claim", map[string]string{"agent_id": agentID}, org)
			return t, nil
		}
	}
	return nil, nil
}

// ListWorkflows returns workflows in a space by querying visibility.
func (ds *DurableStore) ListWorkflows(ctx context.Context, spaceID string, org string) ([]*Workflow, error) {
	if spaceID != "" && !validIDPattern.MatchString(spaceID) {
		return nil, fmt.Errorf("tasks: invalid space ID: %q", spaceID)
	}
	c, err := ds.clientFor(org)
	if err != nil {
		return nil, err
	}
	query := fmt.Sprintf(`TaskQueue = "%s" AND WorkflowType IN ("PipelineWorkflow", "FanOutWorkflow")`, spaceID)
	resp, err := c.ListWorkflows(ctx, query, 0, nil)
	if err != nil {
		return nil, fmt.Errorf("tasks: list workflows failed: %w", err)
	}

	var result []*Workflow
	for _, exec := range resp.Executions {
		wf := &Workflow{
			ID:      exec.WorkflowID,
			SpaceID: spaceID,
			Name:    exec.WorkflowType,
		}
		switch exec.Status {
		case client.WorkflowStatusRunning:
			wf.State = TaskRunning
		case client.WorkflowStatusCompleted:
			wf.State = TaskCompleted
		case client.WorkflowStatusFailed:
			wf.State = TaskFailed
		default:
			wf.State = TaskPending
		}
		result = append(result, wf)
	}
	return result, nil
}

// logDurableError logs a durable store error without failing the operation.
func logDurableError(log logger.Logger, op string, taskID string, err error) {
	if log != nil {
		log.Warn("tasks: durable "+op+" failed, SQLite state is authoritative",
			"task_id", taskID,
			"error", err.Error(),
		)
	}
}