OliveTin/service/internal/executor/executor.go

package executor

import (
	acl "github.com/OliveTin/OliveTin/internal/acl"
	"github.com/OliveTin/OliveTin/internal/auth"
	authpublic "github.com/OliveTin/OliveTin/internal/auth/authpublic"
	config "github.com/OliveTin/OliveTin/internal/config"
	"github.com/OliveTin/OliveTin/internal/entities"
	"github.com/OliveTin/OliveTin/internal/logfilter"
	"github.com/OliveTin/OliveTin/internal/tpl"
	"github.com/google/uuid"
	log "github.com/sirupsen/logrus"

	"gopkg.in/yaml.v3"

	"bytes"
	"context"
	"errors"
	"fmt"
	"maps"
	"os"
	"os/exec"
	"path"
	"regexp"
	"slices"
	"strings"
	"sync"
	"time"
)

const (
	DefaultExitCodeNotExecuted = -1337
	MaxTriggerDepth            = 10
)

var validTrackingIDPattern = regexp.MustCompile(`^[a-fA-F0-9\-]+$`)

func isValidTrackingID(id string) bool {
	const MaxTrackingIDLength = 36

	return id != "" && len(id) <= MaxTrackingIDLength && validTrackingIDPattern.MatchString(id)
}

type ActionBinding struct {
	Action       *config.Action
	Entity       *entities.Entity
	ID           string
	OnDashboards []DashboardNavigationTarget
	ConfigOrder  int
}

type Executor struct {
	logs                  map[string]*InternalLogEntry
	LogsByBindingId       map[string][]*InternalLogEntry
	MapActionBindings     map[string]*ActionBinding
	Cfg                   *config.Config
	logsTrackingIdsByDate []string
	listeners             []listener
	chainOfCommand        []executorStepFunc
	groupQueue            []*queuedExecution
	logmutex              sync.RWMutex
	MapActionBindingsLock sync.RWMutex
	listenersMu           sync.RWMutex
	groupQueueMu          sync.Mutex
}

// ExecutionRequest is a request to execute an action. It's passed to an
// Executor. They're created from the api.
type ExecutionRequest struct {
	Arguments               map[string]string
	Binding                 *ActionBinding
	Cfg                     *config.Config
	AuthenticatedUser       *authpublic.AuthenticatedUser
	executor                *Executor
	logEntry                *InternalLogEntry
	finalParsedCommand      string
	TrackingID              string
	Justification           string
	Tags                    []string
	execArgs                []string
	TriggerDepth            int
	useDirectExec           bool
	skipRequestRegistration bool
}

func (req *ExecutionRequest) mutateLogEntry(mutator func(*InternalLogEntry)) {
	if req.executor == nil {
		mutator(req.logEntry)
		return
	}

	req.executor.logmutex.Lock()
	defer req.executor.logmutex.Unlock()

	mutator(req.logEntry)
}

// LogEntrySnapshot is a copy of selected log entry fields for race-safe reads.
type LogEntrySnapshot struct {
	Output            string
	ExitCode          int32
	Queued            bool
	Blocked           bool
	ExecutionStarted  bool
	ExecutionFinished bool
}

// SnapshotLog returns a copy of selected log entry fields under read lock.
func (e *Executor) SnapshotLog(trackingID string) (LogEntrySnapshot, bool) {
	e.logmutex.RLock()
	defer e.logmutex.RUnlock()

	entry, found := e.logs[trackingID]
	if !found {
		return LogEntrySnapshot{}, false
	}

	return LogEntrySnapshot{
		Queued:            entry.Queued,
		Blocked:           entry.Blocked,
		ExecutionStarted:  entry.ExecutionStarted,
		ExecutionFinished: entry.ExecutionFinished,
		ExitCode:          entry.ExitCode,
		Output:            entry.Output,
	}, true
}

// InternalLogEntry objects are created by an Executor, and represent the final
// state of execution (even if the command is not executed). It's designed to be
// easily serializable.
type InternalLogEntry struct {
	DatetimeStarted     time.Time
	DatetimeFinished    time.Time
	Binding             *ActionBinding
	Process             *os.Process
	Arguments           map[string]string
	ExecutionTrackingID string
	Justification       string
	QueuedForGroup      string
	ActionIcon          string
	ActionTitle         string
	ActionConfigTitle   string
	Output              string
	Username            string
	EntityPrefix        string
	Tags                []string
	Index               int64
	ExitCode            int32
	Blocked             bool
	ExecutionFinished   bool
	ExecutionStarted    bool
	Queued              bool
	TimedOut            bool
}

func cloneActionBinding(binding *ActionBinding) *ActionBinding {
	if binding == nil {
		return nil
	}

	cloned := *binding
	cloned.OnDashboards = slices.Clone(binding.OnDashboards)

	return &cloned
}

func cloneInternalLogEntry(entry *InternalLogEntry) *InternalLogEntry {
	if entry == nil {
		return nil
	}

	cloned := *entry
	cloned.Arguments = maps.Clone(entry.Arguments)
	cloned.Tags = slices.Clone(entry.Tags)
	cloned.Binding = cloneActionBinding(entry.Binding)

	return &cloned
}

// .Binding can be nil, so we need to handle that.
func (e *InternalLogEntry) GetBindingId() string {
	if e.Binding == nil {
		return ""
	}

	return e.Binding.ID
}

type executorStepFunc func(*ExecutionRequest) bool

// DefaultExecutor returns an Executor, with a sensible "chain of command" for
// executing actions.
func DefaultExecutor(cfg *config.Config) *Executor {
	e := Executor{}
	e.Cfg = cfg
	e.logs = make(map[string]*InternalLogEntry)
	e.logsTrackingIdsByDate = make([]string, 0)
	e.LogsByBindingId = make(map[string][]*InternalLogEntry)
	e.MapActionBindings = make(map[string]*ActionBinding)

	e.chainOfCommand = []executorStepFunc{
		stepRequestAction,
		stepConcurrencyCheck,
		stepRateCheck,
		stepACLCheck,
		stepParseArgs,
		stepLogStart,
		stepExec,
		stepExecAfter,
		stepLogFinish,
		stepTrigger,
	}

	return &e
}

