// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package finders

import (
	"context"
	"sync"
	"sync/atomic"
	"time"

	"github.com/hashicorp/go-multierror"

	v3 "skywalking.apache.org/repo/goapi/collect/ebpf/profiling/process/v3"

	"github.com/apache/skywalking-rover/pkg/core"
	"github.com/apache/skywalking-rover/pkg/module"
	"github.com/apache/skywalking-rover/pkg/process/api"
	"github.com/apache/skywalking-rover/pkg/process/finders/base"
	"github.com/apache/skywalking-rover/pkg/tools"
)

type ProcessStorage struct {
	processes map[api.ProcessDetectType][]*ProcessContext
	mutex     sync.Mutex

	// process listeners
	listeners               []api.ProcessListener
	eventQueue              chan *processEvent
	initListenQueue         chan api.ProcessListener
	listenerRecheckInterval time.Duration

	// working with backend
	reportInterval         time.Duration
	propertiesReportFactor int
	roverID                string
	processClient          v3.EBPFProcessServiceClient
	finders                map[api.ProcessDetectType]base.ProcessFinder
	reportedCount          int64

	// report context
	ctx    context.Context
	cancel context.CancelFunc
}

func NewProcessStorage(ctx context.Context, moduleManager *module.Manager, reportInterval time.Duration,
	propertiesReportFactor int, finderList []base.ProcessFinder, listenerRecheckInterval time.Duration) (*ProcessStorage, error) {
	data := make(map[api.ProcessDetectType][]*ProcessContext)
	// working with core module
	coreOperator := moduleManager.FindModule(core.ModuleName).(core.Operator)
	roverID := coreOperator.InstanceID()
	backendConn := coreOperator.BackendOperator().GetConnection()
	processClient := v3.NewEBPFProcessServiceClient(backendConn)
	ctx, cancel := context.WithCancel(ctx)
	fs := make(map[api.ProcessDetectType]base.ProcessFinder)
	for _, f := range finderList {
		fs[f.DetectType()] = f
	}
	return &ProcessStorage{
		processes:               data,
		reportInterval:          reportInterval,
		propertiesReportFactor:  propertiesReportFactor,
		eventQueue:              make(chan *processEvent, 100),
		initListenQueue:         make(chan api.ProcessListener, 100),
		listenerRecheckInterval: listenerRecheckInterval,
		reportedCount:           0,
		roverID:                 roverID,
		processClient:           processClient,
		finders:                 fs,
		ctx:                     ctx,
		cancel:                  cancel,
	}, nil
}

func (s *ProcessStorage) StartReport() {
	// for report all processes
	go func() {
		timeTicker := time.NewTicker(s.reportInterval)
		for {
			select {
			case <-timeTicker.C:
				if err := s.reportAllProcesses(); err != nil {
					log.Errorf("report all processes error: %v", err)
				}
			case <-s.ctx.Done():
				timeTicker.Stop()
				return
			}
		}
	}()

	// for start listener
	go func() {
		timeTicker := time.NewTicker(s.listenerRecheckInterval)
		for {
			select {
			case <-timeTicker.C:
				s.notifyToRecheckAllProcesses(s.listeners)
			case e := <-s.eventQueue:
				s.consumeProcessEvent(s.listeners, e)
			case l := <-s.initListenQueue:
				s.notifyToRecheckAllProcesses([]api.ProcessListener{l})
			case <-s.ctx.Done():
				timeTicker.Stop()
				return
			}
		}
	}()
}

func (s *ProcessStorage) StopReport() error {
	s.cancel()
	return nil
}

func (s *ProcessStorage) reportAllProcesses() error {
	s.mutex.Lock()
	defer s.mutex.Unlock()
	if len(s.processes) == 0 {
		return nil
	}

	// build process list(wait report or keep alive)
	waitReportProcesses := make([]*ProcessContext, 0)
	keepAliveProcesses := make([]*ProcessContext, 0)
	for _, finderProcesses := range s.processes {
		for _, p := range finderProcesses {
			switch p.syncStatus {
			case NotReport:
				waitReportProcesses = append(waitReportProcesses, p)
			case ReportSuccess:
				keepAliveProcesses = append(keepAliveProcesses, p)
			}
		}
	}

	// if rover should report the properties, then need to force remove all keep alive processes to report
	shouldReportProperties := atomic.AddInt64(&s.reportedCount, 1)%int64(s.propertiesReportFactor) == 0
	if shouldReportProperties {
		log.Infof("detection has reached the properties report factor, forced to report all processes properties")
		waitReportProcesses = append(waitReportProcesses, keepAliveProcesses...)
		keepAliveProcesses = make([]*ProcessContext, 0)
	}
	var result error
	if err := s.processesReport(waitReportProcesses); err != nil {
		result = multierror.Append(result, err)
	}
	if err := s.processesKeepAlive(keepAliveProcesses); err != nil {
		result = multierror.Append(result, err)
	}

	return result
}

