// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package report

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"gitea.com/gitea/runner/act/common/logging"
	"gitea.com/gitea/runner/act/runner"
	"gitea.com/gitea/runner/internal/pkg/client"
	"gitea.com/gitea/runner/internal/pkg/config"
	"gitea.com/gitea/runner/internal/pkg/metrics"

	"connectrpc.com/connect"
	"gitea.dev/actionslib/pkg/model"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

// errOutputsNotSent travels the same return path as transport failures but is not one.
var errOutputsNotSent = errors.New("there are still outputs that have not been sent")

// Size limits for the outputs reported to the server.
const (
	maxOutputKeyLen   = 255
	maxOutputValueLen = 1024 * 1024 // 1 MiB
)

// jobOutput is a job output on its way to the server, sent once the server has acknowledged it.
type jobOutput struct {
	value string
	sent  bool
}

type Reporter struct {
	ctx    context.Context
	cancel context.CancelFunc

	closed  bool
	client  client.Client
	clientM sync.Mutex

	logOffset   int
	logRows     []*runnerv1.LogRow
	logReplacer *strings.Replacer
	oldnew      []string

	// lastLogBufferRows is the last value written to the ReportLogBufferRows
	// gauge; guarded by clientM (the same lock held around each ReportLog call)
	// so the gauge skips no-op Set calls when the buffer size is unchanged.
	lastLogBufferRows int

	state        *runnerv1.TaskState
	stateChanged bool
	// The steps in state.ReportedSteps by stage and workflow step number, the step the next log row
	// without a step of its own belongs to, and the workflow steps. Guarded by stateMu.
	reportedSteps       map[reportedStepKey]*runnerv1.StepState
	currentReportedStep *runnerv1.StepState
	stepModels          []*model.Step
	// reportFailing keeps an outage to one log line at each end.
	reportFailing map[string]bool
	// serverResult is what the server decided, e.g. the zombie reaper failing
	// the task. Guarded by stateMu.
	serverResult      runnerv1.Result
	stateMu           sync.RWMutex
	outputsMu         sync.Mutex
	outputs           map[string]jobOutput
	daemon            chan struct{}
	heartbeatStop     chan struct{}
	heartbeatStopOnce sync.Once

	// Unix-nanos of the last successful UpdateTask. Atomic so the heartbeat
	// guard in ReportState reads it without contending stateMu.
	lastReportedAtNanos atomic.Int64

	// Adaptive batching control
	logReportInterval   time.Duration
	logReportMaxLatency time.Duration
	logBatchSize        int
	logReportLevel      logging.Level
	stateReportInterval time.Duration
	// closeTimeout bounds each RPC attempt in the final flush, on a context
	// detached from r.ctx so a server cancel can't abort the acknowledgement.
	closeTimeout time.Duration
	// daemonWait bounds how long Close waits for the daemon loop to acknowledge.
	daemonWait time.Duration

	// Event notification channels (non-blocking, buffered 1)
	logNotify   chan struct{} // signal: new log rows arrived
	stateNotify chan struct{} // signal: step transition (start/stop)

	jobLog *jobLog // this task's rows on the runner's own disk, nil when log.job.dir is unset
}

// extraMasks are values known before the job starts that are not among its secrets, such as
// the password in the runner's proxy URL.
func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.Client, task *runnerv1.Task, cfg *config.Config, extraMasks ...string) *Reporter {
	var oldnew []string
	for _, v := range extraMasks {
		oldnew = runner.AppendSecretMasker(oldnew, v)
	}
	if v := task.Context.Fields["token"].GetStringValue(); v != "" {
		oldnew = runner.AppendSecretMasker(oldnew, v)
	}
	if v := task.Context.Fields["gitea_runtime_token"].GetStringValue(); v != "" {
		oldnew = runner.AppendSecretMasker(oldnew, v)
	}
	if v := task.Context.Fields["actions_id_token_request_token"].GetStringValue(); v != "" {
		oldnew = runner.AppendSecretMasker(oldnew, v)
	}
	oldnew = runner.AppendSecretMaskers(oldnew, task.Secrets)

	rv := &Reporter{
		ctx:                 ctx,
		cancel:              cancel,
		client:              client,
		oldnew:              oldnew,
		logReplacer:         runner.NewSecretReplacer(oldnew),
		logReportInterval:   cfg.Runner.LogReportInterval,
		logReportMaxLatency: cfg.Runner.LogReportMaxLatency,
		logBatchSize:        cfg.Runner.LogReportBatchSize,
		logReportLevel:      logging.GetLevel(),
		stateReportInterval: cfg.Runner.StateReportInterval,
		closeTimeout:        cfg.Runner.ReportCloseTimeout,
		logNotify:           make(chan struct{}, 1),
		stateNotify:         make(chan struct{}, 1),
		state: &runnerv1.TaskState{
			Id: task.Id,
		},
		reportFailing: map[string]bool{},
		daemon:        make(chan struct{}),
		heartbeatStop: make(chan struct{}),
		jobLog:        openJobLog(cfg.Log.Job, task.Id, time.Now()),
	}

	if level, err := logging.ParseLevel(cfg.Runner.LogReportLevel); err == nil { // unset keeps log.level
		rv.logReportLevel = level
	}

	rv.daemonWait = 6 * rv.effectiveCloseTimeout()

	registerGlobalMasks(rv)

	return rv
}

