ThothII/tools/tht/internal/setup/run.go

// Package setup orchestrates the safe local bootstrap of a ThothII installation.
package setup

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/aritmolab/thothii/tools/tht/internal/compose"
	"github.com/aritmolab/thothii/tools/tht/internal/config"
	"github.com/aritmolab/thothii/tools/tht/internal/pi"
	"github.com/aritmolab/thothii/tools/tht/internal/project"
)

const setupHealthTimeout = 5 * time.Minute

var healthPollInterval = time.Second

// Result records the completed setup phases. DescriptorPath always identifies the descriptor
// selected by this invocation, including an idempotent rerun.
type Result struct {
	DescriptorPath string
	ProjectName    string
	Configured     bool
	Built          bool
	Started        bool
	Healthy        bool
}

// Run validates the host, creates or validates non-secret configuration, and by default builds,
// starts, and verifies the current checkout. ConfigureOnly stops after Compose rendering.
func Run(ctx context.Context, runner compose.Runner, request Request, input io.Reader, output io.Writer) (Result, error) {
	if runner == nil {
		return Result{}, errors.New("setup requires a Docker command runner")
	}
	root, err := project.Discover(request.ProjectRoot)
	if err != nil {
		return Result{}, fmt.Errorf("setup project discovery: %w", err)
	}
	request.ProjectRoot = root.Path
	if err := checkHost(ctx, runner, root.Path); err != nil {
		return Result{}, err
	}

	files, err := EnsureFiles(request, input, output)
	if err != nil {
		return Result{}, err
	}
	installation, err := config.Load(files.DescriptorPath)
	if err != nil {
		return Result{}, fmt.Errorf("setup generated configuration is invalid: %w", err)
	}
	result := Result{DescriptorPath: files.DescriptorPath, ProjectName: installation.ProjectName(), Configured: true}
	if err := runCompose(ctx, runner, installation, "config", "--quiet"); err != nil {
		return Result{}, fmt.Errorf("setup Compose configuration: %w", err)
	}
	if request.ConfigureOnly {
		fmt.Fprintf(output, "Configuration is ready: %s\n", result.DescriptorPath)
		return result, nil
	}
	if err := runCompose(ctx, runner, installation, "build"); err != nil {
		return Result{}, fmt.Errorf("setup image build: %w", err)
	}
	result.Built = true
	if err := runCompose(ctx, runner, installation, "up", "--detach", "--remove-orphans"); err != nil {
		return Result{}, withStartupRecovery(fmt.Errorf("setup stack start: %w", err), "core")
	}
	result.Started = true
	if err := waitForHealthyServices(ctx, runner, installation); err != nil {
		return Result{}, withStartupRecovery(err, recoveryService(err))
	}
	result.Healthy = true
	if err := aggregateDoctor(ctx, runner, installation); err != nil {
		return Result{}, withStartupRecovery(err, "core")
	}
	controlled := compose.InstallationRunner{Installation: installation, Runner: runner}
	if err := pi.Doctor(ctx, controlled); err != nil {
		return Result{}, withStartupRecovery(fmt.Errorf("setup Pi doctor: %w", err), "core")
	}
	fmt.Fprintf(output, "ThothII is ready at %s\nInstallation descriptor: %s\nNext: tht status\n", frontendURL(installation), result.DescriptorPath)
	return result, nil
}

func checkHost(ctx context.Context, runner compose.Runner, root string) error {
	checks := []struct {
		name string
		args []string
	}{
		{name: "Docker Engine", args: []string{"version", "--format", "{{.Server.Version}}"}},
		{name: "Docker Compose", args: []string{"compose", "version", "--short"}},
		{name: "supported Docker architecture", args: []string{"version", "--format", "{{.Server.Arch}}"}},
	}
	for _, check := range checks {
		result, err := runner.Run(ctx, check.args, nil)
		if err != nil || strings.TrimSpace(result.Stdout) == "" {
			return fmt.Errorf("setup %s check failed", check.name)
		}
		if check.name == "supported Docker architecture" && !supportedArchitecture(result.Stdout) {
			return fmt.Errorf("setup Docker architecture %q is not supported", strings.TrimSpace(result.Stdout))
		}
	}
	if err := requireLF(root); err != nil {
		return fmt.Errorf("setup line-ending check: %w", err)
	}
	return nil
}

