orchestra/cmd/orchestra/main.go

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"orchestra/internal/authz"
	"orchestra/internal/domain"
	"orchestra/internal/registry"
	"orchestra/internal/router"
	"orchestra/internal/store"
	"os"
	"strconv"
	"strings"
	"time"
)

func id() string { return domain.NewID() }
func main() {
	dir := os.Getenv("ORCHESTRA_DATA")
	if dir == "" {
		dir = "./data"
	}
	s, err := store.Open(dir)
	if err != nil {
		log.Fatal(err)
	}
	var rr registry.Registry
	var rt *router.Router
	if config := os.Getenv("ORCHESTRA_CONFIG"); config != "" {
		if rr, err = registry.Load(config); err != nil {
			log.Fatalf("load orchestra config: %v", err)
		}
		rt = &router.Router{Store: s, Registry: rr, Reachability: registry.TCPReachability{}, Timeout: time.Second, Retry: router.RetryPolicy{MaxAttempts: 3, Backoff: time.Minute}}
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/v1/tasks", func(w http.ResponseWriter, r *http.Request) {
		if r.Method == "GET" {
			json.NewEncoder(w).Encode(s.Tasks())
			return
		}
		if r.Method != "POST" {
			http.Error(w, "method not allowed", 405)
			return
		}
		var p map[string]any
		if json.NewDecoder(r.Body).Decode(&p) != nil {
			http.Error(w, "invalid json", 400)
			return
		}
		b, _ := json.Marshal(p)
		e := domain.Event{ID: id(), Type: "TaskCreated", TaskID: id(), Version: 1, Payload: b}
		if err := s.Append(e); err != nil {
			http.Error(w, err.Error(), 400)
			return
		}
		if rt != nil {
			if _, err := rt.HandleEvent(e); err != nil {
				log.Printf("route task: %v", err)
			}
		}
		e = s.Events(0)[len(s.Events(0))-1]
		w.WriteHeader(201)
		json.NewEncoder(w).Encode(e)
	})
	mux.HandleFunc("/v1/events", func(w http.ResponseWriter, r *http.Request) {
		var n uint64
		if x, err := strconv.ParseUint(r.URL.Query().Get("since"), 10, 64); err == nil {
			n = x
		}
		json.NewEncoder(w).Encode(s.Events(n))
	})
	mux.HandleFunc("/v1/tasks/", func(w http.ResponseWriter, r *http.Request) {
		parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
		if len(parts) < 4 || len(parts) > 5 || r.Method != "POST" {
			http.Error(w, "not found", http.StatusNotFound)
			return
		}
		taskID, action := parts[2], parts[3]
		if action == "approval" {
			if len(parts) == 5 {
				if parts[4] != "grant" && parts[4] != "deny" {
					http.Error(w, "unknown approval action", 404)
					return
				}
				t, ok := s.Task(taskID)
				if !ok {
					http.Error(w, "task not found", 404)
					return
				}
				by := r.Header.Get("X-Orchestra-Actor")
				if by == "" {
					by = "surface"
				}
				typ := "ApprovalGranted"
				if parts[4] == "deny" {
					typ = "ApprovalDenied"
				}
				b, _ := json.Marshal(map[string]any{"subject_ref": taskID, "by": by})
				e := domain.Event{ID: id(), Type: typ, TaskID: taskID, Version: t.Version + 1, Payload: b}
				if err := s.Append(e); err != nil {
					http.Error(w, err.Error(), 409)
					return
				}
				json.NewEncoder(w).Encode(e)
				return
			}
			var p struct {
				Options []any `json:"options"`
			}
			if json.NewDecoder(r.Body).Decode(&p) != nil || len(p.Options) == 0 {
				http.Error(w, "options required", 400)
				return
			}
			if _, ok := s.Task(taskID); !ok {
				http.Error(w, "task not found", 404)
				return
			}
			b, _ := json.Marshal(map[string]any{"subject_ref": taskID, "options": p.Options})
			t, _ := s.Task(taskID)
			e := domain.Event{ID: id(), Type: "ApprovalRequested", TaskID: taskID, Version: t.Version + 1, Payload: b}
			if err := s.Append(e); err != nil {
				http.Error(w, err.Error(), 400)
				return
			}
			json.NewEncoder(w).Encode(e)
			return
		}
		var e domain.Event
		var err error
		switch action {
		case "lease":
			var p struct {
				HarnessID  string `json:"harness_id"`
				TTLSeconds int    `json:"ttl_seconds"`
			}
			if json.NewDecoder(r.Body).Decode(&p) != nil || p.HarnessID == "" {
				http.Error(w, "harness_id required", 400)
				return
			}
			if p.TTLSeconds <= 0 {
				p.TTLSeconds = 1800
			}
			e, err = s.Lease(taskID, p.HarnessID, time.Duration(p.TTLSeconds)*time.Second)
		case "release", "complete", "block":
			t, ok := s.Task(taskID)
			if !ok {
				http.Error(w, "task not found", 404)
				return
			}
			types := map[string]string{"release": "TaskReleased", "complete": "TaskCompleted", "block": "TaskBlocked"}
			e = domain.Event{ID: id(), Type: types[action], TaskID: taskID, Version: t.Version + 1, Payload: json.RawMessage(`{"source":"api"}`)}
			err = s.Append(e)
		default:
			http.Error(w, "unknown action", 404)
			return
		}
		if err != nil {
			http.Error(w, err.Error(), 409)
			return
		}
		if rt != nil {
			if _, routeErr := rt.HandleEvent(e); routeErr != nil {
				log.Printf("route task: %v", routeErr)
			}
		}
		json.NewEncoder(w).Encode(e)
	})
	if rt != nil {
		go func() {
			ticker := time.NewTicker(time.Second)
			defer ticker.Stop()
			for range ticker.C {
				if _, err := s.ExpireLeases(time.Now()); err != nil {
					log.Printf("expire leases: %v", err)
				} else if _, err := rt.AssignPending(); err != nil {
					log.Printf("route expired task: %v", err)
				}
			}
		}()
	}
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) })
	port := os.Getenv("ORCHESTRA_PORT")
	if port == "" {
		port = "9145"
	}
	log.Println("orchestra listening on :" + port)
	tokens := map[authz.Surface]string{
		authz.TUI: os.Getenv("ORCHESTRA_TUI_TOKEN"), authz.Web: os.Getenv("ORCHESTRA_WEB_TOKEN"),
		authz.MCP: os.Getenv("ORCHESTRA_MCP_TOKEN"), authz.Maven: os.Getenv("ORCHESTRA_MAVEN_TOKEN"),
		authz.Telegram: os.Getenv("ORCHESTRA_TELEGRAM_TOKEN"), authz.Ntfy: os.Getenv("ORCHESTRA_NTFY_TOKEN"),
	}
	log.Fatal(http.ListenAndServe(":"+port, authz.HTTP(tokens, mux)))
}
func auth(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		surface := strings.ToLower(r.Header.Get("X-Orchestra-Surface"))
		if surface == "telegram" || surface == "ntfy" {
			if r.Method != "GET" {
				http.Error(w, "notify-only surface", 403)
				return
			}
		}
		next.ServeHTTP(w, r)
	})
}