eclipse/cmd/eclipse-httpd/main.go

// Copyright (C) 2023-2024  Umorpha Systems
// SPDX-License-Identifier: AGPL-3.0-or-later

package main

import (
	"context"
	"embed"
	"encoding/json"
	"errors"
	"fmt"
	"html/template"
	"io"
	"io/fs"
	"log"
	"mime"
	"mime/multipart"
	"net/http"
	"net/url"
	"os"
	"path"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"git.lukeshu.com/go/containers/typedsync"
	"github.com/datawire/ocibuild/pkg/cliutil"
	"github.com/golang/gddo/httputil"
	"github.com/spf13/cobra"

	source "git.mothstuff.lol/lukeshu/eclipse"
	"git.mothstuff.lol/lukeshu/eclipse/lib/common"
	"git.mothstuff.lol/lukeshu/eclipse/lib/leader"
)

func main() {
	argparser := &cobra.Command{
		Use:   os.Args[0] + " [flags]",
		Short: "The leader HTTP server",
		Args:  cliutil.WrapPositionalArgs(cobra.NoArgs),

		SilenceErrors: true, // we'll handle this ourselves after .ExecuteContext()
		SilenceUsage:  true, // FlagErrorFunc will handle this

		CompletionOptions: cobra.CompletionOptions{
			DisableDefaultCmd: true,
		},
	}
	argparser.SetFlagErrorFunc(cliutil.FlagErrorFunc)
	argparser.SetHelpTemplate(cliutil.HelpTemplate)

	var cfgFile string
	argparser.Flags().StringVar(&cfgFile, "config", source.DefaultLeaderConfigFile,
		"Config file to use")
	_ = argparser.MarkFlagFilename("config", "yml", "yaml")

	var socket string
	argparser.Flags().StringVar(&socket, "socket", "tcp::8080",
		"the socket to listen on; a \"family:address\" pair, where the\n"+
			"\"family\" is one of \"tcp\", \"tcp4\", \"tcp6\", \"unix\", \"unixpacket\"\n"+
			"(the usual Go net.Listen() network names), or \"fd\" (to listen\n"+
			"on an already-open file descriptor).  The \"address\" part for\n"+
			"the \"fd\" family is one of\n"+
			"  1. a file descriptor number (or the special names \"stdin\",\n"+
			"     \"stdout\", or \"stderr\" to refer to 0, 1, or 2\n"+
			"     respectively),\n"+
			"  2. \"systemd:N\" to refer to the Nth file descriptor passed in\n"+
			"     via systemd (or \"systemd\" as shorthand for \"systemd:0\")\n"+
			"     (see sd_listend_fds(3)),\n"+
			"  3. \"systemd:NAME\" to refer to a named file descriptor passed\n"+
			"     in via systemd (see sd_listend_fds_with_names(3)).\n")

	argparser.RunE = func(cmd *cobra.Command, args []string) error {
		stype, saddr, ok := strings.Cut(socket, ":")
		if !ok {
			return cliutil.FlagErrorFunc(cmd, fmt.Errorf("invalid address: %q", socket))
		}
		return Serve(cfgFile, stype, saddr)
	}

	ctx := context.Background()
	if err := argparser.ExecuteContext(ctx); err != nil {
		fmt.Fprintf(os.Stderr, "%v: error: %v\n", argparser.CommandPath(), err)
		os.Exit(1)
	}
}

func Serve(cfgFile, stype, saddr string) (err error) {
	maybeSetErr := func(_err error) {
		if err == nil && _err != nil {
			err = _err
		}
	}

	cfg, err := leader.LoadConfig(cfgFile)
	if err != nil {
		return err
	}

	db, err := cfg.DB()
	if err != nil {
		return err
	}
	defer func() { maybeSetErr(db.Close()) }()

	sock, err := netListen(stype, saddr)
	if err != nil {
		return err
	}

	if err := db.GCRunningJobs(); err != nil {
		return err
	}

	// Even tnough most of the routing is happening in big
	// `switch` statements, the ServeMux is important because it
	// cleans the paths, which would sort of be a pain to do
	// ourselves.
	router := http.NewServeMux()
	router.Handle("/", &ServerState{db: db})
	router.Handle("/webhooks/", http.StripPrefix("/webhooks/", &Webhooks{cfgFile: cfgFile}))

	log.Printf("Serving on %v...", sock.Addr())
	return http.Serve(sock, router)
}

// HTTP utils //////////////////////////////////////////////////////////////////

func httpAddSlash(w http.ResponseWriter, r *http.Request) {
	u := &url.URL{Path: r.URL.Path + "/", RawQuery: r.URL.RawQuery}
	http.Redirect(w, r, u.String(), http.StatusMovedPermanently)
}

