tasks/cmd/tasksd/main.go

// Copyright © 2026 Hanzo AI. MIT License.
//
// tasksd is the Hanzo Tasks daemon: one Go binary, native ZAP transport,
// IAM JWT auth, embedded React UI. Zero gRPC, zero go.temporal.io.

package main

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"

	"strings"

	"github.com/hanzoai/authz/edge"
	"github.com/hanzoai/tasks/pkg/auth"
	"github.com/hanzoai/tasks/pkg/tasks"
	"github.com/hanzoai/tasks/pkg/tasks/migration"
	"github.com/hanzoai/tasks/pkg/tasks/replication"
	"github.com/hanzoai/tasks/pkg/tasks/replication/quasar"
	"github.com/hanzoai/tasks/pkg/tasks/routing"
	"github.com/hanzoai/tasks/pkg/tasks/store"
	tasksui "github.com/hanzoai/tasks/ui"
)

// masterKeyEnv supplies the 32-byte root key every shard is encrypted
// under, base64-encoded — the same encoding the platform's KMS hands to
// every other Hanzo store. Unset leaves shards plaintext, which is the
// zero-config dev posture; a set-but-malformed value is fatal, so a
// misconfigured deployment refuses to boot rather than quietly writing
// tenant state in the clear.
const masterKeyEnv = "KMS_MASTER_KEY"

// masterKey decodes masterKeyEnv.
func masterKey() ([]byte, error) {
	raw := strings.TrimSpace(os.Getenv(masterKeyEnv))
	if raw == "" {
		return nil, nil
	}
	k, err := base64.StdEncoding.DecodeString(raw)
	if err != nil {
		return nil, fmt.Errorf("%s must be base64: %w", masterKeyEnv, err)
	}
	if len(k) != store.MasterKeyLen {
		return nil, fmt.Errorf("%s must decode to %d bytes, got %d", masterKeyEnv, store.MasterKeyLen, len(k))
	}
	return k, nil
}

// storeNew is a thin alias used by runMigrate so the rest of main.go
// doesn't import the store package directly.
func storeNew(dir string, master []byte) (*store.Manager, error) { return store.New(dir, master) }

// newCoordinator is a thin closure-bound migration runner used by runMigrate.
func newCoordinator(mgr *store.Manager, rep replication.Replicator) func(ctx context.Context, principal, ns, from, to string) (*migration.Job, error) {
	c := migration.NewCoordinator(mgr, rep)
	return func(ctx context.Context, principal, ns, from, to string) (*migration.Job, error) {
		return c.Migrate(ctx, migration.Job{Principal: principal, Namespace: ns, From: from, To: to})
	}
}

