base/tools/tasks/client.go

// Package tasks provides a durable task client for Hanzo Tasks.
//
// Drop-in replacement for Base's Cron():
//
//	// Before (cron):
//	e.App.Cron().Add("settlement", "*/30 * * * * *", func() { ... })
//
//	// After (tasks — duration):
//	app.Tasks().Add("settlement", "30s", fn)
//
//	// After (tasks — cron expression):
//	app.Tasks().Add("daily-cleanup", "0 3 * * *", fn)
//
// Accepts both Go duration strings ("30s", "5m", "1h") and cron expressions
// ("0 3 * * *", "0 0 * * 1", "0 0 5 1,4,7,10 *"). Detection is automatic:
// if time.ParseDuration succeeds, it's a duration; otherwise it's cron.
//
// If TASKS_URL is set, schedules run as durable Hanzo Tasks workflows
// (retries, dead letter, audit trail). If not, runs locally via goroutine
// timer (dev mode, same behavior as cron but no persistence).
//
// If zapAddr is set, ZAP binary transport is preferred over HTTP for
// submitting tasks (lower latency, same semantics). HTTP is fallback.
package tasks

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	luxlog "github.com/luxfi/log"
	"github.com/luxfi/zap"
)

// taskQueue is the Hanzo Tasks queue every workflow this client submits
// is dispatched on.
const taskQueue = "base"

// namespacePath is where the Hanzo Tasks HTTP API answers for the namespace
// this client schedules into: /v1/tasks/namespaces/{ns}.
const namespacePath = "/v1/tasks/namespaces/default"

// ZAP opcodes for task submission.
const (
	OpcodeTaskSubmit   uint16 = 0x0050 // one-shot task
	OpcodeTaskSchedule uint16 = 0x0051 // recurring schedule
)

// ZAP object field offsets.
const (
	fieldTaskType = 0  // Text: task/workflow type
	fieldPayload  = 8  // Bytes: JSON payload
	fieldInterval = 16 // Text: duration string (schedules only)
	fieldName     = 0  // Text: schedule name (schedules)
	respStatus    = 0  // Uint32: status code
	respBody      = 4  // Bytes: response JSON
)

// Handler processes a one-shot task (webhook delivery, settlement, etc.)
type Handler func(taskType string, payload map[string]any)

// Schedule is a read-only view of a registered schedule.
// Returned by Client.Schedules() for the admin UI and introspection.
type Schedule struct {
	Name       string `json:"id"`
	Expression string `json:"expression"`
}

// scheduleEntry is the internal state tracked per Add().
type scheduleEntry struct {
	expression string             // original "30s" or "*/5 * * * *"
	interval   time.Duration      // resolved tick interval (local mode)
	fn         func()             // user-supplied callback (may be nil for remote-only)
	cancel     context.CancelFunc // local ticker canceller (nil when not ticking)
	remote     bool               // true when ticking happens server-side
}

// Client manages both one-shot tasks and recurring schedules.
type Client struct {
	tasksURL   string
	zapAddr    string
	httpClient *http.Client
	handler    Handler
	logger     luxlog.Logger
	mu         sync.RWMutex
	schedules  map[string]*scheduleEntry

	zapOnce sync.Once
	zapNode *zap.Node
	zapPeer string
	zapErr  error
}

// New creates a Client.
// If both tasksURL and zapAddr are empty, everything runs locally.
// If zapAddr is set, ZAP transport is preferred. HTTP is fallback.
func New(tasksURL, zapAddr string, handler Handler) *Client {
	return &Client{
		tasksURL: tasksURL,
		zapAddr:  zapAddr,
		httpClient: &http.Client{
			Timeout: 10 * time.Second,
		},
		handler:   handler,
		logger:    luxlog.New("component", "tasks"),
		schedules: make(map[string]*scheduleEntry),
	}
}