// Result returns the final job result. Safe to call after Close() returns.
func (r *Reporter) Result() runnerv1.Result {
	r.stateMu.RLock()
	defer r.stateMu.RUnlock()
	return r.state.Result
}

// ResetSteps sets the steps of the workflow job. They are reported both by their index in the
// workflow file, for Gitea versions deriving the steps from the file, and as the steps the runner
// runs, together with the setup, pre, post and cleanup steps, for Gitea versions that show those.
func (r *Reporter) ResetSteps(steps []*model.Step) {
	r.stateMu.Lock()
	defer r.stateMu.Unlock()
	for i := range steps {
		r.state.Steps = append(r.state.Steps, &runnerv1.StepState{
			Id: int64(i),
		})
	}
	if len(steps) == 0 {
		return // a reusable workflow caller, whose called jobs log their own steps through it
	}
	r.stepModels = steps
	r.reportedSteps = map[reportedStepKey]*runnerv1.StepState{}
	setup := r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_SETUP, 0, "Set up job")
	setup.StartedAt = cmp.Or(r.state.StartedAt, timestamppb.Now())
	setup.LogLength = int64(r.logOffset + len(r.logRows)) // the runner's own lines logged before the job was parsed
	r.currentReportedStep = setup
	for i, step := range steps {
		r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_MAIN, i, r.mask(stepDisplayName(step, "")))
	}
	r.addReportedStep(len(r.state.ReportedSteps), runnerv1.StepStage_STEP_STAGE_CLEANUP, 0, "Complete job")
}

type reportedStepKey struct {
	stage  runnerv1.StepStage
	number int
}

// stepDisplayName names a workflow step like Gitea and GitHub do, prefixing an unnamed one with "Run".
// evaluated is step.String() with its expressions evaluated, empty while the step has not run yet.
func stepDisplayName(step *model.Step, evaluated string) string {
	if step.Name != "" {
		return cmp.Or(evaluated, step.Name)
	}
	first, _, _ := strings.Cut(strings.TrimSpace(cmp.Or(evaluated, step.String())), "\n")
	return "Run " + cmp.Or(strings.TrimSpace(first), step.String())
}

// addReportedStep inserts a step at pos of the reported steps. Caller holds stateMu.
func (r *Reporter) addReportedStep(pos int, stage runnerv1.StepStage, number int, name string) *runnerv1.StepState {
	step := &runnerv1.StepState{Name: name, Stage: stage, Number: int64(number)}
	r.state.ReportedSteps = slices.Insert(r.state.ReportedSteps, pos, step)
	for i, v := range r.state.ReportedSteps[pos:] {
		v.Id = int64(pos + i)
	}
	r.reportedSteps[reportedStepKey{stage, number}] = step
	return step
}