func main() {
	// One subcommand: `migrate` moves ONE shard between NODES.
	if len(os.Args) > 1 && os.Args[1] == "migrate" {
		runMigrate(os.Args[2:])
		return
	}
	var (
		zapAddr    = flag.String("zap", envStr("TASKS_ZAP_ADDR", ":9999"), "ZAP listen address: a host:port, or a path to bind a unix socket")
		httpAddr   = flag.String("http", envStr("TASKS_HTTP_ADDR", ":7243"), "HTTP listen address (UI + healthz)")
		dataDir    = flag.String("data", envStr("TASKSD_DATA_DIR", envStr("TASKS_DATA_DIR", "./tasks-data")), "Tasks persistence directory")
		ns         = flag.String("namespace", envStr("TASKS_NAMESPACE", "default"), "Default namespace")
		replKind   = flag.String("replicator", envStr("TASKSD_REPLICATOR", "local"), "Replicator driver (local | quasar)")
		nodeID     = flag.String("node-id", envStr("TASKSD_NODE_ID", ""), "Stable node identifier (required for quasar)")
		validators = flag.String("validators", envStr("TASKSD_VALIDATORS", ""), "Comma-separated validator host:port (required for quasar)")
		_          = flag.String("validator-key", envStr("TASKSD_VALIDATOR_KEY", ""), "Path to PQ validator key (reserved)")
	)
	flag.Parse()

	logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelInfo}))
	slog.SetDefault(logger)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	requireID := envBool("TASKSD_REQUIRE_IDENTITY", false)
	validator := auth.NewValidator(auth.JWTConfig{
		JWKSURL: envStr("TASKSD_JWKS_URL", ""),
		Issuer:  envStr("TASKSD_JWT_ISSUER", ""),
		// The brands this tasksd accepts tokens from beyond the primary — each signs
		// under its own issuer. Adding one is configuration, not a code change.
		Issuers:  strings.Split(envStr("TASKSD_WHITELABEL_ISSUERS", ""), ","),
		Audience: envStr("TASKSD_JWT_AUDIENCE", ""),
	})

	repl, router, err := buildReplicator(ctx, *replKind, *nodeID, *validators)
	if err != nil {
		logger.Error("replicator", "err", err)
		os.Exit(1)
	}
	master, err := masterKey()
	if err != nil {
		logger.Error("master key", "err", err)
		os.Exit(1)
	}

	srv, err := tasks.Embed(ctx, tasks.EmbedConfig{
		DataDir:         *dataDir,
		MasterKey:       master,
		Address:         *zapAddr,
		Namespace:       *ns,
		Logger:          logger,
		JWTValidator:    validator,
		RequireIdentity: requireID,
		Replicator:      repl,
		Router:          router,
		NodeID:          *nodeID,
	})
	if err != nil {
		logger.Error("tasks.Embed", "err", err)
		os.Exit(1)
	}
	defer srv.Stop(context.Background())
	logger.Info("zap listener", "addr", srv.Address(), "service", "_tasks._tcp")

	httpSrv := &http.Server{
		Addr:              *httpAddr,
		Handler:           buildHTTP(*ns, srv, validator, requireID),
		ReadHeaderTimeout: 10 * time.Second,
	}
	go func() {
		logger.Info("http listener", "addr", *httpAddr)
		if err := httpSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			logger.Error("http", "err", err)
			stop()
		}
	}()

	<-ctx.Done()
	logger.Info("shutting down")
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = httpSrv.Shutdown(shutdownCtx)
}

// buildHTTP serves /healthz, /v1/tasks/* (browser JSON shim), and
// /_/tasks/* (embedded React UI). The /v1/tasks/* surface is the same
// data the ZAP node serves on :9999 — same model functions, no drift.
// Identity headers (X-User-Id, X-Org-Id, X-User-Email) are populated by
// hanzoai/gateway after IAM JWT validation; auth.RequireIdentity attaches
// them to the request context. TASKSD_REQUIRE_IDENTITY=true (production)
// rejects requests without identity headers; default false keeps the
// embedded/dev path working without a gateway.
//
// /healthz, /v1/tasks/health, and /_/tasks/* are intentionally
// unauthenticated: probes and the SPA shell run before any session
// exists.
func buildHTTP(ns string, srv *tasks.Embedded, validator *edge.Verifier, requireID bool) http.Handler {
	mux := http.NewServeMux()

	probe := func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		_ = json.NewEncoder(w).Encode(map[string]any{
			"service":   "tasks",
			"status":    "ok",
			"namespace": ns,
		})
	}
	mux.HandleFunc("/healthz", probe)
	mux.HandleFunc("/v1/tasks/health", probe)

	identity := auth.RequireIdentity(validator, requireID)

	// /v1/tasks/settings is the SPA bootstrap — capability flags only,
	// no per-org data, intentionally unauthenticated so the login page
	// itself can render.
	mux.Handle("/v1/tasks/settings", srv.HTTPHandler())

	mux.Handle("/v1/tasks/cluster", srv.ClusterHandler())
	mux.Handle("/v1/tasks/cluster/health", srv.ClusterHandler())
	mux.Handle("/v1/tasks/", identity(srv.HTTPHandler()))
	mux.Handle("/v1/tasks/mcp", identity(srv.MCPHandler()))
	mux.Handle("/v1/tasks/events", identity(srv.EventsHandler()))

	// React Router has basename="/_/tasks". The bundle is only valid at
	// that prefix; serving it at "/" produces a blank page because
	// the router refuses to match. Redirect "/" → "/_/tasks/".
	mux.Handle("/_/tasks/", http.StripPrefix("/_/tasks", tasksui.Handler()))
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/" {
			http.NotFound(w, r)
			return
		}
		http.Redirect(w, r, "/_/tasks/", http.StatusFound)
	})

	return mux
}