type listener interface {
	OnExecutionStarted(logEntry *InternalLogEntry)
	OnExecutionFinished(logEntry *InternalLogEntry)
	OnOutputChunk(o []byte, executionTrackingId string)
	OnActionMapRebuilt()
}

func (e *Executor) AddListener(m listener) {
	e.listenersMu.Lock()
	defer e.listenersMu.Unlock()
	e.listeners = append(e.listeners, m)
}

func (e *Executor) copyListeners() []listener {
	e.listenersMu.RLock()
	defer e.listenersMu.RUnlock()
	out := make([]listener, len(e.listeners))
	copy(out, e.listeners)
	return out
}

// getPagingStartIndex calculates the starting index for log pagination.
// Parameters:
//
//	startOffset: The offset from the most recent log (0 means start from the most recent)
//	totalLogCount: Total number of logs available
//	count: Number of logs to retrieve
//
// Returns: The calculated starting index for pagination
func getPagingStartIndex(startOffset int64, totalLogCount int64) int64 {
	var startIndex int64

	if startOffset <= 0 {
		startIndex = totalLogCount
	} else {
		startIndex = (totalLogCount - startOffset)

		if startIndex < 0 {
			startIndex = 1
		}
	}

	return startIndex - 1
}

type PagingResult struct {
	CountRemaining int64
	PageSize       int64
	TotalCount     int64
	StartOffset    int64
}

func (e *Executor) GetLogTrackingIds(startOffset int64, pageCount int64) ([]*InternalLogEntry, *PagingResult) {
	pagingResult := &PagingResult{
		CountRemaining: 0,
		PageSize:       pageCount,
		TotalCount:     0,
		StartOffset:    startOffset,
	}

	e.logmutex.RLock()

	totalLogCount := int64(len(e.logsTrackingIdsByDate))
	pagingResult.TotalCount = totalLogCount

	startIndex := getPagingStartIndex(startOffset, totalLogCount)

	pageCount = min(totalLogCount, pageCount)

	endIndex := max(0, (startIndex-pageCount)+1)

	log.WithFields(log.Fields{
		"startOffset": startOffset,
		"pageCount":   pageCount,
		"total":       totalLogCount,
		"startIndex":  startIndex,
		"endIndex":    endIndex,
	}).Tracef("GetLogTrackingIds")

	trackingIds := make([]*InternalLogEntry, 0, pageCount)

	if totalLogCount > 0 {
		for i := startIndex; i >= endIndex; i-- {
			trackingIds = append(trackingIds, cloneInternalLogEntry(e.logs[e.logsTrackingIdsByDate[i]]))
		}
	}

	e.logmutex.RUnlock()

	pagingResult.CountRemaining = endIndex

	return trackingIds, pagingResult
}

func isValidLogEntryForACL(entry *InternalLogEntry) bool {
	return entry != nil && entry.Binding != nil && entry.Binding.Action != nil
}

func isLogEntryAllowedByACL(cfg *config.Config, user *authpublic.AuthenticatedUser, entry *InternalLogEntry) bool {
	return acl.IsAllowedLogs(cfg, user, entry.Binding.Action)
}

func (e *Executor) filterLogsByACL(cfg *config.Config, user *authpublic.AuthenticatedUser, dateFilter string) []*InternalLogEntry {
	e.logmutex.RLock()
	defer e.logmutex.RUnlock()

	filtered := make([]*InternalLogEntry, 0, len(e.logsTrackingIdsByDate))
	filterDate, hasDateFilter := parseDateFilter(dateFilter)

	for _, trackingId := range e.logsTrackingIdsByDate {
		entry := e.logs[trackingId]

		if shouldIncludeLogEntry(cfg, user, entry, filterDate, hasDateFilter) {
			filtered = append(filtered, cloneInternalLogEntry(entry))
		}
	}

	return filtered
}

// parseDateFilter parses the date filter string and returns filter information.
func parseDateFilter(dateFilter string) (filterDate time.Time, hasDateFilter bool) {
	if dateFilter == "" {
		return time.Time{}, false
	}

	parsedDate, err := time.Parse("2006-01-02", dateFilter)
	if err != nil {
		log.WithFields(log.Fields{
			"dateFilter": dateFilter,
			"error":      err,
		}).Errorf("Failed to parse date filter, expected format YYYY-MM-DD")
		return time.Time{}, false
	}

	return parsedDate, true
}

// shouldIncludeLogEntry determines if a log entry should be included based on ACL and date filter.
func shouldIncludeLogEntry(cfg *config.Config, user *authpublic.AuthenticatedUser, entry *InternalLogEntry, filterDate time.Time, hasDateFilter bool) bool {
	if !isValidLogEntryForACL(entry) {
		return false
	}

	if !isLogEntryAllowedByACL(cfg, user, entry) {
		return false
	}

	return matchesDateFilter(entry, filterDate, hasDateFilter)
}

// matchesDateFilter checks if the log entry matches the date filter.
func matchesDateFilter(entry *InternalLogEntry, filterDate time.Time, hasDateFilter bool) bool {
	if !hasDateFilter {
		return true
	}

	entryDate := entry.DatetimeStarted.UTC().Truncate(24 * time.Hour)
	filterDateUTC := filterDate.UTC().Truncate(24 * time.Hour)
	return entryDate.Equal(filterDateUTC)
}

// paginateFilteredLogs applies pagination to a filtered list of logs and returns
// the paginated results along with pagination metadata.
func paginateFilteredLogs(filtered []*InternalLogEntry, startOffset int64, pageCount int64) ([]*InternalLogEntry, *PagingResult) {
	total := int64(len(filtered))
	paging := &PagingResult{PageSize: pageCount, TotalCount: total, StartOffset: startOffset}

	if total == 0 {
		paging.CountRemaining = 0
		return []*InternalLogEntry{}, paging
	}

	startIndex := getPagingStartIndex(startOffset, total)
	pageCount = min(total, pageCount)
	endIndex := max(0, (startIndex-pageCount)+1)

	out := make([]*InternalLogEntry, 0, pageCount)
	for i := startIndex; i >= endIndex && i < int64(len(filtered)); i-- {
		out = append(out, filtered[i])
	}

	paging.CountRemaining = endIndex
	return out, paging
}

