// gitbay-runner executes CI builds queued by a gitbay server. It polls over
// SSH — the same authenticated channel everything else uses — claims one
// build at a time, clones the repo, runs each step with `sh -c`, streams the
// combined output back, and reports success or failure.
//
// The account behind the runner's key must be an instance admin: a runner
// executes arbitrary repo code, so handing out jobs is the operator's call.
// v1 runs steps directly on the host under this process's user; run it as a
// dedicated unprivileged user.
package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"gitbay.org/gitbay/internal/buildinfo"
	"gitbay.org/gitbay/internal/toolpath"
)

type job struct {
	ID      int64             `json:"id"`
	Repo    string            `json:"repo"`
	Number  int64             `json:"number"`
	Job     string            `json:"job"`
	SHA     string            `json:"sha"`
	Ref     string            `json:"ref"`
	Steps   []string          `json:"steps"`
	Secrets map[string]string `json:"secrets"`
}

type runner struct {
	remote    string // ssh destination, e.g. git@gitbay.org
	sshOpts   []string
	cloneBase string // e.g. ssh://git@gitbay.org
	workdir   string
	timeout   time.Duration
	// repos limits which repositories this runner claims builds for. Empty
	// means any, which is what a runner on the server itself wants; a runner
	// somewhere that should not execute every repository's steps names them.
	repos []string
}

func main() {
	var (
		remote    = flag.String("remote", "git@gitbay.org", "ssh destination of the gitbay server")
		sshOpts   = flag.String("ssh-opts", "", "extra ssh options, space-separated (also used for git clone)")
		cloneBase = flag.String("clone-base", "", "clone URL prefix (default ssh://<remote>)")
		workdir   = flag.String("workdir", defaultWorkdir(), "build workspace root")
		poll      = flag.Duration("poll", 5*time.Second, "idle poll interval")
		timeout   = flag.Duration("timeout", 30*time.Minute, "per-build time limit")
		repos     = flag.String("repos", "", "only claim builds for these repositories, comma-separated owner/name (default: any)")
		once      = flag.Bool("once", false, "process at most one build, then exit")
		jobs      = flag.Int("jobs", 1, "builds to run at once")
		version   = flag.Bool("version", false, "print the commit this binary was built from, then exit")
	)
	flag.Parse()
	if *version {
		fmt.Println(buildinfo.String())
		return
	}
	// The runner links internal/store, so it goes stale on changes that never
	// touch cmd/gitbay-runner. Say which commit is running.
	log.Printf("gitbay-runner %s", buildinfo.String())
	r := &runner{
		remote:    *remote,
		cloneBase: *cloneBase,
		workdir:   *workdir,
		timeout:   *timeout,
	}
	if *sshOpts != "" {
		r.sshOpts = strings.Fields(*sshOpts)
	}
	for _, name := range strings.Split(*repos, ",") {
		if name = strings.TrimSpace(name); name != "" {
			r.repos = append(r.repos, name)
		}
	}
	if r.cloneBase == "" {
		r.cloneBase = "ssh://" + *remote
	}
	// 0o700, not 0o755: a build's checkout and its secrets-bearing
	// environment are this user's business alone, and the default sits
	// beside other users' data on a shared host.
	if err := os.MkdirAll(r.workdir, 0o700); err != nil {
		log.Fatal(err)
	}
	if err := checkWorkdir(r.workdir); err != nil {
		log.Fatal(err)
	}
	n := *jobs
	if n < 1 {
		log.Fatal("-jobs must be at least 1")
	}
	if *once {
		// "at most one build" is one build, whatever -jobs says.
		n = 1
	}
	// `runner next` claims inside one transaction, so several workers
	// claiming at once is already safe; the runner just never used that.
	// Each build works in its own build-<id> directory, so they do not
	// meet on disk either.
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			// Spread the idle polls across the interval rather than
			// having every worker wake together: n workers asking the
			// same question in the same instant is n times the load for
			// one answer.
			if n > 1 {
				time.Sleep(time.Duration(i) * *poll / time.Duration(n))
			}
			for {
				ran, err := r.step()
				if err != nil {
					log.Printf("runner: %v", err)
				}
				if *once {
					return
				}
				if !ran {
					time.Sleep(*poll)
				}
			}
		}(i)
	}
	wg.Wait()
}

