package log_scanner

import (
	"context"
	"io"
	"sync"

	"github.com/pingcap-inc/alioth/pkg/util"
	"github.com/pingcap-inc/alioth/pkg/worker"
	logpattern_go_proto "github.com/pingcap-inc/alioth/proto"
	recorder "github.com/pingcap-inc/alioth/recorder"
	matcher "github.com/pingcap-inc/alioth/scanner/log_match"
	scanner "github.com/pingcap-inc/alioth/scanner/log_scan"
	"github.com/pingcap-inc/alioth/storage"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

// LogProcessor is responsible for scanning the log,
// looking for the matching log pattern,
// and then recording the coverage or unknown log (not extracted by log cov)
type LogProcessor struct {
	store *storage.LogPatternStore

	matcher    *matcher.PatternMatcher
	scannerSet []*scanner.LogScanner

	coverager  *recorder.Coverage
	unknowLogs *recorder.UnknowLogRecord
}

func NewLogProcessor(store *storage.LogPatternStore, logPaths []string) (*LogProcessor, error) {
	matcher, err := matcher.NewPatternMatcher(store)
	if err != nil {
		return nil, err
	}

	coverager := recorder.NewCoverager(store)
	unknowLogs := recorder.NewUnknowLogRecorder(store)

	scannerSet := make([]*scanner.LogScanner, 0, len(logPaths))
	for _, path := range logPaths {
		scanner, err := scanner.NewLogScanner(path)
		if err != nil {
			return nil, err
		}
		scannerSet = append(scannerSet, scanner)
	}

	return &LogProcessor{
		store:      store,
		scannerSet: scannerSet,
		coverager:  coverager,
		matcher:    matcher,
		unknowLogs: unknowLogs,
	}, nil
}

// Run starts the log processing
func (s *LogProcessor) Run(ctx context.Context, rule *logpattern_go_proto.LogPatternRule, pw *worker.Pool) error {
	if err := s.coverager.LoadCoverage(ctx); err != nil {
		return err
	}

	if err := s.unknowLogs.LoadUnknowLogs(ctx); err != nil {
		return err
	}

	//pipeline := NewPipiline(1024)
	wg := sync.WaitGroup{}

	for _, scanner := range s.scannerSet {
		line := newAssemLine(s.matcher, scanner, s.coverager, s.unknowLogs)
		wg.Add(1)
		go func(l *assemLine) {
			wk := pw.Apply()
			if err := l.run(ctx, rule); err != nil {
				log.Error("process log line failed", zap.String("log file", l.scanner.GetLogPath()), zap.Error(err))
			}
			pw.Recycle(wk)
			wg.Done()
		}(line)
	}

	wg.Wait()
	if err := s.unknowLogs.Flush(ctx); err != nil {
		log.Error("flush unknown log failed", zap.Error(err))
		return err
	}
	if err := s.coverager.Flush(ctx); err != nil {
		log.Error("flush coverge data failed", zap.Error(err))
		return err
	}

	return nil
}

// assemLine is a separate logics processing flow,
// including log scan, log matching, and coverage recording process unit
type assemLine struct {
	matcher    *matcher.PatternMatcher
	scanner    *scanner.LogScanner
	coverager  *recorder.Coverage
	unknowLogs *recorder.UnknowLogRecord
}

func newAssemLine(matcher *matcher.PatternMatcher, scanner *scanner.LogScanner, coverager *recorder.Coverage, unknowLogs *recorder.UnknowLogRecord) *assemLine {
	return &assemLine{
		matcher:    matcher,
		scanner:    scanner,
		coverager:  coverager,
		unknowLogs: unknowLogs,
	}
}

// run starts and runs this processing logic line
func (l *assemLine) run(ctx context.Context, rule *logpattern_go_proto.LogPatternRule) error {
	pipeline := NewPipiline(1024)
	wg := sync.WaitGroup{}

	lineCtx, cancel := context.WithCancel(ctx)
	var (
		retErr error
	)

	wg.Add(1)
	go func() {
		err1 := l.runLogScan(lineCtx, pipeline)
		if err1 != nil {
			log.Error("scan log failed", zap.String("log file", l.scanner.GetLogPath()), zap.Error(err1))
		}
		retErr = err1
		pipeline.Close()
		wg.Done()
	}()

	err := l.runMathAndRecord(ctx, pipeline, rule)
	cancel()
	wg.Wait()

	if err != nil && err != context.DeadlineExceeded {
		log.Error("match log pattern failed", zap.String("log file", l.scanner.GetLogPath()), zap.Error(err))
		retErr = err
	}

	return retErr
}

// runLogScan scans log and send log to pipeline
func (l *assemLine) runLogScan(ctx context.Context, pipeline *Pipeline) error {
	for {
		lg, err := l.scanner.Scan()
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}

		err = pipeline.Write(ctx, &Payload{log: lg})
		if err != nil {
			return err
		}
	}
}

// runMathAndRecord recieve log from pipeline and match them with log pattern. then record the coverage
func (l *assemLine) runMathAndRecord(ctx context.Context, pipeline *Pipeline, rule *logpattern_go_proto.LogPatternRule) error {
	for {
		payload, err := pipeline.Read(ctx)
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}

		res := l.matcher.Match(payload.log)
		if res.Empty() && util.MatchLogPatternRule(rule, payload.log.Level) {
			l.unknowLogs.Record(payload.log.LogPath, payload.log.File, payload.log.Line, payload.log.String(), payload.log.Level)
		} else {
			for _, lp := range res.Patterns {
				l.coverager.Record(payload.log.LogPath, payload.log.String(), lp.Pattern())
			}

			for _, lp := range res.WeakPatterns {
				log.Warn("weak log pattern", zap.Any("pattern", lp), zap.Any("log", payload.log))
				l.coverager.Record(payload.log.LogPath, payload.log.String(), lp.Pattern())
			}
		}
	}
}