cthulhu/internal/logging/async.go

// Cthulhu screen reader.
// Copyright (c) 2026 Storm Dragon
// Authors: Storm Dragon, Angus Kola
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along with
// this program. If not, see <https://www.gnu.org/licenses/gpl-3.0.html>.

package logging

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"sync"
	"sync/atomic"
	"time"
)

const (
	defaultDiagnosticsCapacity = 256
	defaultStopTimeout         = 2 * time.Second
	defaultCancelGrace         = 100 * time.Millisecond
)

type DiagnosticsWriter interface {
	Write([]byte) (int, error)
	Cancel() error
	Close() error
}

type AsyncOptions struct {
	Options
	Capacity    int
	StopTimeout time.Duration
	CancelGrace time.Duration
}

type diagnosticRecord struct {
	data     []byte
	critical bool
}

type Diagnostics struct {
	writer      DiagnosticsWriter
	formatter   *Handler
	logger      *slog.Logger
	queue       []diagnosticRecord
	wake        chan struct{}
	stopSignal  chan struct{}
	done        chan struct{}
	mutex       sync.Mutex
	stopOnce    sync.Once
	errorsMutex sync.Mutex
	stopMutex   sync.Mutex
	writeErrors []error
	stopError   error
	stopSet     bool
	dropped     atomic.Uint64
	reported    atomic.Uint64
	capacity    int
	stopTimeout time.Duration
	cancelGrace time.Duration
}

func NewAsync(writer DiagnosticsWriter, options AsyncOptions) *Diagnostics {
	if writer == nil {
		writer = NewPassthroughWriter(nil)
	}
	if options.Capacity <= 0 {
		options.Capacity = defaultDiagnosticsCapacity
	}
	if options.StopTimeout <= 0 {
		options.StopTimeout = defaultStopTimeout
	}
	if options.CancelGrace <= 0 {
		options.CancelGrace = defaultCancelGrace
	}
	diagnostics := &Diagnostics{
		writer:      writer,
		formatter:   newHandler(nil, options.Options),
		queue:       make([]diagnosticRecord, 0, options.Capacity),
		wake:        make(chan struct{}, 1),
		stopSignal:  make(chan struct{}),
		done:        make(chan struct{}),
		capacity:    options.Capacity,
		stopTimeout: options.StopTimeout,
		cancelGrace: options.CancelGrace,
	}
	diagnostics.logger = slog.New(&asyncHandler{
		diagnostics: diagnostics,
		base:        diagnostics.formatter,
	})
	go diagnostics.run()
	return diagnostics
}

func (diagnostics *Diagnostics) Logger() *slog.Logger {
	return diagnostics.logger
}

func (diagnostics *Diagnostics) Dropped() uint64 {
	return diagnostics.dropped.Load()
}

func (diagnostics *Diagnostics) Stop(ctx context.Context) error {
	if ctx == nil {
		return errors.New("diagnostics stop requires a non-nil context")
	}
	diagnostics.signalStop()

	stopContext, cancel := context.WithTimeout(ctx, diagnostics.stopTimeout)
	defer cancel()
	if !diagnostics.waitOrCancel(stopContext) {
		diagnostics.setStopError(fmt.Errorf("flush diagnostics: %w", stopContext.Err()))
		return diagnostics.stopErrorValue()
	}
	diagnostics.setStopError(errors.Join(diagnostics.writeErrorsValue()...))
	return diagnostics.stopErrorValue()
}

func (diagnostics *Diagnostics) signalStop() {
	diagnostics.stopOnce.Do(func() {
		close(diagnostics.stopSignal)
		diagnostics.wakeup()
	})
}

func (diagnostics *Diagnostics) waitOrCancel(ctx context.Context) bool {
	deadline, hasDeadline := ctx.Deadline()
	if hasDeadline {
		cancelDelay := time.Until(deadline) - diagnostics.cancelGrace
		if cancelDelay > 0 {
			timer := time.NewTimer(cancelDelay)
			select {
			case <-diagnostics.done:
				timer.Stop()
				return true
			case <-timer.C:
			}
		}
	} else {
		select {
		case <-diagnostics.done:
			return true
		case <-time.After(diagnostics.cancelGrace):
		}
	}
	if err := diagnostics.writer.Cancel(); err != nil {
		diagnostics.appendWriteError(fmt.Errorf("cancel diagnostics writer: %w", err))
	}
	select {
	case <-diagnostics.done:
		return true
	case <-ctx.Done():
		return false
	}
}

func (diagnostics *Diagnostics) run() {
	defer close(diagnostics.done)
	for {
		diagnostics.mutex.Lock()
		if len(diagnostics.queue) == 0 {
			if !diagnostics.accepting() {
				diagnostics.mutex.Unlock()
				diagnostics.reportDrops()
				diagnostics.mutex.Lock()
				if len(diagnostics.queue) == 0 {
					diagnostics.mutex.Unlock()
					break
				}
			} else {
				wake := diagnostics.wake
				diagnostics.mutex.Unlock()
				select {
				case <-wake:
				case <-diagnostics.stopSignal:
				}
				continue
			}
		}
		record := diagnostics.queue[0]
		diagnostics.queue = diagnostics.queue[1:]
		diagnostics.mutex.Unlock()

		if _, err := diagnostics.writer.Write(record.data); err != nil {
			diagnostics.appendWriteError(fmt.Errorf("write diagnostics: %w", err))
		}
		diagnostics.reportDrops()
	}
	if err := diagnostics.writer.Close(); err != nil {
		diagnostics.appendWriteError(fmt.Errorf("close diagnostics writer: %w", err))
	}
}

