// 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 server provides HTTP server functionality for exposing Prometheus metrics.
package server

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"

	"github.com/apache/skywalking-banyandb/fodc/agent/internal/exporter"
	"github.com/apache/skywalking-banyandb/pkg/logger"
)

// CollectionProvider exposes collected crash artifact data.
type CollectionProvider interface {
	MarshalCollections() ([]byte, error)
}

const (
	defaultReadHeaderTimeout = 3 * time.Second
	defaultShutdownTimeout   = 5 * time.Second
)

// Config holds configuration for the Prometheus metrics server.
type Config struct {
	ListenAddr          string
	DiagnosisListenAddr string
	ReadHeaderTimeout   time.Duration
	ShutdownTimeout     time.Duration
}

// Server represents a Prometheus metrics HTTP server.
type Server struct {
	server          *http.Server
	diagnosisServer *http.Server
	listener        net.Listener
	diagnosisLn     net.Listener
	logger          *logger.Logger
	config          Config
	started         bool
}

// NewServer creates a new Prometheus metrics server with the given configuration.
func NewServer(config Config) (*Server, error) {
	if config.ListenAddr == "" {
		return nil, fmt.Errorf("listen address cannot be empty")
	}
	if config.DiagnosisListenAddr == "" {
		config.DiagnosisListenAddr = config.ListenAddr
	}
	if config.ReadHeaderTimeout == 0 {
		config.ReadHeaderTimeout = defaultReadHeaderTimeout
	}
	if config.ShutdownTimeout == 0 {
		config.ShutdownTimeout = defaultShutdownTimeout
	}

	return &Server{
		config: config,
		logger: logger.GetLogger("server"),
	}, nil
}

// Start starts the Prometheus metrics server with the provided registry.
// It registers the DatasourceCollector and starts serving metrics on the configured address.
func (s *Server) Start(
	registry *prometheus.Registry,
	datasourceCollector *exporter.DatasourceCollector,
	collectionProvider CollectionProvider,
) (<-chan error, error) {
	if s.started {
		return nil, fmt.Errorf("server is already started")
	}

	// Register the datasource collector with the registry
	if registerErr := registry.Register(datasourceCollector); registerErr != nil {
		return nil, fmt.Errorf("failed to register DatasourceCollector: %w", registerErr)
	}

	metricsMux := http.NewServeMux()
	metricsMux.Handle("/metrics", promhttp.HandlerFor(
		registry,
		promhttp.HandlerOpts{},
	))
	diagnosisMux := http.NewServeMux()
	if collectionProvider != nil {
		diagnosisMux.HandleFunc("/diagnostics", diagnosticsHandler(collectionProvider))
	}

	if s.config.DiagnosisListenAddr == s.config.ListenAddr {
		combinedMux := http.NewServeMux()
		combinedMux.Handle("/metrics", promhttp.HandlerFor(
			registry,
			promhttp.HandlerOpts{},
		))
		if collectionProvider != nil {
			combinedMux.HandleFunc("/diagnostics", diagnosticsHandler(collectionProvider))
		}

		listener, listenErr := net.Listen("tcp", s.config.ListenAddr)
		if listenErr != nil {
			return nil, fmt.Errorf("failed to listen on metrics address: %w", listenErr)
		}

		s.server = &http.Server{
			Addr:              s.config.ListenAddr,
			Handler:           combinedMux,
			ReadHeaderTimeout: s.config.ReadHeaderTimeout,
		}
		s.diagnosisServer = s.server
		s.listener = listener
		s.diagnosisLn = listener

		serverErrCh := make(chan error, 1)
		go func() {
			s.logger.Info().
				Str("listen-addr", listener.Addr().String()).
				Msg("Starting Prometheus metrics server")
			if serveErr := s.server.Serve(listener); serveErr != nil && serveErr != http.ErrServerClosed {
				serverErrCh <- fmt.Errorf("metrics server error: %w", serveErr)
			}
		}()

		s.started = true
		return serverErrCh, nil
	}

	listener, listenErr := net.Listen("tcp", s.config.ListenAddr)
	if listenErr != nil {
		return nil, fmt.Errorf("failed to listen on metrics address: %w", listenErr)
	}
	diagnosisListener, diagnosisListenErr := net.Listen("tcp", s.config.DiagnosisListenAddr)
	if diagnosisListenErr != nil {
		_ = listener.Close()
		return nil, fmt.Errorf("failed to listen on diagnosis address: %w", diagnosisListenErr)
	}

	s.server = &http.Server{
		Addr:              s.config.ListenAddr,
		Handler:           metricsMux,
		ReadHeaderTimeout: s.config.ReadHeaderTimeout,
	}
	s.diagnosisServer = &http.Server{
		Addr:              s.config.DiagnosisListenAddr,
		Handler:           diagnosisMux,
		ReadHeaderTimeout: s.config.ReadHeaderTimeout,
	}
	s.listener = listener
	s.diagnosisLn = diagnosisListener

	// Start server in goroutine
	serverErrCh := make(chan error, 2)
	go func() {
		s.logger.Info().
			Str("listen-addr", listener.Addr().String()).
			Msg("Starting Prometheus metrics server")
		if serveErr := s.server.Serve(listener); serveErr != nil && serveErr != http.ErrServerClosed {
			serverErrCh <- fmt.Errorf("metrics server error: %w", serveErr)
		}
	}()
	go func() {
		s.logger.Info().
			Str("listen-addr", diagnosisListener.Addr().String()).
			Msg("Starting diagnosis server")
		if serveErr := s.diagnosisServer.Serve(diagnosisListener); serveErr != nil && serveErr != http.ErrServerClosed {
			serverErrCh <- fmt.Errorf("diagnosis server error: %w", serveErr)
		}
	}()

	s.started = true
	return serverErrCh, nil
}