func envBool(k string, def bool) bool {
	if v := os.Getenv(k); v != "" {
		if b, err := strconv.ParseBool(v); err == nil {
			return b
		}
	}
	return def
}

func envStr(k, def string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return def
}

// runMigrate is the `tasksd migrate ...` subcommand. Quiesces the
// (tenant, namespace) shard via the consensus barrier, copies the
// SQLite file under <dataDir>/_migrations/<id>/, and releases the
// barrier. Idempotent.
func runMigrate(args []string) {
	fs := flag.NewFlagSet("migrate", flag.ExitOnError)
	dataDir := fs.String("data", envStr("TASKSD_DATA_DIR", "./tasks-data"), "Tasks persistence directory")
	org := fs.String("org", "", "Org id (empty = root tenant)")
	project := fs.String("project", "", "Project id (optional)")
	ns := fs.String("namespace", "", "Namespace to migrate")
	to := fs.String("to", "", "Destination node id")
	from := fs.String("from", "", "Source node id (annotation only)")
	_ = fs.Parse(args)
	if *ns == "" || *to == "" {
		fmt.Fprintln(os.Stderr, "usage: tasksd migrate [--org <id> [--project <id>]] --namespace <name> --to <node> [--from <node>]")
		os.Exit(2)
	}
	// Through the door: the operator writes the org as IAM spells it, and
	// store.Org folds it to the identity the shard is actually stored under.
	// A --user flag used to ride alongside these; the shared namespace layout
	// has no place for a user-scoped shard, so there is nothing to migrate.
	principal := store.OrgProject(*org, *project)
	if err := principal.Valid(); err != nil {
		fmt.Fprintln(os.Stderr, "migrate:", err)
		os.Exit(2)
	}
	master, err := masterKey()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
	mgr, err := storeNew(*dataDir, master)
	if err != nil {
		fmt.Fprintln(os.Stderr, "migrate:", err)
		os.Exit(1)
	}
	defer mgr.Close()
	rep := replication.NewLocal()
	mgr.WithReplicator(rep)
	c := newCoordinator(mgr, rep)
	job, err := c(context.Background(), principal.String(), *ns, *from, *to)
	if err != nil {
		fmt.Fprintln(os.Stderr, "migrate:", err)
		os.Exit(1)
	}
	out, _ := json.MarshalIndent(job, "", "  ")
	fmt.Println(string(out))
}

// buildReplicator wires the requested driver. local → passthrough,
// quasar → consensus.NewPQ over the in-process MemoryHub (single-pod
// boot path); when validators are non-empty the hub is replaced with a
// real network transport at the next deploy phase.
func buildReplicator(ctx context.Context, kind, node, validatorsCSV string) (replication.Replicator, routing.Router, error) {
	router := routing.NewHash([]byte("tasks"))
	switch strings.ToLower(kind) {
	case "", "local":
		router.SetMembership(routing.NodeID(stringOr(node, "tasks-embed")), []routing.NodeID{routing.NodeID(stringOr(node, "tasks-embed"))})
		return replication.NewLocal(), router, nil
	case "quasar":
		if node == "" {
			return nil, nil, fmt.Errorf("quasar replicator requires --node-id")
		}
		hub := quasar.NewMemoryHub()
		t := hub.Connect(node)
		var vs []routing.NodeID
		for _, v := range strings.Split(validatorsCSV, ",") {
			v = strings.TrimSpace(v)
			if v != "" {
				vs = append(vs, routing.NodeID(v))
			}
		}
		if len(vs) == 0 {
			vs = []routing.NodeID{routing.NodeID(node)}
		}
		router.SetMembership(routing.NodeID(node), vs)
		validators := make([]string, len(vs))
		for i, v := range vs {
			validators[i] = string(v)
		}
		rep, err := quasar.NewQuasar(ctx, quasar.QuasarConfig{
			NodeID:     node,
			Validators: validators,
			Transport:  t,
		})
		if err != nil {
			return nil, nil, err
		}
		return rep, router, nil
	default:
		return nil, nil, fmt.Errorf("unknown replicator: %s (want local | quasar)", kind)
	}
}

func stringOr(a, b string) string {
	if a != "" {
		return a
	}
	return b
}