type _httpError struct {
	Err  error
	Code int
}

func (e *_httpError) Error() string {
	return fmt.Sprintf("HTTP %d: %v", e.Code, e.Err)
}

func httpError(err error, code int) *_httpError {
	return &_httpError{Code: code, Err: err}
}

func writeError(err error) *_httpError {
	return &_httpError{Code: -100, Err: err}
}

func writeOrServerError(err error) *_httpError {
	return &_httpError{Code: http.StatusInternalServerError, Err: err}
}

func httpNotFound() *_httpError {
	return httpError(errors.New("404 page not found"), http.StatusNotFound)
}

// ServerState /////////////////////////////////////////////////////////////////

type ServerState struct {
	db      *leader.DB
	running typedsync.Map[common.JobID, chan struct{}]
}

//go:embed style.css
var fileStyle []byte

//go:embed favicon.ico
var fileFavicon []byte

//go:embed html
var tmplFS embed.FS

var (
	tmplIndex = template.Must(template.ParseFS(tmplFS, "html/pages/index.html.tmpl", "html/lib/*.html.tmpl"))
	tmplJobs  = template.Must(template.ParseFS(tmplFS, "html/pages/jobs.html.tmpl", "html/lib/*.html.tmpl"))
	tmplJob   = template.Must(template.ParseFS(tmplFS, "html/pages/job.html.tmpl", "html/lib/*.html.tmpl"))
)

func (o *ServerState) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if err := o.serveHTTP(w, r); err != nil {
		if err.Code != -100 {
			http.Error(w, err.Error(), err.Code)
		}
		if err.Code/100 == 5 {
			log.Printf("error: %q: %v", r.URL, err)
		}
	}
}

func (o *ServerState) serveHTTP(w http.ResponseWriter, r *http.Request) *_httpError {
	switch {
	case r.URL.Path == "/":
		if r.Method != http.MethodGet {
			return httpError(errors.New("only GET is supported"), http.StatusMethodNotAllowed)
		}
		w.Header().Set("Content-Type", "text/html; charset=utf-8")
		if err := tmplIndex.Execute(w, nil); err != nil {
			return writeOrServerError(err)
		}
	case r.URL.Path == "/style.css":
		if r.Method != http.MethodGet {
			return httpError(errors.New("only GET is supported"), http.StatusMethodNotAllowed)
		}
		w.Header().Set("Content-Type", "text/css")
		if _, err := w.Write(fileStyle); err != nil {
			return writeError(err)
		}
	case r.URL.Path == "/favicon.ico":
		if r.Method != http.MethodGet {
			return httpError(errors.New("only GET is supported"), http.StatusMethodNotAllowed)
		}
		w.Header().Set("Content-Type", "image/vnd.microsoft.icon")
		if _, err := w.Write(fileFavicon); err != nil {
			return writeError(err)
		}
	case r.URL.Path == "/eclipse.tar":
		if r.Method != http.MethodGet {
			return httpError(errors.New("only GET is supported"), http.StatusMethodNotAllowed)
		}
		w.Header().Set("Content-Type", "application/x-tar")
		if _, err := w.Write(source.Tarball); err != nil {
			return writeError(err)
		}
	case r.URL.Path == "/jobs":
		httpAddSlash(w, r)
	case r.URL.Path == "/jobs/":
		switch r.Method {
		case http.MethodGet:
			params := r.URL.Query()
			status := common.JobStatus(-1)
			_ = status.UnmarshalText([]byte(params.Get("status"))) // allowed to fail
			wait, _ := strconv.ParseBool(params.Get("wait"))       // allowed to fail
			log.Printf("status=%v wait=%v", status, wait)

			var jobs []common.Job
			for i := 0; i == 0 || (wait && len(jobs) == 0); i++ {
				var err error
				jobs, err = o.db.ListJobs(status)
				if err != nil {
					return httpError(err, http.StatusInternalServerError)
				}
				if wait && len(jobs) == 0 {
					if i%30 == 0 {
						// Send a keepalive, every 30-seconds-ish (nginx's
						// default HTTP 504 timeout is 60s, we could say
						// that you need to disable it, but this feels
						// butter).
						w.WriteHeader(http.StatusContinue)
					}
					time.Sleep(1 * time.Second)
				}
			}
			switch httputil.NegotiateContentType(r, []string{"application/json", "text/html"}, "application/json") {
			case "text/html":
				w.Header().Set("Content-Type", "text/html; charset=utf-8")
				if err := tmplJobs.Execute(w, map[string]any{"Jobs": jobs}); err != nil {
					return writeOrServerError(err)
				}
			case "application/json":
				jsonBytes, err := json.Marshal(jobs)
				if err != nil {
					return httpError(err, http.StatusInternalServerError)
				}
				w.Header().Set("Content-Type", "application/json")
				if _, err := w.Write(jsonBytes); err != nil {
					return writeError(err)
				}
			default:
				return httpError(errors.New("only text/html and application/json are supported"), http.StatusNotAcceptable)
			}
		case http.MethodPost:
			return httpError(errors.New("TODO: allow POSTing to /jobs/ to create a job"), http.StatusNotImplemented)
		default:
			return httpError(errors.New("only GET and POST are supported"), http.StatusMethodNotAllowed)
		}
	case strings.HasPrefix(r.URL.Path, "/jobs/"):
		rest := strings.TrimPrefix(r.URL.Path, "/jobs/")
		jobIDStr, file, haveSlash := strings.Cut(rest, "/")
		jobID, err := strconv.Atoi(jobIDStr)
		if err != nil {
			return httpNotFound()
		}
		job, err := o.db.GetJob(common.JobID(jobID))
		if err != nil {
			return httpNotFound()
		}
		if !haveSlash {
			httpAddSlash(w, r)
			return nil
		}
		switch file {
		case "":
			switch r.Method {
			case http.MethodGet:
				dirfs := job.DirFS()
				artifactNames := []string{}
				if err := fs.WalkDir(dirfs, "artifacts", func(path string, d fs.DirEntry, err error) error {
					if err != nil {
						return err
					}
					if d.Type() == 0 {
						artifactNames = append(artifactNames, strings.TrimPrefix(path, "artifacts/"))
					}
					return nil
				}); err != nil && !os.IsNotExist(err) {
					return httpError(err, http.StatusInternalServerError)
				}
				w.Header().Set("Content-Type", "text/html; charset=utf-8")
				if err := tmplJob.Execute(w, map[string]any{
					"Job":           job,
					"ArtifactNames": artifactNames,
				}); err != nil {
					return writeOrServerError(err)
				}
			case http.MethodPost:
				// TODO: Validate API key
				return o.runJob(w, r, job)
			default:
				return httpError(errors.New("only GET and POST are supported"), http.StatusMethodNotAllowed)
			}
		default:
			if r.Method != http.MethodGet {
				return httpError(errors.New("only GET is supported"), http.StatusMethodNotAllowed)
			}
			if file == "log.txt" && job.Status == common.StatusRunning {
				ch, _ := o.running.Load(job.ID)
				w.Header().Set("Content-Type", "text/plain; charset=utf-8")
				if err := tailf(w, ch, filepath.Join(job.Dir, "log.txt")); err != nil {
					return writeOrServerError(err)
				}
			} else {
				r.URL.Path = "/" + file
				http.FileServer(http.Dir(job.Dir)).ServeHTTP(w, r)
			}
		}
	default:
		return httpNotFound()
	}
	return nil
}