func (s *ProcessStorage) processesKeepAlive(waitKeepAliveProcess []*ProcessContext) error {
	if len(waitKeepAliveProcess) == 0 {
		return nil
	}

	processIDList := make([]*v3.EBPFProcessPingPkg, 0)
	for _, ps := range waitKeepAliveProcess {
		if ps.id == "" {
			log.Warnf("the process id is not found before keep alive, need to report, pid: %d, process name: %s",
				ps.Pid(), ps.Entity().ProcessName)
			ps.syncStatus = NotReport
			continue
		}
		processIDList = append(processIDList, &v3.EBPFProcessPingPkg{
			EntityMetadata: &v3.EBPFProcessEntityMetadata{
				Layer:        ps.Entity().Layer,
				ServiceName:  ps.Entity().ServiceName,
				InstanceName: ps.Entity().InstanceName,
				ProcessName:  ps.Entity().ProcessName,
				Labels:       ps.Entity().Labels,
			},
			Properties: s.finders[ps.detectType].BuildNecessaryProperties(ps.detectProcess),
		})
	}

	_, err := s.processClient.KeepAlive(s.ctx, &v3.EBPFProcessPingPkgList{
		EbpfAgentID: s.roverID,
		Processes:   processIDList,
	})
	return err
}

func (s *ProcessStorage) processesReport(waitReportProcesses []*ProcessContext) error {
	if len(waitReportProcesses) == 0 {
		return nil
	}

	properties := make([]*v3.EBPFProcessProperties, 0)
	buildContext := &base.BuildEBPFProcessContext{}
	buildContext.HostIP = tools.DefaultHostIPAddress()
	for _, ps := range waitReportProcesses {
		properties = append(properties, s.finders[ps.DetectType()].BuildEBPFProcess(buildContext, ps.detectProcess))
	}
	processes, err := s.processClient.ReportProcesses(s.ctx, &v3.EBPFProcessReportList{Processes: properties, EbpfAgentID: s.roverID})
	if err != nil {
		return err
	}

	eventBuilder := s.newProcessEventBuilder(ProcessOperateAdd)
	for _, waitProcess := range waitReportProcesses {
		found := false
		for _, reportedProcess := range processes.GetProcesses() {
			id := s.finders[waitProcess.DetectType()].ParseProcessID(waitProcess.detectProcess, reportedProcess)
			if id == "" {
				continue
			}

			s.updateProcessToUploadSuccess(waitProcess, id)
			found = true
			eventBuilder.AddProcess(waitProcess.Pid(), waitProcess)
			break
		}

		if !found {
			s.updateProcessToUploadIgnored(waitProcess)
		}
	}

	eventBuilder.Send()
	return nil
}

func (s *ProcessStorage) AddNewProcessInFinder(finder api.ProcessDetectType, processes []api.DetectedProcess) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	addProcessBuilder := s.newProcessEventBuilder(ProcessOperateAdd)
	for _, newProcess := range processes {
		if newProcess == nil {
			continue
		}
		founded := false
		for _, existingProcess := range s.processes[finder] {
			if existingProcess.Pid() == newProcess.Pid() && existingProcess.Entity().SameWith(newProcess.Entity()) {
				founded = true
				break
			}
		}

		// if not found in existing processes, need to add this process
		if !founded {
			processContext := s.constructNewProcessContext(finder, newProcess)
			addProcessBuilder.AddProcess(newProcess.Pid(), processContext)
			s.processes[finder] = append(s.processes[finder], processContext)
			log.Infof("detected new process by add process: pid: %d, entity: %s", newProcess.Pid(), newProcess.Entity())
		}
	}
	addProcessBuilder.Send()
}

func (s *ProcessStorage) SyncAllProcessInFinder(finder api.ProcessDetectType, processes []api.DetectedProcess) {
	s.mutex.Lock()
	defer s.mutex.Unlock()

	newProcesses := make([]*ProcessContext, 0)

	existingProcesses := s.processes[finder]
	existingProcessHasFounded := make(map[*ProcessContext]bool)
	for _, p := range existingProcesses {
		existingProcessHasFounded[p] = false
	}

	addProcessBuilder := s.newProcessEventBuilder(ProcessOperateAdd)
	for _, syncProcess := range processes {
		if syncProcess == nil {
			continue
		}
		founded := false
		for _, existingProcess := range existingProcesses {
			if syncProcess.Pid() == existingProcess.Pid() && syncProcess.Entity().SameWith(existingProcess.Entity()) {
				newProcesses = append(newProcesses, existingProcess)
				existingProcessHasFounded[existingProcess] = true
				founded = true
				break
			}
		}

		// if not found in existing processes, need to add this process
		if !founded {
			processContext := s.constructNewProcessContext(finder, syncProcess)
			newProcesses = append(newProcesses, processContext)
			addProcessBuilder.AddProcess(syncProcess.Pid(), newProcesses[len(newProcesses)-1])
			log.Infof("detected new process by sync all: pid: %d, entity: %s", syncProcess.Pid(), syncProcess.Entity())
		}
	}
	addProcessBuilder.Send()

	// log the dead processes
	eventBuilder := s.newProcessEventBuilder(ProcessOperateDelete)
	for p, found := range existingProcessHasFounded {
		if found {
			continue
		}
		log.Infof("the process has been recognized as dead, so deleted. pid: %d, entity: %s, id: %s", p.Pid(), p.Entity(), p.id)
		eventBuilder.AddProcess(p.Pid(), p)
	}

	s.processes[finder] = newProcesses
	eventBuilder.Send()
}

