package main

import (
	"context"
	"os/signal"
	"syscall"
	"testing"
	"time"

	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/config"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/logutil"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/metrics"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/process"
	"github.com/tidbcloud/observability/cmd/diagnosis-compute/pkg/types"
	"go.uber.org/zap"
)

func init() {
	if !testing.Testing() {
		config.InitFlags()
		metrics.InitMetrics()
		logutil.InitLogProcessors()
	}
}

func main() {
	if err := process.CleanupTempFiles(config.LocalMergedFileDir); err != nil {
		log.Error("failed to cleanup temp files", zap.Error(err))
	}

	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer cancel()
	log.Info("diagnosis-compute is running")
	if err := run(ctx); err != nil {
		log.Panic("run failed", zap.Error(err))
	}
}

func run(ctx context.Context) error {
	store, err := process.LoadStorage(ctx)
	if err != nil {
		return err
	}

	if config.Spec && !config.SpecAll {
		cctx, cancel := context.WithCancel(ctx)
		go process.RunCluster(cctx, store, types.ClusterInfo{
			Type:      config.SpecClusterType,
			Env:       config.Env,
			TenantID:  config.SpecTenantID,
			ProjectID: config.SpecProjectID,
			ClusterID: config.SpecClusterID,
			StorageID: config.SpecStorageID,
		})
		<-ctx.Done()
		cancel()
		metrics.CurrentAnalyzeTasks.DeleteLabelValues(config.SpecStorageID)
		return nil
	}

	go process.RunCluster(ctx, store, types.ClusterInfo{
		Type:      1,
		Env:       config.Env,
		TenantID:  "infra-prod",
		ProjectID: "seed",
		ClusterID: "k8s-infra",
		StorageID: "seed",
	})

	go process.RunCluster(ctx, store, types.ClusterInfo{
		Type:      1,
		Env:       config.Env,
		TenantID:  "infra-prod",
		ProjectID: "o11y",
		ClusterID: "k8s-infra",
		StorageID: "o11y",
	})

	go process.RunCluster(ctx, store, types.ClusterInfo{
		Type:      1,
		Env:       config.Env,
		TenantID:  "infra-prod",
		ProjectID: "base",
		ClusterID: "k8s-infra",
		StorageID: "base",
	})

	bucket := config.BucketName()
	clusterList := map[string]types.ClusterInfo{}
	clusterCancel := map[string]context.CancelFunc{}
	metas, err := store.Objects(ctx, bucket, "diagnosis/tasks/processor-dedicated/")
	if err != nil {
		return err
	}
	for _, meta := range metas {
		info, err := process.LoadClusterInfo(ctx, store, bucket, meta.Key)
		if err != nil {
			log.Error("failed to load cluster info", zap.Error(err))
			continue
		}
		cctx, cancel := context.WithCancel(context.Background())
		go process.RunCluster(cctx, store, *info)
		clusterList[meta.Key] = *info
		clusterCancel[meta.Key] = func() {
			cancel()
			metrics.DeleteMetrics(*info)
		}
	}

	missedCluster := map[string]int{}
	ticker := time.NewTicker(1 * time.Hour)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			for _, cancel := range clusterCancel {
				cancel()
			}
			return nil
		case <-ticker.C:
			metas, err := store.Objects(ctx, bucket, "diagnosis/tasks/processor-dedicated/")
			if err != nil {
				log.Error("failed to fetch cluster list", zap.Error(err))
				continue
			}
			keys := map[string]struct{}{}
			added := []string{}
			for _, meta := range metas {
				keys[meta.Key] = struct{}{}
				if _, ok := clusterList[meta.Key]; !ok {
					added = append(added, meta.Key)
				}
			}
			removed := []string{}
			for key := range clusterList {
				if _, ok := keys[key]; !ok {
					val, ok := missedCluster[key]
					// wait 3 times to ensure all logs are collected
					if !ok {
						missedCluster[key] = 1
					} else if val < 3 {
						missedCluster[key] = val + 1
					} else {
						removed = append(removed, key)
						delete(missedCluster, key)
					}
				}
			}
			for _, key := range added {
				info, err := process.LoadClusterInfo(ctx, store, bucket, key)
				if err != nil {
					log.Warn("failed to fetch cluster object", zap.Error(err))
					continue
				}
				cctx, cancel := context.WithCancel(context.Background())
				go process.RunCluster(cctx, store, *info)
				clusterList[key] = *info
				clusterCancel[key] = cancel
			}
			for _, key := range removed {
				clusterCancel[key]()
				delete(clusterList, key)
				delete(clusterCancel, key)
			}
		}
	}
}