// step claims and executes at most one build. ran reports whether there was
// one, so the caller knows when to idle.
func (r *runner) step() (bool, error) {
	out, err := r.ssh(nil, append([]string{"runner", "next"}, append(r.repos, "--json")...)...)
	if err != nil {
		return false, fmt.Errorf("claiming build: %w (%s)", err, out)
	}
	var env struct {
		Data job `json:"data"`
	}
	if err := json.Unmarshal([]byte(out), &env); err != nil {
		return false, fmt.Errorf("parsing job: %w", err)
	}
	if env.Data.ID == 0 {
		return false, nil
	}
	j := env.Data
	log.Printf("build %d: %s %s @ %.10s", j.ID, j.Repo, j.Job, j.SHA)
	status := "failure"
	if r.run(j) {
		status = "success"
	}
	if out, err := r.ssh(nil, "runner", "done", fmt.Sprint(j.ID), status); err != nil {
		return true, fmt.Errorf("reporting build %d: %w (%s)", j.ID, err, out)
	}
	log.Printf("build %d: %s", j.ID, status)
	return true, nil
}

// logSink forwards a build's output to the server and swallows any error
// doing so. os/exec surfaces a write failure on a step's stdout through
// cmd.Wait(), so a sink that can fail is a sink that can fail the build it
// was only recording — a restart or a dropped session used to turn a green
// suite red, with the explaining line written to the same dead pipe. Losing
// log lines is the acceptable failure here; losing the build is not.
type logSink struct {
	mu sync.Mutex
	w  io.Writer // nil once a write has failed
}

func (s *logSink) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.w != nil {
		if _, err := s.w.Write(p); err != nil {
			s.w = nil
		}
	}
	return len(p), nil
}

// broken reports whether the stream was lost, so a build can say its log is
// incomplete rather than appear to have simply stopped.
func (s *logSink) broken() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.w == nil
}

// run clones, checks out, and executes the steps, streaming output to the
// server. Returns whether every step succeeded.
func (r *runner) run(j job) bool {
	dir := filepath.Join(r.workdir, fmt.Sprintf("build-%d", j.ID))
	defer os.RemoveAll(dir)

	// One long-lived `runner log` session receives the whole stream.
	logCmd := exec.Command(toolpath.Look("ssh"), append(r.sshOpts, r.remote, "runner", "log", fmt.Sprint(j.ID))...)
	pipe, err := logCmd.StdinPipe()
	if err != nil {
		log.Printf("build %d: log pipe: %v", j.ID, err)
		return false
	}
	sink := &logSink{w: pipe}
	logCmd.Stdout, logCmd.Stderr = io.Discard, io.Discard
	if err := logCmd.Start(); err != nil {
		log.Printf("build %d: log stream: %v", j.ID, err)
		return false
	}
	// The server ends the log session with exit 3 when the build is
	// cancelled; any other end is a lost stream, which the sink absorbs.
	cancelled := make(chan struct{})
	logExited := make(chan struct{})
	go func() {
		defer close(logExited)
		err := logCmd.Wait()
		if ee, ok := err.(*exec.ExitError); ok && ee.ExitCode() == 3 {
			close(cancelled)
			return
		}
		if err != nil {
			log.Printf("build %d: log session ended: %v", j.ID, err)
		}
	}()
	// runStep starts cmd and waits for it, the cancel signal, or the
	// deadline. Every phase goes through it, so a cancel during the clone
	// lands as fast as one during a step.
	runStep := func(cmd *exec.Cmd, deadline time.Time) (bool, string) {
		select {
		case <-cancelled:
			return false, "cancelled"
		default:
		}
		ownProcessGroup(cmd)
		if err := cmd.Start(); err != nil {
			return false, fmt.Sprintf("start: %v", err)
		}
		done := make(chan error, 1)
		go func() { done <- cmd.Wait() }()
		// After a kill, Wait returns once every holder of the log pipe is
		// gone; the group kill makes that prompt, and the cap makes sure a
		// straggler cannot hold the build open.
		reap := func() {
			killTree(cmd)
			select {
			case <-done:
			case <-time.After(10 * time.Second):
			}
		}
		select {
		case err := <-done:
			if err != nil {
				return false, fmt.Sprintf("step failed: %v", err)
			}
			return true, ""
		case <-cancelled:
			reap()
			return false, "cancelled"
		case <-time.After(time.Until(deadline)):
			reap()
			return false, fmt.Sprintf("build timed out after %s", r.timeout)
		}
	}
	defer func() {
		select {
		case <-cancelled:
			log.Printf("build %d: cancelled", j.ID)
		default:
			if sink.broken() {
				log.Printf("build %d: log stream lost; stored log is incomplete", j.ID)
			}
		}
		pipe.Close()
		<-logExited
	}()

	gitSSH := strings.TrimSpace("ssh " + strings.Join(r.sshOpts, " "))
	cloneURL := r.cloneBase + "/" + j.Repo + ".git"
	deadline := time.Now().Add(r.timeout)
	fmt.Fprintf(sink, "$ git clone %s (%.10s)\n", cloneURL, j.SHA)
	// A merge request head lives under refs/merge-requests/, which a
	// clone does not fetch; ask for the ref before checking out.
	steps := [][]string{{"clone", "-q", cloneURL, dir}}
	if strings.HasPrefix(j.Ref, "refs/") {
		steps = append(steps, []string{"-C", dir, "fetch", "-q", "origin", j.Ref})
	}
	steps = append(steps, []string{"-C", dir, "checkout", "-q", j.SHA})
	for _, args := range steps {
		cmd := exec.Command(toolpath.Look("git"), args...)
		cmd.Env = append(os.Environ(), "GIT_SSH_COMMAND="+gitSSH, "GIT_TERMINAL_PROMPT=0")
		cmd.Stdout, cmd.Stderr = sink, sink
		if ok, why := runStep(cmd, deadline); !ok {
			fmt.Fprintf(sink, "git %s: %s\n", args[0], why)
			return false
		}
	}

	for _, step := range j.Steps {
		fmt.Fprintf(sink, "$ %s\n", step)
		cmd := exec.Command(toolpath.Look("sh"), "-c", step)
		cmd.Dir = dir
		cmd.Env = append(os.Environ(),
			"GITBAY_REPO="+j.Repo, "GITBAY_SHA="+j.SHA, "GITBAY_REF="+j.Ref, "GITBAY_JOB="+j.Job, "CI=true")
		for name, value := range j.Secrets {
			cmd.Env = append(cmd.Env, name+"="+value)
		}
		cmd.Stdout, cmd.Stderr = sink, sink
		if ok, why := runStep(cmd, deadline); !ok {
			fmt.Fprintf(sink, "%s\n", why)
			return false
		}
	}
	return true
}