// reportedStepOf returns the reported step an entry belongs to, adding a pre or post step the
// first time it runs, and whether the entry names that step itself. Caller holds stateMu.
func (r *Reporter) reportedStepOf(entry logging.Record) (*runnerv1.StepState, bool) {
	if r.reportedSteps == nil {
		return nil, false
	}
	stage, _ := recordAttr[string](entry, "stage")
	number, ok := recordAttr[int](entry, "stepNumber")
	if stage == runner.StageComplete {
		return r.reportedSteps[reportedStepKey{stage: runnerv1.StepStage_STEP_STAGE_CLEANUP}], false
	}
	if !ok || number < 0 || number >= len(r.stepModels) {
		return r.currentReportedStep, false
	}
	var key reportedStepKey
	switch stage {
	case runner.StagePre:
		key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_PRE, number}
	case runner.StageMain:
		key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number}
	case runner.StagePost:
		key = reportedStepKey{runnerv1.StepStage_STEP_STAGE_POST, number}
	default:
		return r.currentReportedStep, false
	}
	main, ok := r.reportedSteps[reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number}]
	if !ok {
		return r.currentReportedStep, false
	}
	if evaluated, _ := recordAttr[string](entry, "step"); evaluated != "" {
		// evaluated as GitHub shows it, and masked as an expression can name a secret
		main.Name = r.mask(stepDisplayName(r.stepModels[number], evaluated))
	}
	if step, ok := r.reportedSteps[key]; ok {
		return step, true
	}
	if started, _ := recordAttr[bool](entry, runner.StepStartedField); !started {
		return r.currentReportedStep, false // a pre or post step that does not run is not shown
	}
	mainName := main.Name
	if stage == runner.StagePre {
		// pre steps run in the order of the workflow steps, so each goes right before the main steps
		pos := slices.IndexFunc(r.state.ReportedSteps, func(v *runnerv1.StepState) bool { return v.Stage == runnerv1.StepStage_STEP_STAGE_MAIN })
		return r.addReportedStep(pos, key.stage, number, "Pre "+mainName), true
	}
	return r.addReportedStep(len(r.state.ReportedSteps)-1, key.stage, number, "Post "+mainName), true
}

// recordAttr returns the attribute key of rec if it is a T.
func recordAttr[T any](rec logging.Record, key string) (T, bool) {
	v, _ := rec.Attr(key)
	t, ok := v.(T)
	return t, ok
}

// enterReportedStep makes step the one log rows without a step of their own belong to, finishing
// the setup step once another step starts. It reports whether the step started. Caller holds stateMu.
func (r *Reporter) enterReportedStep(step *runnerv1.StepState, timestamp time.Time) bool {
	if prev := r.currentReportedStep; step != prev && prev.Stage == runnerv1.StepStage_STEP_STAGE_SETUP && prev.StoppedAt == nil {
		if prev.Result == runnerv1.Result_RESULT_UNSPECIFIED {
			prev.Result = runnerv1.Result_RESULT_SUCCESS
		}
		prev.StoppedAt = timestamppb.New(timestamp)
	}
	r.currentReportedStep = step
	if step.StartedAt != nil {
		return false
	}
	step.StartedAt = timestamppb.New(timestamp)
	return true
}

// finishReportedSteps gives a result to the reported steps that have none when the job ends:
// the job's result to a step that was running, except the cleanup that ran in full. A running
// cleanup is left running unless closing, as it still logs after the job result. Caller holds stateMu.
func (r *Reporter) finishReportedSteps(jobResult runnerv1.Result, timestamp time.Time, closing bool) {
	for _, v := range r.state.ReportedSteps {
		if v.Result != runnerv1.Result_RESULT_UNSPECIFIED || !closing && v.StartedAt != nil && v.Stage == runnerv1.StepStage_STEP_STAGE_CLEANUP {
			continue
		}
		switch {
		case v.StartedAt == nil && jobResult == runnerv1.Result_RESULT_CANCELLED:
			v.Result = runnerv1.Result_RESULT_CANCELLED
		case v.StartedAt == nil:
			v.Result = runnerv1.Result_RESULT_SKIPPED // never reached, such as the steps after a failed setup
		case v.Stage == runnerv1.StepStage_STEP_STAGE_CLEANUP:
			v.Result = runnerv1.Result_RESULT_SUCCESS
		default:
			v.Result = jobResult
		}
		if v.StartedAt != nil {
			v.StoppedAt = timestamppb.New(timestamp)
		}
	}
}

// Enabled makes the Reporter see every record, the job logger's level filter runs before it.
func (r *Reporter) Enabled(context.Context, logging.Level) bool { return true }

// WithAttrs returns a Handler feeding the same Reporter, with attrs added to every record.
func (r *Reporter) WithAttrs(attrs []logging.Attr) logging.Handler {
	return &reporterHandler{r: r, attrs: attrs}
}

type reporterHandler struct {
	r     *Reporter
	attrs []logging.Attr
}

func (h *reporterHandler) Enabled(ctx context.Context, level logging.Level) bool {
	return h.r.Enabled(ctx, level)
}

func (h *reporterHandler) Handle(ctx context.Context, rec logging.Record) error {
	rec.Attrs = slices.Concat(h.attrs, rec.Attrs)
	return h.r.Handle(ctx, rec)
}