func (diagnostics *Diagnostics) accepting() bool {
	select {
	case <-diagnostics.stopSignal:
		return false
	default:
		return true
	}
}

func (diagnostics *Diagnostics) enqueue(record diagnosticRecord) {
	diagnostics.mutex.Lock()
	if !diagnostics.accepting() {
		diagnostics.mutex.Unlock()
		diagnostics.dropped.Add(1)
		return
	}
	if len(diagnostics.queue) == diagnostics.capacity {
		diagnostics.dropped.Add(1)
		var admitted bool
		diagnostics.queue, admitted = makeRoom(diagnostics.queue, record.critical)
		if !admitted {
			diagnostics.mutex.Unlock()
			return
		}
	}
	diagnostics.queue = append(diagnostics.queue, record)
	diagnostics.mutex.Unlock()
	diagnostics.wakeup()
}

func makeRoom(queue []diagnosticRecord, incomingCritical bool) ([]diagnosticRecord, bool) {
	for index, record := range queue {
		if !record.critical {
			copy(queue[index:], queue[index+1:])
			queue[len(queue)-1] = diagnosticRecord{}
			return queue[:len(queue)-1], true
		}
	}
	if !incomingCritical {
		return queue, false
	}
	copy(queue, queue[1:])
	queue[len(queue)-1] = diagnosticRecord{}
	return queue[:len(queue)-1], true
}

func (diagnostics *Diagnostics) reportDrops() {
	total := diagnostics.dropped.Load()
	reported := diagnostics.reported.Swap(total)
	if total == reported {
		return
	}
	record := slog.NewRecord(time.Now(), slog.LevelWarn, "diagnostics queue overflow", 0)
	record.AddAttrs(
		slog.Uint64("dropped_messages", total-reported),
		slog.Uint64("total_dropped_messages", total),
	)
	if _, err := diagnostics.writer.Write(diagnostics.formatter.format(record)); err != nil {
		diagnostics.appendWriteError(fmt.Errorf("write diagnostics overflow report: %w", err))
	}
}

func (diagnostics *Diagnostics) wakeup() {
	select {
	case diagnostics.wake <- struct{}{}:
	default:
	}
}

func (diagnostics *Diagnostics) appendWriteError(err error) {
	diagnostics.errorsMutex.Lock()
	defer diagnostics.errorsMutex.Unlock()
	diagnostics.writeErrors = append(diagnostics.writeErrors, err)
}

func (diagnostics *Diagnostics) writeErrorsValue() []error {
	diagnostics.errorsMutex.Lock()
	defer diagnostics.errorsMutex.Unlock()
	return append([]error(nil), diagnostics.writeErrors...)
}

func (diagnostics *Diagnostics) setStopError(err error) {
	diagnostics.stopMutex.Lock()
	defer diagnostics.stopMutex.Unlock()
	if !diagnostics.stopSet {
		diagnostics.stopError = err
		diagnostics.stopSet = true
	}
}

func (diagnostics *Diagnostics) stopErrorValue() error {
	diagnostics.stopMutex.Lock()
	defer diagnostics.stopMutex.Unlock()
	return diagnostics.stopError
}

type asyncHandler struct {
	diagnostics *Diagnostics
	base        *Handler
}

func (handler *asyncHandler) Enabled(ctx context.Context, level slog.Level) bool {
	return handler.base.Enabled(ctx, level)
}

func (handler *asyncHandler) Handle(_ context.Context, record slog.Record) error {
	if !handler.base.Enabled(context.Background(), record.Level) {
		return nil
	}
	handler.diagnostics.enqueue(diagnosticRecord{
		data:     handler.base.format(record),
		critical: record.Level >= slog.LevelWarn,
	})
	return nil
}

func (handler *asyncHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
	clone := *handler
	clone.base = handler.base.WithAttrs(attrs).(*Handler)
	return &clone
}

func (handler *asyncHandler) WithGroup(name string) slog.Handler {
	clone := *handler
	clone.base = handler.base.WithGroup(name).(*Handler)
	return &clone
}

type passthroughWriter struct {
	writer io.Writer
}

func NewPassthroughWriter(writer io.Writer) DiagnosticsWriter {
	if writer == nil {
		writer = io.Discard
	}
	return passthroughWriter{writer: writer}
}

func (writer passthroughWriter) Write(data []byte) (int, error) {
	return writer.writer.Write(data)
}

func (writer passthroughWriter) Cancel() error { return nil }

func (writer passthroughWriter) Close() error { return nil }

type fileWriter struct {
	file      *os.File
	fallback  io.Writer
	closeOnce sync.Once
}

func OpenFile(path string, fallback io.Writer) (DiagnosticsWriter, error) {
	if path == "" || path == "." || path == ".." {
		return nil, errors.New("invalid diagnostics file path")
	}
	file, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
	if err != nil {
		return nil, fmt.Errorf("open diagnostics file: %w", err)
	}
	return &fileWriter{file: file, fallback: fallback}, nil
}

func (writer *fileWriter) Write(data []byte) (int, error) {
	if writer.fallback != nil {
		if _, err := writer.fallback.Write(data); err != nil {
			return 0, fmt.Errorf("write diagnostics fallback: %w", err)
		}
	}
	return writer.file.Write(data)
}

func (writer *fileWriter) Cancel() error {
	return writer.close()
}

func (writer *fileWriter) Close() error {
	return writer.close()
}

func (writer *fileWriter) close() error {
	var closeError error
	writer.closeOnce.Do(func() {
		closeError = writer.file.Close()
	})
	if errors.Is(closeError, os.ErrClosed) {
		return nil
	}
	return closeError
}