func supportedArchitecture(value string) bool {
	switch strings.ToLower(strings.TrimSpace(value)) {
	case "amd64", "x86_64", "arm64", "aarch64":
		return true
	default:
		return false
	}
}

func runCompose(ctx context.Context, runner compose.Runner, installation config.Installation, command ...string) error {
	result, err := runner.Run(ctx, installation.ComposeArgs(command...), nil)
	if err != nil {
		return composeFailure(result, err)
	}
	return nil
}

func composeFailure(result compose.Result, cause error) error {
	if result.ExitCode != 0 {
		return fmt.Errorf("Docker exited with status %d", result.ExitCode)
	}
	return cause
}

func aggregateDoctor(ctx context.Context, runner compose.Runner, installation config.Installation) error {
	result, err := runner.Run(ctx, installation.ComposeArgs("config", "--format", "json"), nil)
	if err != nil {
		return fmt.Errorf("setup doctor: %w", composeFailure(result, err))
	}
	if err := requireVolumes(result.Stdout); err != nil {
		return fmt.Errorf("setup doctor: %w", err)
	}
	return nil
}

func withStartupRecovery(cause error, service string) error {
	if service == "" {
		service = "core"
	}
	return fmt.Errorf("%w; containers were left running for diagnosis: tht logs %s; then run tht status", cause, service)
}

func recoveryService(cause error) string {
	var healthFailure healthFailure
	if errors.As(cause, &healthFailure) && healthFailure.service != "" {
		return healthFailure.service
	}
	return "core"
}

type serviceStatus struct {
	Service  string          `json:"Service"`
	State    string          `json:"State"`
	Health   string          `json:"Health"`
	ExitCode json.RawMessage `json:"ExitCode"`
}

type healthFailure struct {
	service string
	state   string
}

func (e healthFailure) Error() string {
	return fmt.Sprintf("setup health timed out waiting for %s (last state: %s)", e.service, e.state)
}

func waitForHealthyServices(ctx context.Context, runner compose.Runner, installation config.Installation) error {
	healthContext, cancel := context.WithTimeout(ctx, setupHealthTimeout)
	defer cancel()
	lastService, lastState := "core", "unknown"
	for {
		result, err := runner.Run(healthContext, installation.ComposeArgs("ps", "--all", "--format", "json"), nil)
		if err == nil {
			statuses, parseErr := parseServiceStatuses(result.Stdout)
			if parseErr == nil {
				if service, state, ready := healthyServices(statuses); ready {
					return nil
				} else {
					lastService, lastState = service, state
				}
			} else {
				lastState = "Compose returned invalid status"
			}
		} else if result.ExitCode != 0 {
			lastState = fmt.Sprintf("Compose exited with status %d", result.ExitCode)
		} else {
			lastState = "Compose status command failed"
		}
		select {
		case <-healthContext.Done():
			return healthFailure{service: lastService, state: lastState}
		case <-time.After(healthPollInterval):
		}
	}
}

func parseServiceStatuses(value string) ([]serviceStatus, error) {
	var statuses []serviceStatus
	if err := json.Unmarshal([]byte(value), &statuses); err == nil && len(statuses) > 0 {
		return statuses, nil
	}
	decoder := json.NewDecoder(strings.NewReader(value))
	for {
		var status serviceStatus
		err := decoder.Decode(&status)
		if errors.Is(err, io.EOF) {
			break
		}
		if err != nil {
			return nil, errors.New("Compose returned invalid service status")
		}
		statuses = append(statuses, status)
	}
	if len(statuses) == 0 {
		return nil, errors.New("Compose returned invalid service status")
	}
	return statuses, nil
}