func (h *reporterHandler) WithAttrs(attrs []logging.Attr) logging.Handler {
	return &reporterHandler{r: h.r, attrs: slices.Concat(h.attrs, attrs)}
}

// appendLogRow masks a row before buffering it for Gitea and the local job.log, the one point
// feeding both. Caller holds stateMu.
func (r *Reporter) appendLogRow(row *runnerv1.LogRow) {
	row.Content = r.mask(row.Content)
	if step := r.currentReportedStep; step != nil {
		if step.LogLength == 0 {
			step.LogIndex = int64(r.logOffset + len(r.logRows))
		}
		step.LogLength++
	}
	r.logRows = append(r.logRows, row)
	r.jobLog.write(row.Time.AsTime(), row.Content)
}

func attrArgs(attrs []logging.Attr) []any {
	args := make([]any, len(attrs))
	for i, a := range attrs {
		args[i] = a
	}
	return args
}

// isJobStepEntry is used to not report composite step results incorrectly as step result
// returns true if the logentry is on job level
// returns false for composite action step messages
func isJobStepEntry(rec logging.Record) bool {
	if v, ok := rec.Attr("stepID"); ok {
		if v, ok := v.([]string); ok && len(v) > 1 {
			return false
		}
	}
	return true
}

// notifyLog sends a non-blocking signal that new log rows are available.
func (r *Reporter) notifyLog() {
	select {
	case r.logNotify <- struct{}{}:
	default:
	}
}

// notifyState sends a non-blocking signal that a UX-critical state change occurred (step start/stop, job result).
func (r *Reporter) notifyState() {
	select {
	case r.stateNotify <- struct{}{}:
	default:
	}
}

// unlockAndNotify releases stateMu and sends channel notifications.
// Must be called with stateMu held.
func (r *Reporter) unlockAndNotify(urgentState bool) {
	r.stateMu.Unlock()
	r.notifyLog()
	if urgentState {
		r.notifyState()
	}
}

func (r *Reporter) Handle(_ context.Context, entry logging.Record) error {
	urgentState := false

	r.stateMu.Lock()

	r.stateChanged = true

	if logging.Enabled(logging.LevelTrace) {
		logging.With(attrArgs(entry.Attrs)...).Trace(r.mask(entry.Message)) // the process masker has no ::add-mask:: value
	}

	timestamp := entry.Time
	if r.state.StartedAt == nil {
		r.state.StartedAt = timestamppb.New(timestamp)
	}

	if reported, own := r.reportedStepOf(entry); reported != nil {
		if r.enterReportedStep(reported, timestamp) {
			urgentState = true
		}
		if !own && entry.Level >= logging.LevelError && reported.Result == runnerv1.Result_RESULT_UNSPECIFIED {
			reported.Result = runnerv1.Result_RESULT_FAILURE // a job level error, such as an action that cannot be downloaded
			urgentState = true
		}
		if v, ok := entry.Attr("stepResult"); ok && own && isJobStepEntry(entry) {
			if stepResult, ok := r.parseResult(v); ok {
				reported.Result = stepResult
				reported.StoppedAt = timestamppb.New(timestamp)
				urgentState = true
			}
		}
	}

	stage, _ := entry.Attr("stage")

	if stage != runner.StageMain {
		if v, ok := entry.Attr("jobResult"); ok {
			if jobResult, ok := r.parseResult(v); ok {
				// We need to ensure log upload before this upload
				r.state.Result = jobResult
				r.state.StoppedAt = timestamppb.New(timestamp)
				for _, s := range r.state.Steps {
					if s.Result == runnerv1.Result_RESULT_UNSPECIFIED {
						s.Result = runnerv1.Result_RESULT_CANCELLED
						if jobResult == runnerv1.Result_RESULT_SKIPPED {
							s.Result = runnerv1.Result_RESULT_SKIPPED
						}
					}
				}
				r.finishReportedSteps(jobResult, timestamp, false)
				urgentState = true
			}
		}
		if r.shouldAppendLogRow(entry) {
			r.appendLogRow(logRow(entry))
		}
		r.unlockAndNotify(urgentState)
		return nil
	}

	var step *runnerv1.StepState
	if v, ok := entry.Attr("stepNumber"); ok {
		if v, ok := v.(int); ok && len(r.state.Steps) > v {
			step = r.state.Steps[v]
		}
	}
	if step == nil {
		if r.shouldAppendLogRow(entry) {
			r.appendLogRow(logRow(entry))
		}
		r.unlockAndNotify(false)
		return nil
	}

	if step.StartedAt == nil {
		step.StartedAt = timestamppb.New(timestamp)
		urgentState = true
	}

	// Force reporting log errors as raw output to prevent silent failures
	v, ok := entry.Attr("raw_output")
	if entry.Level >= logging.LevelError {
		v, ok = true, true
	}

	if ok {
		if rawOutput, ok := v.(bool); ok && rawOutput {
			if step.LogLength == 0 {
				step.LogIndex = int64(r.logOffset + len(r.logRows))
			}
			step.LogLength++
			r.appendLogRow(logRow(entry))
		}
	} else if r.shouldAppendLogRow(entry) {
		r.appendLogRow(logRow(entry))
	}
	if v, ok := entry.Attr("stepResult"); ok && isJobStepEntry(entry) {
		if stepResult, ok := r.parseResult(v); ok {
			if step.LogLength == 0 {
				step.LogIndex = int64(r.logOffset + len(r.logRows))
			}
			step.Result = stepResult
			step.StoppedAt = timestamppb.New(timestamp)
			urgentState = true
		}
	}

	r.unlockAndNotify(urgentState)
	return nil
}