// Stop gracefully stops the Prometheus metrics server.
// It waits for ongoing requests to complete within the configured shutdown timeout.
func (s *Server) Stop(ctx context.Context) error {
	if !s.started {
		return fmt.Errorf("server is not started")
	}

	s.logger.Info().Msg("Stopping Prometheus metrics server")

	shutdownCtx, shutdownCancel := context.WithTimeout(ctx, s.config.ShutdownTimeout)
	defer shutdownCancel()

	if shutdownErr := s.server.Shutdown(shutdownCtx); shutdownErr != nil {
		s.logger.Warn().Err(shutdownErr).Msg("Error shutting down metrics server")
		return shutdownErr
	}
	if s.diagnosisServer != nil && s.diagnosisServer != s.server {
		if shutdownErr := s.diagnosisServer.Shutdown(shutdownCtx); shutdownErr != nil {
			s.logger.Warn().Err(shutdownErr).Msg("Error shutting down diagnosis server")
			return shutdownErr
		}
	}

	s.started = false
	s.logger.Info().Msg("Prometheus metrics server stopped")
	return nil
}

// IsStarted returns true if the server has been started.
func (s *Server) IsStarted() bool {
	return s.started
}

// GetListenAddr returns the configured listen address.
func (s *Server) GetListenAddr() string {
	if s.listener != nil {
		return s.listener.Addr().String()
	}
	return s.config.ListenAddr
}

// GetDiagnosisListenAddr returns the configured diagnosis endpoint listen address.
func (s *Server) GetDiagnosisListenAddr() string {
	if s.diagnosisLn != nil {
		return s.diagnosisLn.Addr().String()
	}
	return s.config.DiagnosisListenAddr
}

func diagnosticsHandler(collectionProvider CollectionProvider) http.HandlerFunc {
	return func(writer http.ResponseWriter, request *http.Request) {
		if request.Method != http.MethodGet {
			writer.WriteHeader(http.StatusMethodNotAllowed)
			return
		}
		writer.Header().Set("Content-Type", "application/json")
		payload, payloadErr := collectionProvider.MarshalCollections()
		if payloadErr != nil {
			http.Error(writer, payloadErr.Error(), http.StatusInternalServerError)
			return
		}
		if _, writeErr := writer.Write(payload); writeErr != nil {
			http.Error(writer, writeErr.Error(), http.StatusInternalServerError)
		}
	}
}