// tailf is like the `tail -f --retry -n +1`, but a Go function.
func tailf(w http.ResponseWriter, cancel <-chan struct{}, filename string) error {
	ctl := http.NewResponseController(w)

	fh, _ := os.Open(filename)
	if fh != nil {
		if _, err := io.Copy(w, fh); err != nil {
			return err
		}
		if err := ctl.Flush(); err != nil {
			return err
		}
	}

	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()
	for keepGoing := true; keepGoing; {
		select {
		case <-cancel:
			keepGoing = false
		case <-ticker.C:
		}
		if fh == nil {
			fh, _ = os.Open(filename)
		}
		if fh != nil {
			if _, err := io.Copy(w, fh); err != nil {
				return err
			}
			if err := ctl.Flush(); err != nil {
				return err
			}
		}
	}

	return nil
}

func (o *ServerState) runJob(w http.ResponseWriter, r *http.Request, job common.Job) (_reterr *_httpError) {
	log.Printf("running job: %q", r.URL)

	ctl := http.NewResponseController(w)
	if err := ctl.EnableFullDuplex(); err != nil {
		return httpError(err, http.StatusInternalServerError)
	}

	ch := make(chan struct{})
	if _, conflict := o.running.LoadOrStore(job.ID, ch); conflict {
		return httpError(errors.New("job has already started (ch)"), http.StatusConflict)
	}
	defer func() {
		close(ch)
		o.running.Delete(job.ID)
	}()
	swapped, err := o.db.SwapJobStatus(job.ID, common.StatusNew, common.StatusRunning, "")
	if err != nil {
		return httpError(err, http.StatusInternalServerError)
	}
	if !swapped {
		return httpError(errors.New("job has already started (db)"), http.StatusConflict)
	}
	defer func() {
		if _reterr != nil {
			if _, err := o.db.SwapJobStatus(job.ID, common.StatusRunning, common.StatusError, _reterr.Err.Error()); err != nil {
				_reterr = httpError(err, http.StatusInternalServerError)
			}
		}
	}()

	bs, err := json.Marshal(job)
	if err != nil {
		return httpError(fmt.Errorf("encoding job description: %w", err), http.StatusInternalServerError)
	}
	mediaType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
	if err != nil {
		return httpError(fmt.Errorf("invalid Content-Type: %q: %w", r.Header.Get("Content-Type"), err), http.StatusBadRequest)
	}
	if mediaType != "multipart/mixed" {
		return httpError(fmt.Errorf("invalid Content-Type: %q", r.Header.Get("Content-Type")), http.StatusBadRequest)
	}
	reader := multipart.NewReader(r.Body, params["boundary"])

	if strings.Contains(r.Header.Get("Expect"), "100-continue") {
		// The stdlib net/http server is supposed to automatically do this if needed, but it
		// triggers this based on us reading the request body.  Normally that would be a
		// good trigger; usually you *must* have read the body before you start writing the
		// response... but ctl.EnableFullDuplex() broke that assumption.  If we start
		// writing before we read, the net/http server would skip straight to the "HTTP 200"
		// response, skipping the require "HTTP 100" response.  (as of Go 1.21.6)
		//
		// So manually emit the "HTTP 100" response if it looks like we need it.  Our
		// strings.Contains() check is a little loose--it's OK for is to emit this even if
		// it wasn't asked for.
		w.WriteHeader(http.StatusContinue)
	}

	w.Header().Set("Content-Type", "application/json")

	// This is the line between "can return a status code" and "can't" ///////////////

	if _, err := w.Write(bs); err != nil {
		return writeError(fmt.Errorf("write job description: %w", err))
	}
	if err := ctl.Flush(); err != nil {
		return writeError(fmt.Errorf("write job description: %w", err))
	}

	var status string
	for {
		part, err := reader.NextPart()
		if err != nil {
			if err == io.EOF {
				break
			}
			return httpError(fmt.Errorf("read next part: %w", err), http.StatusBadRequest)
		}
		log.Printf("part: %q", part.Header.Get("Content-Disposition"))
		disposition, params, err := mime.ParseMediaType(part.Header.Get("Content-Disposition"))
		if err != nil {
			return httpError(fmt.Errorf("invalid Content-Disposition: %q: %w", part.Header.Get("Content-Disposition"), err), http.StatusBadRequest)
		}
		filename := path.Clean(params["filename"])
		switch {
		case disposition == "attachment" && filename == "log.txt" || strings.HasPrefix(filename, "artifacts/"):
			fsFilename := filepath.Join(job.Dir, filepath.FromSlash(filename))
			if err := os.MkdirAll(filepath.Dir(fsFilename), 0755); err != nil {
				return httpError(fmt.Errorf("saving file: %q: %w", filename, err), http.StatusInternalServerError)
			}
			fh, err := os.Create(fsFilename)
			if err != nil {
				return httpError(fmt.Errorf("saving file: %q: %w", filename, err), http.StatusInternalServerError)
			}
			if _, err := io.Copy(fh, part); err != nil {
				return writeOrServerError(fmt.Errorf("saving file: %q: %w", filename, err))
			}
			if err := fh.Close(); err != nil {
				return httpError(fmt.Errorf("saving file: %q: %w", filename, err), http.StatusInternalServerError)
			}
		case disposition == "form-data" && params["name"] == "status":
			statusBytes, err := io.ReadAll(part)
			if err != nil {
				return httpError(fmt.Errorf("reading status: %w (got: %q)", err, statusBytes), http.StatusInternalServerError)
			}
			status = strings.TrimSpace(string(statusBytes))
		default:
			return httpError(fmt.Errorf("invalid Content-Disposition: %q", part.Header.Get("Content-Disposition")), http.StatusBadRequest)
		}
		if err := part.Close(); err != nil {
			return httpError(fmt.Errorf("closing part: %w", err), http.StatusBadRequest)
		}
	}

	switch status {
	case "0":
		if _, err := o.db.SwapJobStatus(job.ID, common.StatusRunning, common.StatusSucceeded, ""); err != nil {
			return httpError(err, http.StatusInternalServerError)
		}
	case "":
		return httpError(fmt.Errorf("no status: %w", io.ErrUnexpectedEOF), http.StatusBadRequest)
	default:
		if _, err := o.db.SwapJobStatus(job.ID, common.StatusRunning, common.StatusFailed, status); err != nil {
			return httpError(err, http.StatusInternalServerError)
		}
	}

	return nil
}