// Only the daemon loop calls this, so reportFailing needs no lock.
func (r *Reporter) noteReport(method string, err error) {
	if errors.Is(err, errOutputsNotSent) {
		err = nil // the RPC itself succeeded
	}
	switch {
	case err != nil && !r.reportFailing[method]:
		r.reportFailing[method] = true
		logging.Warnf("%s error: %v, retrying until reconnected", method, err)
	case err == nil && r.reportFailing[method]:
		delete(r.reportFailing, method)
		logging.Infof("%s reconnected", method)
	}
}

func (r *Reporter) RunDaemon() {
	go r.runDaemonLoop()
}

// StopHeartbeats stops periodic UpdateTask heartbeats without cancelling the
// task context. Close() still delivers the final flush. Safe to call multiple
// times and when the context is already cancelled.
func (r *Reporter) StopHeartbeats() {
	r.heartbeatStopOnce.Do(func() {
		close(r.heartbeatStop)
	})
}

func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
	if *active {
		if !timer.Stop() {
			select {
			case <-timer.C:
			default:
			}
		}
		*active = false
	}
}

func (r *Reporter) runDaemonLoop() {
	defer close(r.daemon)

	logTicker := time.NewTicker(r.logReportInterval)
	stateTicker := time.NewTicker(r.stateReportInterval)

	// maxLatencyTimer ensures the first buffered log row is sent within logReportMaxLatency.
	// Start inactive — it is armed when the first log row arrives in an empty buffer.
	maxLatencyTimer := time.NewTimer(0)
	if !maxLatencyTimer.Stop() {
		<-maxLatencyTimer.C
	}
	maxLatencyActive := false

	defer logTicker.Stop()
	defer stateTicker.Stop()
	defer maxLatencyTimer.Stop()

	for {
		select {
		case <-logTicker.C:
			r.reportLogWithState()
			r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)

		case <-stateTicker.C:
			r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))

		case <-r.logNotify:
			r.stateMu.RLock()
			n := len(r.logRows)
			r.stateMu.RUnlock()

			if n >= r.logBatchSize {
				r.reportLogWithState()
				r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
			} else if !maxLatencyActive && n > 0 {
				maxLatencyTimer.Reset(r.logReportMaxLatency)
				maxLatencyActive = true
			}

		case <-r.stateNotify:
			// Step transition or job result — flush both immediately for frontend UX.
			r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
			r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
			r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)

		case <-maxLatencyTimer.C:
			maxLatencyActive = false
			r.reportLogWithState()

		case <-r.ctx.Done():
			// Stop heartbeating on cancel so Gitea sees the runner as offline
			// during cleanup and won't assign an overlapping task. Close() still
			// delivers the final flush on a detached context (flushFinal).
			return

		case <-r.heartbeatStop:
			// Stop heartbeating during post-task script execution. Close() still
			// delivers the final flush on a detached context (flushFinal).
			return
		}

		r.stateMu.RLock()
		closed := r.closed
		r.stateMu.RUnlock()
		if closed {
			return
		}
	}
}

// Gitea slices the single log stream into steps by the LogIndex/LogLength the state
// carries, so rows acked without a state report behind them stay unattributed, and stay
// that way for good if the runner never reports again.
func (r *Reporter) reportLogWithState() {
	took, err := r.reportLog(false)
	r.noteReport(metrics.LabelMethodUpdateLog, err)
	if took {
		r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
	}
}