// GetLogTrackingIdsACL returns logs filtered by ACL visibility for the user and
// paginated correctly based on the filtered set.
// dateFilter is optional and should be in YYYY-MM-DD format. If empty, no date filtering is applied.
// expressionFilter is an optional filter expression applied after ACL checks.
func (e *Executor) GetLogTrackingIdsACL(cfg *config.Config, user *authpublic.AuthenticatedUser, startOffset int64, pageCount int64, dateFilter string, expressionFilter string) ([]*InternalLogEntry, *PagingResult, error) {
	filtered := e.filterLogsByACL(cfg, user, dateFilter)

	program, err := logfilter.Compile(expressionFilter)
	if err != nil {
		return nil, nil, err
	}

	filtered, err = applyLogFilter(filtered, program)
	if err != nil {
		return nil, nil, err
	}

	logs, paging := paginateFilteredLogs(filtered, startOffset, pageCount)
	return logs, paging, nil
}

func (e *Executor) GetLog(trackingID string) (*InternalLogEntry, bool) {
	e.logmutex.RLock()
	defer e.logmutex.RUnlock()

	entry, found := e.logs[trackingID]
	return cloneInternalLogEntry(entry), found
}

func (e *Executor) GetLogsByBindingId(bindingId string) []*InternalLogEntry {
	e.logmutex.RLock()
	defer e.logmutex.RUnlock()

	logs, found := e.LogsByBindingId[bindingId]

	if !found {
		return make([]*InternalLogEntry, 0)
	}

	cloned := make([]*InternalLogEntry, 0, len(logs))
	for _, entry := range logs {
		cloned = append(cloned, cloneInternalLogEntry(entry))
	}

	return cloned
}

// shouldCountExecution checks if a log entry should be counted for rate limiting.
func shouldCountExecution(logEntry *InternalLogEntry, windowStart time.Time) bool {
	return !logEntry.Blocked && !logEntry.Queued && logEntry.DatetimeStarted.After(windowStart)
}

// updateOldestExecution updates the oldest execution time if this entry is older.
func updateOldestExecution(oldestExecutionTime **time.Time, logEntry *InternalLogEntry) {
	if *oldestExecutionTime == nil {
		*oldestExecutionTime = &logEntry.DatetimeStarted
	} else if logEntry.DatetimeStarted.Before(**oldestExecutionTime) {
		*oldestExecutionTime = &logEntry.DatetimeStarted
	}
}

// findOldestExecutionInWindow finds the oldest execution within the time window and counts executions.
// Returns the count of executions and the oldest execution time, or nil if none found.
func findOldestExecutionInWindow(logs []*InternalLogEntry, windowStart time.Time) (int, *time.Time) {
	executions := 0
	var oldestExecutionTime *time.Time

	for _, logEntry := range logs {
		if !shouldCountExecution(logEntry, windowStart) {
			continue
		}

		executions++
		updateOldestExecution(&oldestExecutionTime, logEntry)
	}

	return executions, oldestExecutionTime
}

// calculateExpiryTime calculates when the oldest execution will fall outside the rate limit window.
func calculateExpiryTime(oldestExecutionTime time.Time, duration time.Duration, now time.Time) time.Time {
	expiryTime := oldestExecutionTime.Add(duration)
	if !expiryTime.After(now) {
		return time.Time{}
	}
	return expiryTime
}

// updateMaxExpiryTime updates maxExpiryTime if expiryTime is later.
func updateMaxExpiryTime(maxExpiryTime *time.Time, expiryTime time.Time) {
	if expiryTime.IsZero() {
		return
	}

	if maxExpiryTime.IsZero() || expiryTime.After(*maxExpiryTime) {
		*maxExpiryTime = expiryTime
	}
}

// calculateExpiryForRate calculates the expiry time for a single rate limit rule.
// Returns the expiry time if the rate limit is exceeded, or zero time if not.
func calculateExpiryForRate(rate config.RateSpec, logs []*InternalLogEntry, now time.Time) time.Time {
	duration := parseDuration(rate)
	if duration <= 0 {
		return time.Time{}
	}

	windowStart := now.Add(-duration)
	executions, oldestExecutionTime := findOldestExecutionInWindow(logs, windowStart)

	if executions < rate.Limit || oldestExecutionTime == nil {
		return time.Time{}
	}

	return calculateExpiryTime(*oldestExecutionTime, duration, now)
}

// getLogsForBinding retrieves logs for a binding ID.
func (e *Executor) getLogsForBinding(bindingId string) []*InternalLogEntry {
	e.logmutex.RLock()
	logs, found := e.LogsByBindingId[bindingId]
	e.logmutex.RUnlock()

	if !found || len(logs) == 0 {
		return nil
	}

	return logs
}

// calculateMaxExpiryTimeFromRates calculates the maximum expiry time across all rate limit rules.
func calculateMaxExpiryTimeFromRates(rates []config.RateSpec, logs []*InternalLogEntry, now time.Time) time.Time {
	var maxExpiryTime time.Time

	for _, rate := range rates {
		expiryTime := calculateExpiryForRate(rate, logs, now)
		updateMaxExpiryTime(&maxExpiryTime, expiryTime)
	}

	return maxExpiryTime
}

// GetTimeUntilAvailable calculates when an action will be available again based on rate limits.
// Returns the Unix timestamp in seconds when the rate limit expires, or 0 if the action is available now.
func (e *Executor) GetTimeUntilAvailable(binding *ActionBinding) int64 {
	if len(binding.Action.MaxRate) == 0 {
		return 0
	}

	logs := e.getLogsForBinding(binding.ID)
	if logs == nil {
		return 0
	}

	maxExpiryTime := calculateMaxExpiryTimeFromRates(binding.Action.MaxRate, logs, time.Now())

	if maxExpiryTime.IsZero() {
		return 0
	}

	return maxExpiryTime.Unix()
}