// Add registers a recurring task. schedule is either a Go duration string
// ("30s", "5m", "1h") or a cron expression ("0 3 * * *", "0 0 * * 1").
// Detection is automatic via time.ParseDuration.
//
//	app.Tasks().Add("settlement", "30s", fn)            // every 30 seconds
//	app.Tasks().Add("daily-cleanup", "0 3 * * *", fn)   // daily at 3am UTC
//	app.Tasks().Add("weekly-report", "0 0 * * 1", fn)   // mondays at midnight
//
// Re-adding with the same name replaces the previous schedule.
func (c *Client) Add(name, schedule string, fn func()) error {
	dur, durErr := time.ParseDuration(schedule)
	isCron := durErr != nil

	if isCron {
		dur = approximateCron(schedule)
	}

	// Cancel any previous ticker for this name before registering the new one.
	c.mu.Lock()
	if prev, ok := c.schedules[name]; ok && prev.cancel != nil {
		prev.cancel()
		prev.cancel = nil
	}
	c.mu.Unlock()

	if c.zapAddr != "" || c.tasksURL != "" {
		scheduled := false
		if c.zapAddr != "" {
			if err := c.scheduleZAP(name, dur); err == nil {
				scheduled = true
			} else {
				c.logger.Warn("tasks: ZAP schedule failed", "name", name, "error", err)
			}
		}
		if !scheduled && c.tasksURL != "" {
			var err error
			if isCron {
				err = c.createCronSchedule(name, schedule)
			} else {
				err = c.createSchedule(name, dur)
			}
			if err != nil {
				c.logger.Warn("tasks: HTTP schedule failed, falling back to local",
					"name", name, "error", err)
				c.addLocal(name, schedule, dur, fn)
				return nil
			}
			// Remote schedule succeeded: record metadata without local ticker.
			c.recordRemote(name, schedule, fn)
			return nil
		} else if !scheduled {
			c.addLocal(name, schedule, dur, fn)
			return nil
		}
		// ZAP schedule succeeded: record metadata without local ticker.
		c.recordRemote(name, schedule, fn)
		return nil
	}

	c.addLocal(name, schedule, dur, fn)
	return nil
}

// Remove cancels a named schedule. No-op if the schedule is not registered.
// For remote schedules, also sends a DELETE to the tasks server.
func (c *Client) Remove(name string) {
	c.mu.Lock()
	entry, ok := c.schedules[name]
	if ok {
		if entry.cancel != nil {
			entry.cancel()
		}
		delete(c.schedules, name)
	}
	c.mu.Unlock()

	if ok && entry.remote {
		c.deleteRemoteSchedule(name)
	}
}

// RemoveAll cancels every registered schedule and clears the registry.
// For remote schedules, issues best-effort DELETE to the tasks server.
func (c *Client) RemoveAll() {
	c.mu.Lock()
	remoteNames := make([]string, 0, len(c.schedules))
	for name, entry := range c.schedules {
		if entry.cancel != nil {
			entry.cancel()
		}
		if entry.remote {
			remoteNames = append(remoteNames, name)
		}
	}
	c.schedules = make(map[string]*scheduleEntry)
	c.mu.Unlock()

	for _, name := range remoteNames {
		c.deleteRemoteSchedule(name)
	}
}

// HasJob reports whether a schedule with the given name is registered.
func (c *Client) HasJob(name string) bool {
	c.mu.RLock()
	defer c.mu.RUnlock()
	_, ok := c.schedules[name]
	return ok
}

// Total returns the number of registered schedules.
func (c *Client) Total() int {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return len(c.schedules)
}

// ActiveTickerCount returns the number of schedules currently ticking
// locally. Remote (server-side) schedules are counted as active when
// registered. Used by the legacy cron.HasStarted() accessor.
func (c *Client) ActiveTickerCount() int {
	c.mu.RLock()
	defer c.mu.RUnlock()
	n := 0
	for _, entry := range c.schedules {
		if entry.remote || entry.cancel != nil {
			n++
		}
	}
	return n
}

// Schedules returns a snapshot of all registered schedules.
// The returned slice is safe for the caller to mutate.
func (c *Client) Schedules() []Schedule {
	c.mu.RLock()
	defer c.mu.RUnlock()
	out := make([]Schedule, 0, len(c.schedules))
	for name, entry := range c.schedules {
		out = append(out, Schedule{Name: name, Expression: entry.expression})
	}
	return out
}

