intake/core/execute.go

package core

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"os"
	"os/exec"
	"strings"
	"time"
)

func readPipe(
	f io.ReadCloser,
	send chan []byte,
	done chan int,
) {
	defer func() {
		done <- 0
	}()

	var data []byte
	var err error
	reader := bufio.NewReader(f)
	for data, err = reader.ReadBytes('\n'); err == nil; data, err = reader.ReadBytes('\n') {
		send <- data
	}
	if err != io.EOF {
		log.Printf("error: failed to read pipe: %v", err)
	}
	// In case the last line has no newline
	if len(data) > 0 {
		send <- data
	}
}

// Two-stage error item helper, one stage to capture the immutable parameters and a second
// to include the variable error and logs.
func makeMakeErrItem(source string, argv []string) func(err error, logs []string) (item Item) {
	return func(err error, logs []string) (item Item) {
		item.Source = "default"
		item.Id = RandomHex(16)
		item.Title = fmt.Sprintf("Failed to execute for %s", source)
		log := "no logs"
		if len(logs) > 0 {
			log = strings.Join(logs, "\n")
		}
		item.Body = fmt.Sprintf("<p><i>executing:</i> %#v</p><p><i>error:</i> %s</p><pre>%s</pre>", argv, err.Error(), log)
		return
	}
}

func Execute(
	source string,
	argv []string,
	env []string,
	state []byte,
	input string,
	timeout time.Duration,
	postProcess func(item Item) Item,
) (
	items []Item,
	newState []byte,
	errItem Item,
	err error,
) {
	log.Printf("executing %v", argv)

	makeErrorItem := makeMakeErrItem(source, argv)
	var logs []string
	addLog := func(format string, args ...any) {
		msg := fmt.Sprintf(format, args...)
		log.Print(msg)
		logs = append(logs, msg)
	}

	if source == "" {
		err = fmt.Errorf("empty source")
		errItem = makeErrorItem(err, logs)
		return
	}
	if len(argv) == 0 {
		err = fmt.Errorf("empty argv for %s", source)
		errItem = makeErrorItem(err, logs)
		return
	}

	stateFile, err := os.CreateTemp("", "intake_state_*")
	if err != nil {
		err = fmt.Errorf("failed to create temp state file: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}
	defer func() {
		if err := os.Remove(stateFile.Name()); err != nil {
			log.Printf("error: failed to delete %s", stateFile.Name())
		}
	}()

	_, err = stateFile.Write(state)
	if err != nil {
		err = fmt.Errorf("failed to write state file: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}

	env = append(env, "STATE_PATH="+stateFile.Name())

	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	defer cancel()
	cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
	cmd.Env = append(os.Environ(), env...)
	cmd.WaitDelay = time.Second * 5

	// Open pipes to the command
	stdin, err := cmd.StdinPipe()
	if err != nil {
		err = fmt.Errorf("failed to open stdin pipe: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		err = fmt.Errorf("failed to open stdout pipe: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		err = fmt.Errorf("failed to open stderr pipe: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}

	cout := make(chan []byte)
	cerr := make(chan []byte)
	coutDone := make(chan int)
	cerrDone := make(chan int)

	// Routines handling the process i/o
	go readPipe(stdout, cout, coutDone)
	go readPipe(stderr, cerr, cerrDone)

	// Kick off the command
	err = cmd.Start()
	if err != nil {
		err = fmt.Errorf("failed to start execution: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}

	// Write any input to stdin and close it
	if _, err = io.WriteString(stdin, input); err != nil {
		err = fmt.Errorf("failed to write to stdin: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}
	if err = stdin.Close(); err != nil {
		err = fmt.Errorf("failed to close stdin: %v", err)
		errItem = makeErrorItem(err, logs)
		return
	}

	// Collect outputs until std{out,err} close
	parseError := false
	stdoutDone := false
	stderrDone := false
	duplicateItem := ""
	itemIds := make(map[string]bool)
monitor:
	for {
		select {
		case data := <-cout:
			var item Item
			err := json.Unmarshal(data, &item)
			if err != nil || item.Id == "" {
				if err != nil {
					addLog("[%s:   item] parse error: %v", source, err)
				}
				addLog("[%s: stdout] %s", source, strings.TrimSpace(string(data)))
				parseError = true
			} else {
				if itemIds[item.Id] {
					addLog("[%s:   item] %s (duplicate)", source, item.Id)
					duplicateItem = item.Id
					cmd.Cancel()
					break monitor
				}
				itemIds[item.Id] = true

				if postProcess != nil {
					item = postProcess(item)
				}
				item.Active = true   // These fields aren't up to
				item.Created = 0     // the action to set and
				item.Source = source // shouldn't be overrideable
				addLog("[%s:   item] %s", source, item.Id)
				items = append(items, item)
			}

		case data := <-cerr:
			addLog("[%s: stderr] %s", source, strings.TrimSpace(string(data)))

		case <-coutDone:
			stdoutDone = true
			if stdoutDone && stderrDone {
				break monitor
			}

		case <-cerrDone:
			stderrDone = true
			if stdoutDone && stderrDone {
				break monitor
			}
		}
	}

	err = cmd.Wait()
	if duplicateItem != "" {
		err = fmt.Errorf("returned item %s twice: %v", duplicateItem, err)
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	} else if ctx.Err() == context.DeadlineExceeded {
		err = fmt.Errorf("timed out after %v", timeout)
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	} else if exiterr, ok := err.(*exec.ExitError); ok {
		err = fmt.Errorf("%s failed with exit code %d: %v", argv[0], exiterr.ExitCode(), exiterr)
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	} else if err != nil {
		err = fmt.Errorf("error: %s failed with error: %s", argv[0], err)
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	}

	if parseError {
		err = fmt.Errorf("could not parse an item")
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	}

	newState, err = os.ReadFile(stateFile.Name())
	if err != nil {
		err = fmt.Errorf("error: failed to read state file: %v", err)
		errItem = makeErrorItem(err, logs)
		log.Printf("error: %v", err)
		return nil, nil, errItem, err
	}

	return
}

// Execute an action that takes an item as input and returns the item modified.
// This is basically just a wrapper over [Execute] that handles the input and backfilling.
func ExecuteItemAction(
	item Item,
	argv []string,
	env []string,
	state []byte,
	timeout time.Duration,
	postProcess func(item Item) Item,
) (
	newItem Item,
	newState []byte,
	errItem Item,
	err error,
) {
	makeErrorItem := makeMakeErrItem(item.Source, argv)

	itemJson, err := json.Marshal(item)
	if err != nil {
		err = fmt.Errorf("failed to serialize item: %v", err)
		errItem = makeErrorItem(err, nil)
		return
	}

	res, newState, errItem, err := Execute(item.Source, argv, env, state, string(itemJson), timeout, postProcess)
	if err != nil {
		err = fmt.Errorf("failed to execute action for %s/%s: %v", item.Source, item.Id, err)
		errItem = makeErrorItem(err, nil)
		return
	}
	if len(res) != 1 {
		err = fmt.Errorf("expected action to produce exactly one item, got %d", len(res))
		errItem = makeErrorItem(err, nil)
		return
	}
	newItem = res[0]
	BackfillItem(&newItem, &item)

	return
}