func (e *Executor) SetLog(trackingID string, entry *InternalLogEntry) string {
	e.logmutex.Lock()
	defer e.logmutex.Unlock()

	if _, found := e.logs[trackingID]; found || !isValidTrackingID(trackingID) {
		trackingID = uuid.NewString()
		entry.ExecutionTrackingID = trackingID
	}

	entry.Index = int64(len(e.logsTrackingIdsByDate))

	e.logs[trackingID] = entry
	e.logsTrackingIdsByDate = append(e.logsTrackingIdsByDate, trackingID)

	return trackingID
}

// ExecRequest processes an ExecutionRequest
func (e *Executor) ExecRequest(req *ExecutionRequest) (*sync.WaitGroup, string) {
	e.initializeExecRequest(req)

	log.Tracef("executor.ExecRequest(): trackingID=%s bindingID=%s", req.TrackingID, bindingIDForTrace(req))

	req.TrackingID = e.SetLog(req.TrackingID, req.logEntry)

	wg := new(sync.WaitGroup)
	wg.Add(1)

	go func() {
		queued := e.execChain(req, wg)
		if !queued {
			wg.Done()
		}
	}()

	return wg, req.TrackingID
}

func (e *Executor) initializeExecRequest(req *ExecutionRequest) {
	if req.AuthenticatedUser == nil {
		req.AuthenticatedUser = auth.UserGuest(req.Cfg)
	}

	req.executor = e
	req.logEntry = &InternalLogEntry{
		Binding:             req.Binding,
		DatetimeStarted:     time.Now(),
		ExecutionTrackingID: req.TrackingID,
		Output:              "",
		ExitCode:            DefaultExitCodeNotExecuted,
		ExecutionStarted:    false,
		ExecutionFinished:   false,
		ActionTitle:         "notfound",
		ActionIcon:          "&#x1f4a9;",
		Username:            req.AuthenticatedUser.Username,
	}
}

func bindingIDForTrace(req *ExecutionRequest) string {
	if req.Binding == nil {
		return ""
	}

	return req.Binding.ID
}

func (e *Executor) execChain(req *ExecutionRequest, wg *sync.WaitGroup) bool {
	if !req.skipRequestRegistration {
		finished, queued := e.registerOrQueueRequest(req, wg)
		if finished || queued {
			return queued
		}
	}

	e.runExecutionSteps(req)
	e.finishExecChain(req)

	return false
}

func (e *Executor) registerOrQueueRequest(req *ExecutionRequest, wg *sync.WaitGroup) (finished bool, queued bool) {
	if !stepRequestAction(req) {
		e.finishExecChain(req)
		return true, false
	}

	if e.finishIfConcurrencyBlocked(req) {
		return true, false
	}

	return e.queueRequestIfGroupLimited(req, wg)
}

func (e *Executor) finishIfConcurrencyBlocked(req *ExecutionRequest) bool {
	if actionNeedsGroupLimit(req) {
		return false
	}

	if stepConcurrencyCheck(req) {
		return false
	}

	e.finishExecChain(req)
	return true
}

func (e *Executor) queueRequestIfGroupLimited(req *ExecutionRequest, wg *sync.WaitGroup) (finished bool, queued bool) {
	if !actionNeedsGroupLimit(req) || e.groupsHaveCapacityForActive(req) {
		return false, false
	}

	return e.queueRequestAfterACL(req, wg)
}

func (e *Executor) queueRequestAfterACL(req *ExecutionRequest, wg *sync.WaitGroup) (finished bool, queued bool) {
	if !stepACLCheck(req) {
		e.finishExecChain(req)
		return true, false
	}

	if e.queueRequest(req, wg) {
		e.finishExecChain(req)
		return true, false
	}

	notifyListenersStarted(req)

	return false, true
}

func (e *Executor) runExecutionSteps(req *ExecutionRequest) {
	for _, step := range e.chainOfCommand[1:] {
		if !step(req) {
			break
		}
	}
}

func (e *Executor) finishExecChain(req *ExecutionRequest) {
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		if entry.DatetimeFinished.IsZero() {
			entry.DatetimeFinished = time.Now()
		}

		entry.ExecutionFinished = true
	})

	recordExecutionMetrics(req.logEntry)

	notifyListenersFinished(req)
	stepSaveLog(req)
	e.drainGroupQueue()
}

func getConcurrentCount(req *ExecutionRequest) int {
	concurrentCount := 0

	req.executor.logmutex.RLock()
	logs := req.executor.LogsByBindingId[req.Binding.ID]

	for _, logEntry := range logs {
		if !logEntry.ExecutionFinished && !logEntry.Queued {
			concurrentCount += 1
		}
	}

	req.executor.logmutex.RUnlock()

	return concurrentCount
}

func stepConcurrencyCheck(req *ExecutionRequest) bool {
	if actionNeedsGroupLimit(req) {
		return true
	}

	concurrentCount := getConcurrentCount(req)

	// Note that the current execution is counted int the logs, so when checking we +1
	if concurrentCount >= (req.Binding.Action.MaxConcurrent + 1) {
		log.WithFields(log.Fields{
			"actionTitle":     req.logEntry.ActionTitle,
			"concurrentCount": concurrentCount,
			"maxConcurrent":   req.Binding.Action.MaxConcurrent,
		}).Warnf("Blocked from executing due to concurrency limit")

		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output = "Blocked from executing due to concurrency limit"
			entry.Blocked = true
		})
		return false
	}

	return true
}

func parseDuration(rate config.RateSpec) time.Duration {
	duration, err := time.ParseDuration(rate.Duration)

	if err != nil {
		log.Warnf("Could not parse duration: %v", rate.Duration)

		return -1 * time.Minute
	}

	return duration
}

func entityPrefixForRequest(req *ExecutionRequest) string {
	if req.Binding != nil && req.Binding.Entity != nil {
		return req.Binding.Entity.UniqueKey
	}

	return ""
}

func rateExecutionMatchesScope(logEntry *InternalLogEntry, req *ExecutionRequest, entityPrefix string) bool {
	if logEntry.EntityPrefix != entityPrefix {
		return false
	}

	return !logEntry.Queued && logEntry.ExecutionTrackingID != req.TrackingID
}