func (r *Reporter) Logf(format string, a ...any) {
	r.stateMu.Lock()
	defer r.stateMu.Unlock()

	r.logf(format, a...)
}

func (r *Reporter) logf(format string, a ...any) {
	if !r.duringSteps() {
		// Masked like any other row, as a caller can interpolate a secret such as a configured URL
		// carrying credentials.
		r.appendLogRow(&runnerv1.LogRow{Time: timestamppb.Now(), Content: fmt.Sprintf(format, a...)})
	}
}

func (r *Reporter) SetOutputs(outputs map[string]string) {
	r.stateMu.Lock()
	defer r.stateMu.Unlock()
	r.outputsMu.Lock()
	defer r.outputsMu.Unlock()

	if r.outputs == nil {
		r.outputs = map[string]jobOutput{}
	}
	for k, v := range outputs {
		if l := len(k); l > maxOutputKeyLen {
			logging.Warnf("ignore output %q because the key is too long: %d > %d", k, l, maxOutputKeyLen)
			r.logf("ignore output %q because the key is too long: %d > %d", k, l, maxOutputKeyLen)
			continue
		}
		if l := len(v); l > maxOutputValueLen {
			logging.Warnf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen)
			r.logf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen)
			continue
		}
		if r.logReplacer.Replace(v) != v { // GitHub skips an output that may carry a secret rather than masking it
			logging.Warnf("ignore output %q because it may contain a secret", k)
			r.logf("ignore output %q because it may contain a secret", k)
			continue
		}
		if _, ok := r.outputs[k]; !ok {
			r.outputs[k] = jobOutput{value: v}
		}
	}
}

func (r *Reporter) Close(lastWords string) error {
	defer deregisterGlobalMasks(r) // deferred so a panic below cannot strand this task's masks
	r.stateMu.Lock()
	r.closed = true
	if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED {
		// No result of its own, so say why it stopped.
		result, words := runnerv1.Result_RESULT_FAILURE, "Early termination"
		switch {
		case r.serverResult != runnerv1.Result_RESULT_UNSPECIFIED:
			result, words = r.serverResult, "Ended by the server"
		case errors.Is(r.ctx.Err(), context.Canceled):
			result, words = runnerv1.Result_RESULT_CANCELLED, "Cancelled"
		}
		if lastWords == "" {
			lastWords = words
		}
		for _, v := range r.state.Steps {
			if v.Result == runnerv1.Result_RESULT_UNSPECIFIED {
				v.Result = runnerv1.Result_RESULT_CANCELLED
			}
		}
		r.state.Result = result
		r.appendLogRow(&runnerv1.LogRow{
			Time:    timestamppb.Now(),
			Content: lastWords,
		})
		r.state.StoppedAt = timestamppb.Now()
	} else if lastWords != "" {
		r.appendLogRow(&runnerv1.LogRow{
			Time:    timestamppb.Now(),
			Content: lastWords,
		})
	}
	r.finishReportedSteps(r.state.Result, time.Now(), true)
	r.stateMu.Unlock()

	// Wake up the daemon loop so it detects closed promptly.
	r.notifyLog()

	// Wait for Acknowledge
	select {
	case <-r.daemon:
	case <-time.After(r.daemonWait):
		logging.Errorf("No Response from RunDaemon for %s, continue best effort", r.daemonWait)
	}

	// Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore,
	// so a final empty request never runs TransferLogs and dbfs_data leaks.
	// Inject a sentinel row after the daemon has exited so it can't be flushed
	// before ReportLog(true).
	// TODO: Remove after https://github.com/go-gitea/gitea/pull/37631 is in all
	// supported branches, e.g. v1.28+.
	r.stateMu.Lock()
	if len(r.logRows) == 0 {
		// Not appendLogRow: the sentinel is not job output and has no place in job.log.
		r.logRows = append(r.logRows, &runnerv1.LogRow{
			Time:    timestamppb.Now(),
			Content: "",
		})
	}
	r.stateMu.Unlock()

	// Separate budgets so a slow ReportLog can't starve the ReportState that
	// carries the cancel acknowledgement.
	err := errors.Join(
		r.flushFinal(func() error { return r.ReportLog(true) }),
		r.flushFinal(func() error { return r.ReportState(true) }),
	)

	// After the flush so a failed handover is in the file too, under stateMu so a late entry cannot race.
	r.stateMu.Lock()
	trailer := fmt.Sprintf("task %d finished: %s", r.state.Id, metrics.ResultToStatusLabel(r.state.Result))
	if err != nil {
		trailer += fmt.Sprintf(", the final flush to Gitea failed: %v", err)
	}
	r.jobLog.close(r.mask(trailer))
	r.stateMu.Unlock()

	return err
}

