bee/audit/internal/webui/jobs.go

package webui

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"strings"
	"sync"
	"time"
)

// jobState holds the output lines and completion status of an async job.
type jobState struct {
	lines        []string
	done         bool
	err          string
	mu           sync.Mutex
	subs         []chan string
	cancel       func() // optional cancel function; nil if job is not cancellable
	logPath      string
	serialPrefix string
	logFile      *os.File    // kept open for the task lifetime to avoid per-line open/close
	logBuf       *bufio.Writer
}

// readTaskLogFile reads a task log, refusing files over 50 MB.
func readTaskLogFile(path string) ([]byte, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	data, err := io.ReadAll(io.LimitReader(f, 50<<20+1))
	if err != nil {
		return nil, err
	}
	if int64(len(data)) > 50<<20 {
		return nil, fmt.Errorf("task log %s too large (exceeds 50 MB)", path)
	}
	return data, nil
}

// abort cancels the job if it has a cancel function and is not yet done.
func (j *jobState) abort() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.done || j.cancel == nil {
		return false
	}
	j.cancel()
	return true
}

func (j *jobState) append(line string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.lines = append(j.lines, line)
	if j.logPath != "" {
		j.writeLogLineLocked(line)
	}
	if j.serialPrefix != "" {
		taskSerialWriteLine(j.serialPrefix + line)
	}
	for _, ch := range j.subs {
		select {
		case ch <- line:
		default:
		}
	}
}

// writeLogLineLocked writes a line to the persistent log file, opening it lazily.
// Must be called with j.mu held. Uses a buffered writer kept open for the task
// lifetime — avoids thousands of open/close syscalls during high-frequency logs.
func (j *jobState) writeLogLineLocked(line string) {
	if j.logFile == nil {
		f, err := os.OpenFile(j.logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
		if err != nil {
			return
		}
		j.logFile = f
		j.logBuf = bufio.NewWriterSize(f, 64*1024)
	}
	_, _ = j.logBuf.WriteString(line + "\n")
}

// closeLog flushes and closes the log file. Called after all task output is done.
func (j *jobState) closeLog() {
	j.mu.Lock()
	defer j.mu.Unlock()
	if j.logBuf != nil {
		_ = j.logBuf.Flush()
	}
	if j.logFile != nil {
		_ = j.logFile.Close()
		j.logFile = nil
		j.logBuf = nil
	}
}

func (j *jobState) finish(errMsg string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	j.done = true
	j.err = errMsg
	for _, ch := range j.subs {
		close(ch)
	}
	j.subs = nil
}

// subscribe returns a channel that receives all future lines.
// Existing lines are returned first, then the channel streams new ones.
func (j *jobState) subscribe() ([]string, <-chan string) {
	j.mu.Lock()
	defer j.mu.Unlock()
	existing := make([]string, len(j.lines))
	copy(existing, j.lines)
	if j.done {
		return existing, nil
	}
	ch := make(chan string, 256)
	j.subs = append(j.subs, ch)
	return existing, ch
}

// jobManager manages async jobs identified by string IDs.
type jobManager struct {
	mu   sync.Mutex
	jobs map[string]*jobState
}

var globalJobs = &jobManager{jobs: make(map[string]*jobState)}

func (m *jobManager) create(id string) *jobState {
	m.mu.Lock()
	defer m.mu.Unlock()
	j := &jobState{}
	m.jobs[id] = j
	// Schedule cleanup after 30 minutes
	goRecoverOnce("job cleanup", func() {
		time.Sleep(30 * time.Minute)
		m.mu.Lock()
		delete(m.jobs, id)
		m.mu.Unlock()
	})
	return j
}

// isDone returns true if the job has finished (either successfully or with error).
func (j *jobState) isDone() bool {
	j.mu.Lock()
	defer j.mu.Unlock()
	return j.done
}

func (m *jobManager) get(id string) (*jobState, bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	j, ok := m.jobs[id]
	return j, ok
}

func newTaskJobState(logPath string, serialPrefix ...string) *jobState {
	j := &jobState{logPath: logPath}
	if len(serialPrefix) > 0 {
		j.serialPrefix = serialPrefix[0]
	}
	if logPath == "" {
		return j
	}
	data, err := readTaskLogFile(logPath)
	if err != nil || len(data) == 0 {
		return j
	}
	lines := strings.Split(strings.ReplaceAll(string(data), "\r\n", "\n"), "\n")
	if len(lines) > 0 && lines[len(lines)-1] == "" {
		lines = lines[:len(lines)-1]
	}
	j.lines = append(j.lines, lines...)
	return j
}

func appendJobLog(path, line string) {
	if path == "" {
		return
	}
	f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
	if err != nil {
		return
	}
	defer f.Close()
	_, _ = f.WriteString(line + "\n")
}