func logEntryStartedInWindow(logEntry *InternalLogEntry, windowStart time.Time) bool {
	return logEntry.DatetimeStarted.After(windowStart) && !logEntry.Blocked
}

func rateExecutionCountsForRate(logEntry *InternalLogEntry, req *ExecutionRequest, entityPrefix string, windowStart time.Time) bool {
	return rateExecutionMatchesScope(logEntry, req, entityPrefix) && logEntryStartedInWindow(logEntry, windowStart)
}

func countRateExecutions(logs []*InternalLogEntry, req *ExecutionRequest, entityPrefix string, windowStart time.Time) int {
	executions := 0

	for _, logEntry := range logs {
		if rateExecutionCountsForRate(logEntry, req, entityPrefix, windowStart) {
			executions += 1
		}
	}

	return executions
}

func getExecutionsCount(rate config.RateSpec, req *ExecutionRequest) int {
	duration := parseDuration(rate)
	then := time.Now().Add(-duration)

	req.executor.logmutex.RLock()
	logs := req.executor.LogsByBindingId[req.Binding.ID]
	executions := countRateExecutions(logs, req, entityPrefixForRequest(req), then)
	req.executor.logmutex.RUnlock()

	return executions
}

func stepRateCheck(req *ExecutionRequest) bool {
	for _, rate := range req.Binding.Action.MaxRate {
		executions := getExecutionsCount(rate, req)

		if executions >= rate.Limit {
			log.WithFields(log.Fields{
				"actionTitle": req.logEntry.ActionTitle,
				"executions":  executions,
				"limit":       rate.Limit,
				"duration":    rate.Duration,
			}).Infof("Blocked from executing due to rate limit")

			req.mutateLogEntry(func(entry *InternalLogEntry) {
				entry.Output = "Blocked from executing due to rate limit"
				entry.Blocked = true
			})
			return false
		}
	}

	return true
}

func stepACLCheck(req *ExecutionRequest) bool {
	canExec := acl.IsAllowedExec(req.Cfg, req.AuthenticatedUser, req.Binding.Action)

	if !canExec {
		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output = "ACL check failed. Blocked from executing."
			entry.Blocked = true
		})

		log.WithFields(log.Fields{
			"actionTitle": req.logEntry.ActionTitle,
		}).Warnf("ACL check failed. Blocked from executing.")
	}

	return canExec
}

func stepParseArgs(req *ExecutionRequest) bool {
	if !prepareArgumentsForExecution(req) {
		return false
	}

	ok := parseActionForExecution(req)
	if ok {
		copyStorableArgumentsToLogEntry(req)
	}

	return ok
}

func prepareArgumentsForExecution(req *ExecutionRequest) bool {
	ensureArgumentMap(req)

	if !hasBindingAndAction(req) {
		return fail(req, fmt.Errorf("cannot parse arguments: Binding or Action is nil"))
	}

	filterToDefinedArgumentsOnly(req)
	if err := injectSystemArgs(req); err != nil {
		return fail(req, err)
	}

	mangleInvalidArgumentValues(req)
	return true
}

func parseActionForExecution(req *ExecutionRequest) bool {
	if hasExec(req) {
		return handleExecBranch(req)
	}

	return handleShellBranch(req)
}

func handleExecBranch(req *ExecutionRequest) bool {
	args, err := parseActionExec(req.Arguments, req.Binding.Action, req.Binding.Entity)

	if err != nil {
		return fail(req, err)
	}

	req.useDirectExec = true
	req.execArgs = args
	return true
}

func handleShellBranch(req *ExecutionRequest) bool {
	if hasWebhookTag(req) {
		return fail(req, fmt.Errorf("webhooks cannot use Shell execution; use exec instead. See https://docs.olivetin.app/action_execution/shellvsexec.html"))
	}
	if err := checkShellArgumentSafety(req.Binding.Action); err != nil {
		return fail(req, err)
	}

	cmd, err := parseActionArguments(req)

	if err != nil {
		return fail(req, err)
	}

	req.useDirectExec = false
	req.finalParsedCommand = cmd
	return true
}

func ensureArgumentMap(req *ExecutionRequest) {
	if req.Arguments == nil {
		req.Arguments = make(map[string]string)
	}
}

func filterToDefinedArgumentsOnly(req *ExecutionRequest) {
	definedNames := make(map[string]struct{})
	for _, arg := range req.Binding.Action.Arguments {
		definedNames[arg.Name] = struct{}{}
	}
	filtered := make(map[string]string)
	for k, v := range req.Arguments {
		if keepArgument(k, definedNames) {
			filtered[k] = v
		}
	}
	req.Arguments = filtered
}

func keepArgument(name string, definedNames map[string]struct{}) bool {
	_, ok := definedNames[name]
	return ok
}

func hasWebhookTag(req *ExecutionRequest) bool {
	return slices.Contains(req.Tags, "webhook")
}

var systemArgumentDefinitions = []config.ActionArgument{
	{Name: "ot_executionTrackingId", Type: "ascii_identifier", RejectNull: true},
	{Name: "ot_username", Type: "shell_safe_identifier", RejectNull: true},
}

func injectSystemArgs(req *ExecutionRequest) error {
	args, err := validatedSystemArgs(req)
	if err != nil {
		return err
	}

	maps.Copy(req.Arguments, args)

	return nil
}

func validatedSystemArgs(req *ExecutionRequest) (map[string]string, error) {
	values := map[string]string{
		"ot_executionTrackingId": req.TrackingID,
		"ot_username":            req.AuthenticatedUser.Username,
	}

	for i := range systemArgumentDefinitions {
		arg := &systemArgumentDefinitions[i]
		if err := ValidateArgument(arg, values[arg.Name], req.Binding.Action); err != nil {
			return nil, fmt.Errorf("system argument %q failed validation: %w", arg.Name, err)
		}
	}

	return values, nil
}

func hasBindingAndAction(req *ExecutionRequest) bool {
	return req.Binding != nil && req.Binding.Action != nil
}

func hasExec(req *ExecutionRequest) bool {
	return len(req.Binding.Action.Exec) > 0
}

