/*
 * Copyright (c) 2026 Huawei Technologies Co., Ltd.
 * openFuyao is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package analyzer

import (
	"context"
	"errors"
	"fmt"
	"net"
	"time"

	"many-core-orchestrator/pkg/api/analyzerv1"

	"google.golang.org/grpc"
	olog "gopkg.openfuyao.cn/common-modules/ologger/log"
)

type Runner struct {
	cfg       Config
	logger    *olog.Logger
	rules     *RuleConfigStore
	crUpdater CRUpdater
}

func NewRunner(cfg Config, logger *olog.Logger, rules *RuleConfigStore, crUpdater CRUpdater) *Runner {
	return &Runner{cfg: cfg, logger: logger, rules: rules, crUpdater: crUpdater}
}

func (r *Runner) Start(ctx context.Context) error {
	r.logger.Info("analyzer starting",
		"grpc_addr", r.cfg.GRPCAddr,
		"metrics_addr", r.cfg.MetricsAddr,
		"smooth_window", r.cfg.SmoothWindow.String(),
		"report_ttl", r.cfg.ReportTTL.String(),
		"cr_flush_interval", r.cfg.CRFlushInterval.String(),
		"rules_resync_interval", r.cfg.DetectionRulesResyncInterval.String(),
		"rules_namespace", r.cfg.DetectionRulesNamespace,
		"rules_name", r.cfg.DetectionRulesName,
		"rules_resource", r.cfg.DetectionRulesResource,
		"cr_update_enabled", r.cfg.CRUpdateEnabled,
	)

	store := NewNodeStateStore(r.cfg.SmoothWindow, r.cfg.ReportTTL, r.rules, r.rules)
	startCRFlushLoop(ctx, store, r.rules, r.crUpdater, r.cfg.CRFlushInterval, r.logger)

	service := NewAnalyzerService(store, r.logger)
	grpcServer := grpc.NewServer()
	analyzerv1.RegisterAnalyzerServer(grpcServer, service)

	listener, err := net.Listen("tcp", r.cfg.GRPCAddr)
	if err != nil {
		return fmt.Errorf("listen grpc: %w", err)
	}

	errCh := make(chan error, 1)
	go func() {
		r.logger.Info("analyzer grpc listening", "addr", r.cfg.GRPCAddr)
		errCh <- grpcServer.Serve(listener)
	}()

	select {
	case <-ctx.Done():
		grpcServer.GracefulStop()
		return nil
	case err := <-errCh:
		if errors.Is(err, grpc.ErrServerStopped) {
			return nil
		}
		return err
	}
}

func startCRFlushLoop(ctx context.Context, store *NodeStateStore, rules *RuleConfigStore, crUpdater CRUpdater, interval time.Duration, logger *olog.Logger) {
	if interval <= 0 {
		interval = time.Second
	}
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				report, ok := store.ConsumeDirtyClusterReportWithRuleConfig(rules.Metadata())
				if !ok {
					continue
				}
				if err := crUpdater.Update(ctx, report); err != nil {
					store.MarkDirty()
					crUpdateTotal.WithLabelValues("error").Inc()
					logger.Error("update CR failed",
						"error", err,
						"nodes", len(report.Nodes),
						"cr_flush_interval", interval.String(),
					)
				} else {
					crUpdateTotal.WithLabelValues("ok").Inc()
					crLastSuccessTimestamp.Set(float64(time.Now().Unix()))
				}
			}
		}
	}()
}