func healthyServices(statuses []serviceStatus) (string, string, bool) {
	wanted := map[string]bool{"core": false, "frontend": false, "qdrant": false, "embedding": false, "embedding-model-init": false}
	for _, status := range statuses {
		if _, required := wanted[status.Service]; !required {
			continue
		}
		if status.Service == "embedding-model-init" {
			if strings.EqualFold(status.State, "exited") && exitCodeZero(status.ExitCode) {
				wanted[status.Service] = true
				continue
			}
			return status.Service, status.State, false
		}
		if strings.EqualFold(status.State, "running") && strings.EqualFold(status.Health, "healthy") {
			wanted[status.Service] = true
			continue
		}
		return status.Service, strings.TrimSpace(status.State + "/" + status.Health), false
	}
	for _, service := range []string{"core", "frontend", "qdrant", "embedding", "embedding-model-init"} {
		if !wanted[service] {
			return service, "not reported by Docker Compose", false
		}
	}
	return "", "", true
}

func exitCodeZero(value json.RawMessage) bool {
	if len(value) == 0 || string(value) == "null" {
		return false
	}
	var number int
	if json.Unmarshal(value, &number) == nil {
		return number == 0
	}
	var text string
	return json.Unmarshal(value, &text) == nil && strings.TrimSpace(text) == "0"
}

func frontendURL(installation config.Installation) string {
	port, err := installation.EnvironmentValue("THOTH_HTTP_PORT")
	if err != nil || strings.TrimSpace(port) == "" {
		port = "8080"
	}
	if number, err := strconv.Atoi(port); err != nil || number < 1 || number > 65535 {
		port = "8080"
	}
	return "http://127.0.0.1:" + port
}

func requireVolumes(renderedConfig string) error {
	var document struct {
		Volumes map[string]json.RawMessage `json:"volumes"`
	}
	if err := json.Unmarshal([]byte(renderedConfig), &document); err != nil {
		return errors.New("Compose returned invalid rendered configuration")
	}
	for _, volume := range []string{"settings", "pi-state", "workspace-registry", "workspace-secrets", "sessions", "qdrant-data", "embedding-models"} {
		if _, exists := document.Volumes[volume]; !exists {
			return fmt.Errorf("rendered Compose configuration is missing required volume %s", volume)
		}
	}
	return nil
}

func requireLF(root string) error {
	for _, path := range []string{filepath.Join(root, "compose.yaml"), filepath.Join(root, "deploy"), filepath.Join(root, "docker")} {
		if err := requireLFPath(path); err != nil {
			return err
		}
	}
	return nil
}

func requireLFPath(path string) error {
	info, err := os.Lstat(path)
	if errors.Is(err, os.ErrNotExist) {
		return nil
	}
	if err != nil || info.Mode()&os.ModeSymlink != 0 {
		return nil
	}
	if !info.IsDir() {
		return requireLFFile(path, info.Mode())
	}
	return filepath.WalkDir(path, func(path string, entry os.DirEntry, walkErr error) error {
		if walkErr != nil {
			return walkErr
		}
		if entry.IsDir() || entry.Type()&os.ModeSymlink != 0 {
			return nil
		}
		return requireLFFile(path, entry.Type())
	})
}

func requireLFFile(path string, mode os.FileMode) error {
	if !mode.IsRegular() || !requiresLF(filepath.Base(path)) {
		return nil
	}
	contents, err := os.ReadFile(path)
	if err != nil {
		return err
	}
	if strings.Contains(string(contents), "\r\n") {
		return fmt.Errorf("CRLF line endings found in %s", filepath.Base(path))
	}
	return nil
}

func requiresLF(name string) bool {
	if name == "Dockerfile" || strings.HasPrefix(name, "Dockerfile.") || strings.HasSuffix(name, ".Dockerfile") {
		return true
	}
	for _, suffix := range []string{".sh", ".yml", ".yaml"} {
		if strings.HasSuffix(name, suffix) {
			return true
		}
	}
	return false
}