func fail(req *ExecutionRequest, err error) bool {
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Output = err.Error()
	})
	log.Warn(err.Error())
	return false
}

func stepRequestAction(req *ExecutionRequest) bool {
	metricActionsRequested.Inc()

	if !stepRequestActionHasBinding(req) {
		return false
	}

	stepRequestActionPopulateLogEntry(req)
	stepRequestActionRegisterLog(req)

	log.WithFields(log.Fields{
		"actionTitle": req.logEntry.ActionTitle,
		"tags":        req.Tags,
	}).Infof("Action requested")

	notifyListenersStarted(req)

	return true
}

func stepRequestActionHasBinding(req *ExecutionRequest) bool {
	if req.Binding == nil || req.Binding.Action == nil {
		log.Warnf("Action request has no binding/action; skipping execution")
		return false
	}
	return true
}

func stepRequestActionPopulateLogEntry(req *ExecutionRequest) {
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Binding = req.Binding
		entry.ActionConfigTitle = req.Binding.Action.Title
		entry.ActionTitle = tpl.ParseTemplateOfActionBeforeExec(req.Binding.Action.Title, req.Binding.Entity)
		entry.ActionIcon = tpl.ParseTemplateOfActionBeforeExec(req.Binding.Action.Icon, req.Binding.Entity)
		entry.Tags = req.Tags
		entry.Justification = ResolveJustification(req)
		if req.Binding.Entity != nil {
			entry.EntityPrefix = req.Binding.Entity.UniqueKey
		}
	})
}

func stepRequestActionRegisterLog(req *ExecutionRequest) {
	req.executor.logmutex.Lock()
	defer req.executor.logmutex.Unlock()

	if _, containsKey := req.executor.LogsByBindingId[req.Binding.ID]; !containsKey {
		req.executor.LogsByBindingId[req.Binding.ID] = make([]*InternalLogEntry, 0)
	}
	req.executor.LogsByBindingId[req.Binding.ID] = append(req.executor.LogsByBindingId[req.Binding.ID], req.logEntry)
}

func stepLogStart(req *ExecutionRequest) bool {
	log.WithFields(log.Fields{
		"actionTitle": req.logEntry.ActionTitle,
		"timeout":     req.Binding.Action.Timeout,
	}).Infof("Action started")

	return true
}

func stepLogFinish(req *ExecutionRequest) bool {
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.ExecutionFinished = true
	})

	log.WithFields(log.Fields{
		"actionTitle":  req.logEntry.ActionTitle,
		"outputLength": len(req.logEntry.Output),
		"timedOut":     req.logEntry.TimedOut,
		"exit":         req.logEntry.ExitCode,
	}).Infof("Action finished")

	return true
}

func notifyListenersFinished(req *ExecutionRequest) {
	for _, listener := range req.executor.copyListeners() {
		listener.OnExecutionFinished(req.logEntry)
	}
}

func notifyListenersStarted(req *ExecutionRequest) {
	for _, listener := range req.executor.copyListeners() {
		listener.OnExecutionStarted(req.logEntry)
	}
}

func appendErrorToStderr(req *ExecutionRequest, err error) {
	if err == nil {
		return
	}

	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Output = err.Error() + "\n\n" + entry.Output
	})
}

type OutputStreamer struct {
	Req    *ExecutionRequest
	output bytes.Buffer
	mu     sync.Mutex
}

func (ost *OutputStreamer) Write(o []byte) (n int, err error) {
	for _, listener := range ost.Req.executor.copyListeners() {
		listener.OnOutputChunk(o, ost.Req.TrackingID)
	}

	ost.mu.Lock()
	n, err = ost.output.Write(o)
	outputSoFar := ""
	if err == nil {
		outputSoFar = ost.output.String()
	}
	ost.mu.Unlock()

	if err != nil {
		return n, err
	}

	// Keep the log entry's Output in sync while the command is still running so
	// ExecutionStatus / mid-run result views can show output produced so far.
	ost.Req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Output = outputSoFar
	})

	return n, nil
}

func (ost *OutputStreamer) String() string {
	ost.mu.Lock()
	defer ost.mu.Unlock()

	return ost.output.String()
}

func buildEnv(args map[string]string) []string {
	ret := append(os.Environ(), "OLIVETIN=1")

	for k, v := range args {
		varName := fmt.Sprintf("%v", strings.TrimSpace(strings.ToUpper(k)))

		// Skip arguments that might not have a name (eg, confirmation), as this causes weird bugs on Windows.
		if varName == "" {
			continue
		}

		ret = append(ret, fmt.Sprintf("%v=%v", varName, v))
	}

	return ret
}

func commandExitCode(cmd *exec.Cmd) int {
	if cmd == nil || cmd.ProcessState == nil {
		return -1
	}
	return cmd.ProcessState.ExitCode()
}

func stepExec(req *ExecutionRequest) bool {
	ctx, cancel := newTimeoutContext(context.Background(), time.Duration(req.Binding.Action.Timeout)*time.Second, req.executor)
	defer cancel()
	streamer := &OutputStreamer{Req: req}
	cmd := buildCommand(ctx, req)
	if cmd == nil {
		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output = "Cannot execute: no command arguments provided"
		})
		log.Warn("Cannot execute: no command arguments provided")
		return false
	}
	prepareCommand(cmd, streamer, req)
	runerr := cmd.Start()
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Process = cmd.Process
	})
	ctx.setProcess(cmd.Process)
	waiterr := cmd.Wait()
	finalOutput := streamer.String()
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.ExitCode = int32(commandExitCode(cmd))
		entry.Output = finalOutput
	})

	appendErrorToStderr(req, runerr)
	appendErrorToStderr(req, waiterr)

	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
		log.WithFields(log.Fields{
			"actionTitle": req.logEntry.ActionTitle,
		}).Warnf("Action timed out")

		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.TimedOut = true
			entry.Output += "OliveTin::timeout - this action timed out after " + fmt.Sprintf("%v", req.Binding.Action.Timeout) + " seconds. If you need more time for this action, set a longer timeout. See https://docs.olivetin.app/action_customization/timeouts.html for more help."
		})
	}

	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.DatetimeFinished = time.Now()
	})

	return true
}