// ssh runs one control command against the server and returns stdout.
func (r *runner) ssh(stdin io.Reader, args ...string) (string, error) {
	cmd := exec.Command(toolpath.Look("ssh"), append(append(r.sshOpts, r.remote), args...)...)
	if stdin != nil {
		cmd.Stdin = stdin
	}
	var out, errOut strings.Builder
	cmd.Stdout, cmd.Stderr = &out, &errOut
	if err := cmd.Run(); err != nil {
		return out.String() + errOut.String(), err
	}
	return out.String(), nil
}

// defaultWorkdir picks a build workspace that another local user cannot
// have created first.
//
// The default used to be <tmp>/gitbay-runner: a fixed name inside a
// world-writable directory, created with MkdirAll, which succeeds against
// an existing directory whoever owns it. On a shared host another user
// could have made it — or symlinked it — before the runner started, and
// this is the process that clones repositories and exports build secrets
// into step environments (go:S5445, #153).
//
// The user's cache directory is not world-writable and is per-user by
// construction. Falling back to tmp keeps a runner working where HOME is
// unset, and checkWorkdir refuses the unsafe cases there.
func defaultWorkdir() string {
	if cache, err := os.UserCacheDir(); err == nil && cache != "" {
		return filepath.Join(cache, "gitbay-runner")
	}
	return filepath.Join(os.TempDir(), "gitbay-runner")
}

// checkWorkdir makes sure the workspace is a directory this user owns
// privately. MkdirAll is happy with one that already exists, so being
// able to create it proves nothing about who made it.
//
// A directory we own that is merely too permissive is tightened rather
// than refused: every runner before this one created its workspace 0755,
// so refusing would take the runner down on upgrade to fix a permission
// we are entitled to change. What cannot be repaired — a symlink, or
// something owned by someone else — is refused, because those are what an
// attacker leaves behind and neither is ours to correct.
func checkWorkdir(dir string) error {
	fi, err := os.Lstat(dir)
	if err != nil {
		return err
	}
	if fi.Mode()&os.ModeSymlink != 0 {
		return fmt.Errorf("workdir %s is a symlink; point -workdir at a real directory", dir)
	}
	if !fi.IsDir() {
		return fmt.Errorf("workdir %s is not a directory", dir)
	}
	if st, ok := fi.Sys().(*syscall.Stat_t); ok && int(st.Uid) != os.Getuid() {
		return fmt.Errorf("workdir %s is owned by uid %d, not this process's %d", dir, st.Uid, os.Getuid())
	}
	if perm := fi.Mode().Perm(); perm&0o077 != 0 {
		log.Printf("workdir %s was mode %04o; tightening to 0700 (builds and their secrets are this user's alone)", dir, perm)
		if err := os.Chmod(dir, 0o700); err != nil {
			return fmt.Errorf("tightening workdir %s: %w", dir, err)
		}
	}
	return nil
}