// flushFinal retries fn on a detached, bounded context so a cancelled r.ctx
// does not abort the final flush. Each call gets its own fresh budget.
func (r *Reporter) flushFinal(fn func() error) error {
	ctx, cancel := context.WithTimeout(context.Background(), 3*r.effectiveCloseTimeout())
	defer cancel()
	var errs []error
	for attempt := range 10 {
		err := fn()
		if err == nil {
			return nil
		}
		errs = append(errs, err)
		if attempt == 9 {
			break
		}
		select {
		case <-time.After(100*time.Millisecond<<attempt + rand.N(100*time.Millisecond)):
		case <-ctx.Done():
			return errors.Join(append(errs, ctx.Err())...)
		}
	}
	return errors.Join(errs...)
}

// effectiveCloseTimeout returns closeTimeout, or 10s when unset, so a zero
// value can't produce an already-expired context for the final flush.
func (r *Reporter) effectiveCloseTimeout() time.Duration {
	if r.closeTimeout <= 0 {
		return 10 * time.Second
	}
	return r.closeTimeout
}

// rpcCtx returns the context for an outbound RPC plus a cancel func. While
// r.ctx is alive it's used directly; once cancelled (server RESULT_CANCELLED),
// RPCs switch to a fresh bounded context so Close()'s final flush still lands.
func (r *Reporter) rpcCtx() (context.Context, context.CancelFunc) {
	select {
	case <-r.ctx.Done():
		return context.WithTimeout(context.Background(), r.effectiveCloseTimeout())
	default:
		return r.ctx, func() {}
	}
}

func (r *Reporter) ReportLog(noMore bool) error {
	_, err := r.reportLog(noMore)
	return err
}

// reportLog also reports whether the server took rows it had not taken before.
func (r *Reporter) reportLog(noMore bool) (bool, error) {
	r.clientM.Lock()
	defer r.clientM.Unlock()

	r.stateMu.RLock()
	rows := r.logRows
	r.stateMu.RUnlock()

	if !noMore && len(rows) == 0 {
		return false, nil
	}

	rpcCtx, rpcCancel := r.rpcCtx()
	defer rpcCancel()

	start := time.Now()
	resp, err := r.client.UpdateLog(rpcCtx, connect.NewRequest(&runnerv1.UpdateLogRequest{
		TaskId: r.state.Id,
		Index:  int64(r.logOffset),
		Rows:   rows,
		NoMore: noMore,
	}))
	metrics.ReportLogDuration.Observe(time.Since(start).Seconds())
	if err != nil {
		metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultError).Inc()
		metrics.ClientErrors.WithLabelValues(metrics.LabelMethodUpdateLog, connect.CodeOf(err).String()).Inc()
		return false, err
	}
	metrics.ReportLogTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()

	ack := int(resp.Msg.AckIndex)
	if ack < r.logOffset {
		return false, errors.New("submitted logs are lost")
	}

	r.stateMu.Lock()
	submitted := r.logOffset + len(rows)
	// A server can ack beyond what it was sent; clamp to stay within the buffer.
	ack = min(ack, submitted)
	took := ack > r.logOffset
	r.logRows = r.logRows[ack-r.logOffset:]
	r.logOffset = ack
	remaining := len(r.logRows)
	r.stateMu.Unlock()
	if remaining != r.lastLogBufferRows {
		metrics.ReportLogBufferRows.Set(float64(remaining))
		r.lastLogBufferRows = remaining
	}

	if noMore && ack < submitted {
		return took, errors.New("not all logs are submitted")
	}

	return took, nil
}