// Run invokes the locally registered callback for a schedule, if any.
// Returns true if a callback was found and executed, false otherwise.
// Used by the admin "run now" endpoint.
func (c *Client) Run(name string) bool {
	c.mu.RLock()
	entry, ok := c.schedules[name]
	var fn func()
	if ok {
		fn = entry.fn
	}
	c.mu.RUnlock()
	if !ok || fn == nil {
		return false
	}
	fn()
	return true
}

// recordRemote stores metadata for a schedule whose ticking is server-side.
func (c *Client) recordRemote(name, expression string, fn func()) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.schedules[name] = &scheduleEntry{expression: expression, fn: fn, remote: true}
}

// PauseAll cancels every local tick goroutine but leaves registry entries
// intact. Use ResumeAll to restart them. Remote (server-side) schedules are
// untouched — they keep firing unless explicitly Removed.
//
// This is the backing for the legacy cron.Cron.Stop() semantics (stop
// ticking without losing jobs).
func (c *Client) PauseAll() {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, entry := range c.schedules {
		if entry.cancel != nil {
			entry.cancel()
			entry.cancel = nil
		}
	}
}

// ResumeAll restarts tick goroutines for every local entry that has an
// associated callback but no active ticker. Remote entries are untouched.
//
// This is the backing for the legacy cron.Cron.Start() semantics.
func (c *Client) ResumeAll() {
	c.mu.Lock()
	entries := make(map[string]*scheduleEntry, len(c.schedules))
	for name, entry := range c.schedules {
		if !entry.remote && entry.fn != nil && entry.cancel == nil {
			entries[name] = entry
		}
	}
	c.mu.Unlock()

	for name, entry := range entries {
		c.startEntryTicker(name, entry)
	}
}

// Enqueue submits a one-shot task for durable execution.
// Prefers ZAP transport when zapAddr is configured, falls back to HTTP.
func (c *Client) Now(taskType string, payload map[string]any) error {
	if c == nil {
		return nil
	}
	if c.tasksURL == "" && c.zapAddr == "" {
		return c.execDirect(taskType, payload)
	}
	if c.zapAddr != "" {
		if err := c.submitZAP(taskType, payload); err == nil {
			return nil
		}
		c.logger.Warn("taskqueue: ZAP submit failed, falling back to HTTP",
			"type", taskType, "zapAddr", c.zapAddr)
	}
	if c.tasksURL != "" {
		return c.submitHTTP(taskType, payload)
	}
	return c.execDirect(taskType, payload)
}

// Stop cancels every local ticker, clears the schedule registry, and closes
// the ZAP connection. Call on graceful shutdown. After Stop(), the Client
// must not be reused.
func (c *Client) Stop() {
	c.mu.Lock()
	for name, entry := range c.schedules {
		if entry.cancel != nil {
			entry.cancel()
		}
		delete(c.schedules, name)
	}
	c.mu.Unlock()
	if c.zapNode != nil {
		c.zapNode.Stop()
	}
}

// connectZAP lazily establishes the ZAP connection to the Tasks server.
//
// The node dials and never listens, so it is not started: with discovery off,
// Start only binds a listener, and with Port 0 that was an ephemeral port on
// every interface. Nothing reaches this node through one. A Call is answered
// on the connection it went out on, the node registers no handler, and no peer
// is told where it is.
func (c *Client) connectZAP() error {
	c.zapOnce.Do(func() {
		c.zapNode = zap.NewNode(zap.NodeConfig{
			NodeID:      "tasks-sdk",
			ServiceType: "_tasks-sdk._tcp",
			NoDiscovery: true,
		})
		if err := c.zapNode.ConnectDirect(c.zapAddr); err != nil {
			c.zapErr = fmt.Errorf("taskqueue: zap connect %s: %w", c.zapAddr, err)
			return
		}
		peers := c.zapNode.Peers()
		if len(peers) > 0 {
			c.zapPeer = peers[0]
		} else {
			c.zapPeer = c.zapAddr
		}
		c.logger.Info("taskqueue: ZAP connected", "addr", c.zapAddr, "peer", c.zapPeer)
	})
	return c.zapErr
}