func (s *ProcessStorage) constructNewProcessContext(finder api.ProcessDetectType, process api.DetectedProcess) *ProcessContext {
	exporsedPorts := make(map[int]bool)
	for _, p := range process.ExposePorts() {
		exporsedPorts[p] = true
	}
	return &ProcessContext{
		syncStatus:    NotReport,
		detectProcess: process,
		detectType:    finder,
		exposedPorts:  exporsedPorts,
	}
}

func (s *ProcessStorage) updateProcessToUploadSuccess(pc *ProcessContext, id string) {
	reported := pc.id == id
	pc.id = id
	pc.syncStatus = ReportSuccess
	if !reported {
		log.Infof("uploaded process pid: %d, name: %s, id: %s", pc.detectProcess.Pid(), pc.detectProcess.Entity().ProcessName, id)
	}
}

func (s *ProcessStorage) updateProcessToUploadIgnored(pc *ProcessContext) {
	pc.syncStatus = Ignore
	log.Infof("could not found the process id from upstream, pid: %d, entity: %v", pc.Pid(), pc.Entity())
}

func (s *ProcessStorage) GetAllProcesses() []api.ProcessInterface {
	result := make([]api.ProcessInterface, 0)
	for _, processList := range s.processes {
		for _, p := range processList {
			result = append(result, p)
		}
	}
	return result
}

func (s *ProcessStorage) FindAllRegisteredProcesses() []api.ProcessInterface {
	result := make([]api.ProcessInterface, 0)
	for _, processList := range s.processes {
		for _, p := range processList {
			if p.syncStatus == ReportSuccess {
				result = append(result, p)
			}
		}
	}
	return result
}

func (s *ProcessStorage) FindProcessByID(processID string) api.ProcessInterface {
	for _, finderProcesses := range s.processes {
		for _, p := range finderProcesses {
			if p.id == processID {
				return p
			}
		}
	}
	return nil
}

func (s *ProcessStorage) FindProcessByPID(pid int32) []api.ProcessInterface {
	result := make([]api.ProcessInterface, 0)
	for _, finderProcesses := range s.processes {
		for _, p := range finderProcesses {
			if p.Pid() == pid {
				result = append(result, p)
			}
		}
	}
	return result
}

func (s *ProcessStorage) AddListener(listener api.ProcessListener) {
	s.listeners = append(s.listeners, listener)
	s.initListenQueue <- listener
}

func (s *ProcessStorage) DeleteListener(listener api.ProcessListener) {
	result := make([]api.ProcessListener, 0)
	for _, l := range s.listeners {
		if l != listener {
			result = append(result, l)
		}
	}
	s.listeners = result
}

type ProcessOperate int

const (
	ProcessOperateAdd    = 1
	ProcessOperateDelete = 2
)

type processEventBuilder struct {
	processes map[int32][]api.ProcessInterface
	operate   ProcessOperate
	storage   *ProcessStorage
}

func (s *ProcessStorage) newProcessEventBuilder(operate ProcessOperate) *processEventBuilder {
	return &processEventBuilder{
		processes: make(map[int32][]api.ProcessInterface),
		operate:   operate,
		storage:   s,
	}
}

func (p *processEventBuilder) AddProcess(pid int32, pi api.ProcessInterface) {
	ps := p.processes[pid]
	ps = append(ps, pi)
	p.processes[pid] = ps
}

func (p *processEventBuilder) Send() {
	for pid, processes := range p.processes {
		p.storage.eventQueue <- &processEvent{
			pid:       pid,
			processes: processes,
			operate:   p.operate,
		}
	}
}

type processEvent struct {
	pid       int32
	processes []api.ProcessInterface
	operate   ProcessOperate
}

func (s *ProcessStorage) consumeProcessEvent(listeners []api.ProcessListener, e *processEvent) {
	for _, l := range listeners {
		if e.operate == ProcessOperateAdd {
			l.AddNewProcess(e.pid, e.processes)
		} else {
			l.RemoveProcess(e.pid, e.processes)
		}
	}
}

func (s *ProcessStorage) notifyToRecheckAllProcesses(listeners []api.ProcessListener) {
	if len(listeners) == 0 {
		return
	}
	// build all processes
	events := s.newProcessEventBuilder(ProcessOperateAdd)
	for _, pcs := range s.processes {
		for _, pc := range pcs {
			events.AddProcess(pc.Pid(), pc)
		}
	}
	for _, l := range listeners {
		l.RecheckAllProcesses(events.processes)
	}
}