func buildCommand(ctx context.Context, req *ExecutionRequest) *exec.Cmd {
	if req.useDirectExec {
		return wrapCommandDirect(ctx, req.execArgs)
	}
	return wrapCommandInShell(ctx, req.finalParsedCommand)
}

func prepareCommand(cmd *exec.Cmd, streamer *OutputStreamer, req *ExecutionRequest) {
	cmd.Stdout = streamer
	cmd.Stderr = streamer
	cmd.Env = buildEnv(req.Arguments)

	started := false
	req.mutateLogEntry(func(entry *InternalLogEntry) {
		if entry.ExecutionStarted {
			return
		}
		entry.ExecutionStarted = true
		started = true
	})
	if started {
		notifyListenersStarted(req)
	}
}

func stepExecAfter(req *ExecutionRequest) bool {
	ctx, cancel := newTimeoutContext(context.Background(), time.Duration(req.Binding.Action.Timeout)*time.Second, req.executor)
	defer cancel()

	var stdout bytes.Buffer
	var stderr bytes.Buffer

	cmd, args, err := buildShellAfterCommand(ctx, req, &stdout, &stderr)
	if err != nil {
		return fail(req, err)
	}
	if cmd == nil {
		return true
	}

	cmd.Env = buildEnv(args)

	runerr := cmd.Start()
	ctx.setProcess(cmd.Process)

	waiterr := cmd.Wait()

	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Output += "\n"
		entry.Output += "OliveTin::shellAfterCompleted stdout\n"
		entry.Output += stdout.String()
		entry.Output += "OliveTin::shellAfterCompleted stderr\n"
		entry.Output += stderr.String()
		entry.Output += "OliveTin::shellAfterCompleted errors and summary\n"
	})

	appendErrorToStderr(req, runerr)
	appendErrorToStderr(req, waiterr)

	if errors.Is(ctx.Err(), context.DeadlineExceeded) {
		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output += "Your shellAfterCompleted command timed out."
		})
	}

	req.mutateLogEntry(func(entry *InternalLogEntry) {
		entry.Output += fmt.Sprintf("Your shellAfterCompleted exited with code %v\n", commandExitCode(cmd))
		entry.Output += "OliveTin::shellAfterCompleted output complete\n"
	})

	return true
}

func shellAfterCompletedAction(req *ExecutionRequest) (*config.Action, bool) {
	if req == nil {
		return nil, false
	}
	if !hasBindingAndAction(req) {
		return nil, false
	}
	if req.Binding.Action.ShellAfterCompleted == "" {
		return nil, false
	}
	return req.Binding.Action, true
}

// Matches legacy and modern template forms for shellAfterCompleted output/exitCode,
// including optional .Arguments. prefix and flexible whitespace. These must become
// quoted env refs before template execution so command output cannot inject into sh -c.
var (
	shellAfterOutputRef   = regexp.MustCompile(`\{\{\s*(?:\.Arguments\.)?output\s*\}\}`)
	shellAfterExitCodeRef = regexp.MustCompile(`\{\{\s*(?:\.Arguments\.)?exitCode\s*\}\}`)
)

func substituteShellAfterCompletedEnvRefs(command string) string {
	command = replaceShellAfterEnvRef(command, shellAfterOutputRef, "$OUTPUT")
	command = replaceShellAfterEnvRef(command, shellAfterExitCodeRef, "$EXITCODE")
	return command
}

func replaceShellAfterEnvRef(command string, pattern *regexp.Regexp, envRef string) string {
	matches := pattern.FindAllStringIndex(command, -1)
	for i := len(matches) - 1; i >= 0; i-- {
		start, end := matches[i][0], matches[i][1]
		replacement := `"` + envRef + `"`
		if shellPosInsideSingleQuotes(command, start) {
			// Break out of single quotes so the env ref can expand at runtime.
			replacement = `'` + replacement + `'`
		}
		command = command[:start] + replacement + command[end:]
	}
	return command
}

func shellPosInsideSingleQuotes(command string, pos int) bool {
	inSingle := false
	inDouble := false
	i := 0

	for i < pos {
		inSingle, inDouble, i = advanceShellQuoteState(command, i, pos, inSingle, inDouble)
	}

	return inSingle
}

func advanceShellQuoteState(command string, i, pos int, inSingle, inDouble bool) (bool, bool, int) {
	if inSingle {
		return advanceInsideSingleQuote(command, i, inSingle, inDouble)
	}
	if inDouble {
		return advanceInsideDoubleQuote(command, i, pos, inSingle, inDouble)
	}
	return advanceOutsideQuotes(command, i, inSingle, inDouble)
}

func advanceInsideSingleQuote(command string, i int, inSingle, inDouble bool) (bool, bool, int) {
	if command[i] == '\'' {
		return false, inDouble, i + 1
	}
	return inSingle, inDouble, i + 1
}

func advanceInsideDoubleQuote(command string, i, pos int, inSingle, inDouble bool) (bool, bool, int) {
	if command[i] == '\\' && i+1 < pos {
		return inSingle, inDouble, i + 2
	}
	if command[i] == '"' {
		return inSingle, false, i + 1
	}
	return inSingle, inDouble, i + 1
}

func advanceOutsideQuotes(command string, i int, inSingle, inDouble bool) (bool, bool, int) {
	switch command[i] {
	case '\'':
		return true, inDouble, i + 1
	case '"':
		return inSingle, true, i + 1
	default:
		return inSingle, inDouble, i + 1
	}
}

// shellAfterTemplateArgs omits output/exitCode so templates cannot expand them
// raw. Those values are only provided as OUTPUT/EXITCODE process environment.
func shellAfterTemplateArgs(args map[string]string) map[string]string {
	templateArgs := make(map[string]string, len(args))
	for name, value := range args {
		if name == "output" || name == "exitCode" {
			continue
		}
		templateArgs[name] = value
	}
	return templateArgs
}