// submitZAP sends a one-shot task over ZAP (opcode 0x0050).
func (c *Client) submitZAP(taskType string, payload map[string]any) error {
	if err := c.connectZAP(); err != nil {
		return err
	}

	payloadBytes, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("taskqueue: marshal: %w", err)
	}

	b := zap.NewBuilder(len(taskType) + len(payloadBytes) + 64)
	obj := b.StartObject(24)
	obj.SetText(fieldTaskType, taskType)
	obj.SetBytes(fieldPayload, payloadBytes)
	obj.FinishAsRoot()
	flags := uint16(OpcodeTaskSubmit) << 8
	data := b.FinishWithFlags(flags)

	msg, err := zap.Parse(data)
	if err != nil {
		return fmt.Errorf("taskqueue: build zap msg: %w", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	resp, err := c.zapNode.Call(ctx, c.zapPeer, msg)
	if err != nil {
		return fmt.Errorf("taskqueue: zap call: %w", err)
	}

	status := resp.Root().Uint32(respStatus)
	if status != 0 && status != 200 {
		body := resp.Root().Bytes(respBody)
		return fmt.Errorf("taskqueue: zap status %d: %s", status, string(body))
	}
	return nil
}

// scheduleZAP creates a recurring schedule over ZAP (opcode 0x0051).
func (c *Client) scheduleZAP(name string, interval time.Duration) error {
	if err := c.connectZAP(); err != nil {
		return err
	}

	b := zap.NewBuilder(len(name) + 64)
	obj := b.StartObject(24)
	obj.SetText(fieldName, name)
	obj.SetText(fieldInterval, interval.String())
	obj.FinishAsRoot()
	flags := uint16(OpcodeTaskSchedule) << 8
	data := b.FinishWithFlags(flags)

	msg, err := zap.Parse(data)
	if err != nil {
		return fmt.Errorf("taskqueue: build zap msg: %w", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	resp, err := c.zapNode.Call(ctx, c.zapPeer, msg)
	if err != nil {
		return fmt.Errorf("taskqueue: zap call: %w", err)
	}

	status := resp.Root().Uint32(respStatus)
	if status != 0 && status != 200 {
		body := resp.Root().Bytes(respBody)
		return fmt.Errorf("taskqueue: zap status %d: %s", status, string(body))
	}

	c.logger.Info("taskqueue: ZAP schedule created", "name", name, "interval", interval)
	return nil
}

// createCronSchedule creates a durable cron-based schedule on Hanzo Tasks.
func (c *Client) createCronSchedule(name, cronExpr string) error {
	if err := c.postSchedule(name, map[string]any{"cronString": []string{cronExpr}}); err != nil {
		return err
	}
	c.logger.Info("tasks: durable cron schedule created", "name", name, "cron", cronExpr)
	return nil
}

// approximateCron converts a cron expression to a rough duration for local dev fallback.
// Production uses Hanzo Tasks's real cron scheduler; this is only for dev mode tickers.
func approximateCron(expr string) time.Duration {
	fields := strings.Fields(expr)
	switch len(fields) {
	case 6: // second-precision: "*/30 * * * * *"
		if strings.HasPrefix(fields[0], "*/") {
			if n, err := strconv.Atoi(fields[0][2:]); err == nil {
				return time.Duration(n) * time.Second
			}
		}
		return time.Minute
	case 5: // standard cron
		// "*/N * * * *" → every N minutes
		if strings.HasPrefix(fields[0], "*/") {
			if n, err := strconv.Atoi(fields[0][2:]); err == nil {
				return time.Duration(n) * time.Minute
			}
		}
		// "0 */N * * *" → every N hours
		if fields[0] == "0" && strings.HasPrefix(fields[1], "*/") {
			if n, err := strconv.Atoi(fields[1][2:]); err == nil {
				return time.Duration(n) * time.Hour
			}
		}
		// Every minute
		if fields[0] == "*" {
			return time.Minute
		}
		// Specific month-day → monthly
		if fields[2] != "*" {
			return 720 * time.Hour
		}
		// Specific day-of-week → weekly
		if fields[4] != "*" {
			return 168 * time.Hour
		}
		// Specific hour → daily
		if fields[1] != "*" {
			return 24 * time.Hour
		}
		// Hourly
		return time.Hour
	default:
		return time.Hour
	}
}

// addLocal runs fn on a ticker (dev fallback, no persistence).
// expression is the raw "30s" / "*/5 * * * *" string as passed to Add().
// interval is the resolved tick duration.
func (c *Client) addLocal(name, expression string, interval time.Duration, fn func()) {
	c.mu.Lock()
	entry := &scheduleEntry{
		expression: expression,
		interval:   interval,
		fn:         fn,
	}
	c.schedules[name] = entry
	c.mu.Unlock()

	c.startEntryTicker(name, entry)
}

// startEntryTicker spins up the goroutine that fires entry.fn on its interval.
// Must be called with the entry already stored in c.schedules.
func (c *Client) startEntryTicker(name string, entry *scheduleEntry) {
	if entry.fn == nil {
		return // nothing to tick
	}

	ctx, cancel := context.WithCancel(context.Background())

	c.mu.Lock()
	if entry.cancel != nil {
		entry.cancel()
	}
	entry.cancel = cancel
	c.mu.Unlock()

	c.logger.Info("taskqueue: local schedule started", "name", name, "interval", entry.interval)

	go func(fn func(), interval time.Duration) {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				func() {
					defer func() {
						if r := recover(); r != nil {
							c.logger.Error("taskqueue: schedule panic", "name", name, "panic", r)
						}
					}()
					fn()
				}()
			}
		}
	}(entry.fn, entry.interval)
}

// deleteRemoteSchedule issues a best-effort DELETE to the tasks server.
// Errors are logged but not returned — local state is already cleared.
func (c *Client) deleteRemoteSchedule(name string) {
	req, err := http.NewRequest(http.MethodDelete, c.tasksURL+namespacePath+"/schedules/"+name, nil)
	if err != nil {
		c.logger.Warn("tasks: remote delete request build failed", "name", name, "error", err)
		return
	}
	resp, err := c.httpClient.Do(req)
	if err != nil {
		c.logger.Warn("tasks: remote delete failed", "name", name, "error", err)
		return
	}
	resp.Body.Close()
}

// createSchedule creates a durable schedule on Hanzo Tasks.
func (c *Client) createSchedule(name string, interval time.Duration) error {
	if err := c.postSchedule(name, map[string]any{"interval": []map[string]any{{"interval": interval.String()}}}); err != nil {
		return err
	}
	c.logger.Info("taskqueue: durable schedule created", "name", name, "interval", interval)
	return nil
}

// postSchedule creates or replaces the schedule `name` with the given spec. A
// create is POST to the collection and replaces a schedule of the same id, so
// registering a job twice leaves one schedule.
func (c *Client) postSchedule(name string, spec map[string]any) error {
	return c.post(namespacePath+"/schedules", map[string]any{
		"scheduleId": name,
		"spec":       spec,
		"action": map[string]any{
			"workflowType": map[string]any{"name": name},
			"taskQueue":    taskQueue,
		},
	})
}

// post sends v as JSON to path on the tasks server and fails on any non-2xx.
func (c *Client) post(path string, v any) error {
	body, err := json.Marshal(v)
	if err != nil {
		return fmt.Errorf("marshal: %w", err)
	}
	req, err := http.NewRequest(http.MethodPost, c.tasksURL+path, bytes.NewReader(body))
	if err != nil {
		return fmt.Errorf("request: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := c.httpClient.Do(req)
	if err != nil {
		return fmt.Errorf("http: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return fmt.Errorf("status %d", resp.StatusCode)
	}
	return nil
}

// execDirect runs the handler in a goroutine (dev mode).
func (c *Client) execDirect(taskType string, payload map[string]any) error {
	c.mu.RLock()
	h := c.handler
	c.mu.RUnlock()

	if h == nil {
		c.logger.Warn("taskqueue: no handler, dropping task", "type", taskType)
		return nil
	}

	go h(taskType, payload)
	return nil
}

// submitHTTP starts a one-shot workflow on the Hanzo Tasks HTTP API, and runs
// the task here when the server cannot take it.
func (c *Client) submitHTTP(taskType string, payload map[string]any) error {
	err := c.post(namespacePath+"/workflows", map[string]any{
		"workflowType": map[string]any{"name": taskType},
		"taskQueue":    map[string]any{"name": taskQueue},
		"input":        payload,
	})
	if err != nil {
		c.logger.Warn("taskqueue: server refused, executing directly", "type", taskType, "error", err)
		return c.execDirect(taskType, payload)
	}
	return nil
}