// ReportState only reports the job result if reportResult is true
// RunDaemon never reports results even if result is set
func (r *Reporter) ReportState(reportResult bool) error {
	r.clientM.Lock()
	defer r.clientM.Unlock()

	outputs := make(map[string]string)
	r.outputsMu.Lock()
	for key, out := range r.outputs {
		if !out.sent {
			outputs[key] = out.value
		}
	}
	r.outputsMu.Unlock()

	// Consume stateChanged atomically with the snapshot; restored on error
	// below so a log record handled during UpdateTask isn't silently lost.
	// Heartbeat at stateReportInterval even when nothing changed, so the server
	// doesn't time out long-running silent jobs as orphaned (#826).
	last := r.lastReportedAtNanos.Load()
	withinHeartbeatInterval := last != 0 && time.Since(time.Unix(0, last)) < r.stateReportInterval
	r.stateMu.Lock()
	if !reportResult && !r.stateChanged && len(outputs) == 0 && withinHeartbeatInterval {
		r.stateMu.Unlock()
		return nil
	}
	state := &runnerv1.TaskState{}
	proto.Merge(state, r.state)
	r.stateChanged = false
	r.stateMu.Unlock()

	if !reportResult {
		state.Result = runnerv1.Result_RESULT_UNSPECIFIED
	}

	rpcCtx, rpcCancel := r.rpcCtx()
	defer rpcCancel()

	start := time.Now()
	resp, err := r.client.UpdateTask(rpcCtx, connect.NewRequest(&runnerv1.UpdateTaskRequest{
		State:   state,
		Outputs: outputs,
	}))
	metrics.ReportStateDuration.Observe(time.Since(start).Seconds())
	if err != nil {
		metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultError).Inc()
		metrics.ClientErrors.WithLabelValues(metrics.LabelMethodUpdateTask, connect.CodeOf(err).String()).Inc()
		r.stateMu.Lock()
		r.stateChanged = true
		r.stateMu.Unlock()
		return err
	}
	metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
	r.lastReportedAtNanos.Store(time.Now().UnixNano())

	var noSent []string
	r.outputsMu.Lock()
	for _, k := range resp.Msg.SentOutputs {
		if _, ok := r.outputs[k]; ok {
			r.outputs[k] = jobOutput{sent: true}
		}
	}
	for key, out := range r.outputs {
		if !out.sent {
			noSent = append(noSent, key)
		}
	}
	r.outputsMu.Unlock()

	// A terminal result means the server is done with this task; keep running and
	// the job holds a capacity slot until runner.timeout.
	if state := resp.Msg.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
		r.stateMu.Lock()
		r.serverResult = state.Result
		r.stateMu.Unlock()
		r.cancel()
	}
	if len(noSent) > 0 {
		return fmt.Errorf("%w: %v", errOutputsNotSent, noSent)
	}

	return nil
}

func (r *Reporter) duringSteps() bool {
	if steps := r.state.Steps; len(steps) == 0 {
		return false
	} else if first := steps[0]; first.Result == runnerv1.Result_RESULT_UNSPECIFIED && first.LogLength == 0 {
		return false
	} else if last := steps[len(steps)-1]; last.Result != runnerv1.Result_RESULT_UNSPECIFIED {
		return false
	}
	return true
}

// shouldAppendLogRow reports whether a non-raw_output entry should be written
// to the job log: only when we are between steps and the entry's level is
// within runner.log_report_level.
func (r *Reporter) shouldAppendLogRow(entry logging.Record) bool {
	return !r.duringSteps() && entry.Level >= r.logReportLevel
}

var stringToResult = map[string]runnerv1.Result{
	"success":   runnerv1.Result_RESULT_SUCCESS,
	"failure":   runnerv1.Result_RESULT_FAILURE,
	"skipped":   runnerv1.Result_RESULT_SKIPPED,
	"cancelled": runnerv1.Result_RESULT_CANCELLED,
}

func (r *Reporter) parseResult(result any) (runnerv1.Result, bool) {
	str := ""
	if v, ok := result.(string); ok { // for jobResult
		str = v
	} else if v, ok := result.(fmt.Stringer); ok { // for stepResult
		str = v.String()
	}

	ret, ok := stringToResult[str]
	return ret, ok
}

func logRow(entry logging.Record) *runnerv1.LogRow {
	return &runnerv1.LogRow{Time: timestamppb.New(entry.Time), Content: strings.TrimRight(entry.Message, "\r\n")}
}

// mask repairs the content first, so a secret the repair itself spells out is still caught.
func (r *Reporter) mask(content string) string {
	return r.logReplacer.Replace(strings.ToValidUTF8(content, "?"))
}

// AddMask deliberately leaves the process-wide masker alone. Its entries come from every live
// task at once, so a workflow could otherwise mask "error" there and rewrite the runner's own
// log, and every other task's, for the rest of the job.
func (r *Reporter) AddMask(msg string) {
	r.stateMu.Lock()
	defer r.stateMu.Unlock()
	r.oldnew = runner.AppendSecretMasker(r.oldnew, msg)
	r.logReplacer = runner.NewSecretReplacer(r.oldnew)
}