func parseShellAfterCompletedCommand(req *ExecutionRequest, commandTemplate string, args map[string]string) (string, error) {
	finalParsedCommand, err := tpl.ParseTemplateWithActionContext(commandTemplate, req.Binding.Entity, args)
	if err != nil {
		msg := "Could not prepare shellAfterCompleted command: " + err.Error() + "\n"
		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output += msg
		})
		log.Warn(msg)
		return "", err
	}

	return finalParsedCommand, nil
}

//gocyclo:ignore
func buildShellAfterCommand(ctx context.Context, req *ExecutionRequest, stdout, stderr *bytes.Buffer) (*exec.Cmd, map[string]string, error) {
	action, ok := shellAfterCompletedAction(req)
	if !ok {
		return nil, nil, nil
	}

	if hasWebhookTag(req) {
		return nil, nil, fmt.Errorf("webhooks cannot use shellAfterCompleted; use exec without after-completion shell instead. See https://docs.olivetin.app/action_execution/shellvsexec.html")
	}

	args, err := buildShellAfterArgs(req)
	if err != nil {
		return nil, nil, err
	}

	commandTemplate := substituteShellAfterCompletedEnvRefs(action.ShellAfterCompleted)
	finalParsedCommand, err := parseShellAfterCompletedCommand(req, commandTemplate, shellAfterTemplateArgs(args))
	if err != nil {
		return nil, nil, err
	}

	cmd := wrapCommandInShell(ctx, finalParsedCommand)
	cmd.Stdout = stdout
	cmd.Stderr = stderr

	return cmd, args, nil
}

func buildShellAfterArgs(req *ExecutionRequest) (map[string]string, error) {
	args, err := validatedSystemArgs(req)
	if err != nil {
		return nil, err
	}

	args["output"] = req.logEntry.Output
	args["exitCode"] = fmt.Sprintf("%v", req.logEntry.ExitCode)

	return args, nil
}

//gocyclo:ignore
func stepTrigger(req *ExecutionRequest) bool {
	if req.Binding.Action.Triggers == nil {
		return true
	}

	if req.TriggerDepth >= MaxTriggerDepth {
		log.WithFields(log.Fields{
			"actionTitle": req.logEntry.ActionTitle,
			"depth":       req.TriggerDepth,
		}).Warnf("Trigger action reached maximum depth of %v. Not triggering further actions.", MaxTriggerDepth)
		req.mutateLogEntry(func(entry *InternalLogEntry) {
			entry.Output += fmt.Sprintf("OliveTin::trigger - this action reached maximum trigger depth of %v. Not triggering further actions.", MaxTriggerDepth)
		})
		return true
	}

	if len(req.Tags) > 0 && req.Tags[0] == "trigger" {
		log.Warnf("Trigger action is triggering another trigger action. This is allowed, but be careful not to create trigger loops.")
	}

	triggerLoop(req)

	return true
}

func triggerLoop(req *ExecutionRequest) {
	for _, triggerTitle := range req.Binding.Action.Triggers {
		binding := req.executor.findBindingByActionTitle(triggerTitle, "")
		if binding == nil {
			log.WithFields(log.Fields{
				"triggerTitle": triggerTitle,
				"fromAction":   req.logEntry.ActionTitle,
			}).Warnf("Trigger references unknown action title; skipping")
			continue
		}
		trigger := &ExecutionRequest{
			Binding:           binding,
			TrackingID:        uuid.NewString(),
			Tags:              []string{"trigger"},
			AuthenticatedUser: req.AuthenticatedUser,
			Arguments:         req.Arguments,
			Cfg:               req.Cfg,
			TriggerDepth:      req.TriggerDepth + 1,
			Justification:     fmt.Sprintf("Triggered by action: %s", req.logEntry.ActionTitle),
		}

		req.executor.ExecRequest(trigger)
	}
}

func stepSaveLog(req *ExecutionRequest) bool {
	if !canSaveExecutionLog(req) {
		log.Warnf("Cannot save execution log; missing request, log entry, binding/action, or config")
		return false
	}

	filename := fmt.Sprintf("%v.%v.%v", sanitizeLogFilename(req.logEntry.ActionTitle), req.logEntry.DatetimeStarted.Unix(), req.logEntry.ExecutionTrackingID)

	saveLogResults(req, filename)
	saveLogOutput(req, filename)

	return true
}

func canSaveExecutionLog(req *ExecutionRequest) bool {
	return req != nil && req.logEntry != nil && req.Binding != nil && req.Binding.Action != nil && req.Cfg != nil
}

// sanitizeLogFilename replaces characters that are unsafe in filenames so action
// titles like "Create/update Report" do not create nested paths or fail to write.
func sanitizeLogFilename(title string) string {
	oldnew := []string{
		"/", "_",
		"\\", "_",
		":", "_",
		"*", "_",
		"?", "_",
		"\"", "_",
		"<", "_",
		">", "_",
		"|", "_",
	}

	// NUL and other C0 controls plus DEL are invalid or problematic in filenames.
	for i := 0; i < 32; i++ {
		oldnew = append(oldnew, string(rune(i)), "_")
	}
	oldnew = append(oldnew, "\x7f", "_")

	return strings.NewReplacer(oldnew...).Replace(title)
}

func firstNonEmpty(one, two string) string {
	if one != "" {
		return one
	}

	return two
}

func saveLogResults(req *ExecutionRequest, filename string) {
	dir := firstNonEmpty(req.Binding.Action.SaveLogs.ResultsDirectory, req.Cfg.SaveLogs.ResultsDirectory)

	if dir != "" {
		data, err := yaml.Marshal(req.logEntry)

		if err != nil {
			log.Warnf("%v", err)
		}

		filepath := path.Join(dir, filename+".yaml")
		err = os.WriteFile(filepath, data, 0600)

		if err != nil {
			log.Warnf("%v", err)
		}
	}
}

func saveLogOutput(req *ExecutionRequest, filename string) {
	dir := firstNonEmpty(req.Binding.Action.SaveLogs.OutputDirectory, req.Cfg.SaveLogs.OutputDirectory)

	if dir != "" {
		data := req.logEntry.Output
		filepath := path.Join(dir, filename+".log")
		err := os.WriteFile(filepath, []byte(data), 0600)

		if err != nil {
			log.Warnf("%v", err)
		}
	}
}