// 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 schemaserver implements a standalone gRPC server for schema property management.
package schemaserver

import (
	"context"
	"fmt"
	"net"
	"path/filepath"
	"runtime/debug"
	"strconv"
	"sync"
	"time"

	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/recovery"
	grpc_validator "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/validator"
	"github.com/pkg/errors"
	grpclib "google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/status"

	"github.com/apache/skywalking-banyandb/api/common"
	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	schemav1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/schema/v1"
	"github.com/apache/skywalking-banyandb/banyand/backup/snapshot"
	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
	"github.com/apache/skywalking-banyandb/banyand/observability"
	"github.com/apache/skywalking-banyandb/banyand/property/db"
	"github.com/apache/skywalking-banyandb/banyand/property/gossip"
	"github.com/apache/skywalking-banyandb/pkg/fs"
	"github.com/apache/skywalking-banyandb/pkg/logger"
	"github.com/apache/skywalking-banyandb/pkg/panicdiag"
	banyandbpath "github.com/apache/skywalking-banyandb/pkg/path"
	"github.com/apache/skywalking-banyandb/pkg/run"
	pkgtls "github.com/apache/skywalking-banyandb/pkg/tls"
)

const (
	defaultShardID  = common.ShardID(0)
	defaultRecvSize = 10 << 20
)

var (
	_ run.PreRunner = (*server)(nil)
	_ run.Config    = (*server)(nil)
	_ run.Service   = (*server)(nil)
	_ run.Role      = (*server)(nil)
)

// Server is the interface for the standalone schema server.
type Server interface {
	run.PreRunner
	run.Config
	run.Service
	GetPort() *uint32
	RegisterGossip(messenger gossip.Messenger)
	TakeSnapshot(ctx context.Context) *databasev1.Snapshot
}

type server struct {
	db                       db.Database
	lfs                      fs.FileSystem
	omr                      observability.MetricsRegistry
	closer                   *run.Closer
	l                        *logger.Logger
	ser                      *grpclib.Server
	tlsReloader              *pkgtls.Reloader
	watchers                 *watcherManager
	schemaService            *schemaManagementServer
	updateService            *schemaUpdateServer
	host                     string
	certFile                 string
	root                     string
	keyFile                  string
	addr                     string
	repairBuildTreeCron      string
	snapshotDir              string
	minFileSnapshotAge       time.Duration
	flushTimeout             time.Duration
	snapshotSeq              uint64
	expireTimeout            time.Duration
	tombstoneRetention       time.Duration
	repairQuickBuildTreeTime time.Duration
	maxRecvMsgSize           run.Bytes
	maxFileSnapshotNum       int
	repairTreeSlotCount      int
	snapshotMu               sync.Mutex
	port                     uint32
	tls                      bool
}

// NewServer returns a new standalone schema server.
func NewServer(omr observability.MetricsRegistry) Server {
	return &server{
		omr:    omr,
		closer: run.NewCloser(0),
	}
}

// GetPort returns the gRPC server port.
func (s *server) GetPort() *uint32 {
	return &s.port
}

// WatcherCount returns the number of active schema watchers.
func (s *server) WatcherCount() int {
	if s.watchers == nil {
		return 0
	}
	return s.watchers.Count()
}

// RegisterGossip registers the DB's repair gRPC services with the gossip messenger.
func (s *server) RegisterGossip(messenger gossip.Messenger) {
	s.db.RegisterGossip(messenger)
}

func (s *server) Role() databasev1.Role {
	return databasev1.Role_ROLE_META
}

func (s *server) Name() string {
	return "schema-server"
}

func (s *server) FlagSet() *run.FlagSet {
	flagS := run.NewFlagSet("schema-server")
	s.maxRecvMsgSize = defaultRecvSize
	flagS.StringVar(&s.root, "schema-server-root-path", "/tmp", "root storage path")
	flagS.StringVar(&s.host, "schema-server-grpc-host", "", "the host of schema server")
	flagS.Uint32Var(&s.port, "schema-server-grpc-port", 17916, "the port of schema server")
	flagS.DurationVar(&s.flushTimeout, "schema-server-flush-timeout", 5*time.Second, "memory flush interval")
	flagS.DurationVar(&s.expireTimeout, "schema-server-expire-delete-timeout", time.Hour*24*7, "soft-delete expiration")
	flagS.DurationVar(&s.tombstoneRetention, "schema-server-tombstone-retention", time.Hour*24*7,
		"schema tombstone retention duration; tombstones older than this value are physically excluded from queries")
	flagS.BoolVar(&s.tls, "schema-server-tls", false, "connection uses TLS if true")
	flagS.StringVar(&s.certFile, "schema-server-cert-file", "", "the TLS cert file")
	flagS.StringVar(&s.keyFile, "schema-server-key-file", "", "the TLS key file")
	flagS.VarP(&s.maxRecvMsgSize, "schema-server-max-recv-msg-size", "", "max gRPC receive message size")
	flagS.IntVar(&s.repairTreeSlotCount, "schema-server-repair-tree-slot-count", 32, "repair tree slot count")
	flagS.StringVar(&s.repairBuildTreeCron, "schema-server-repair-build-tree-cron", "@every 1h",
		"cron for repair tree building")
	flagS.DurationVar(&s.repairQuickBuildTreeTime, "schema-server-repair-quick-build-tree-time",
		time.Minute*10, "schema-quick build tree duration")
	flagS.IntVar(&s.maxFileSnapshotNum, "schema-server-max-file-snapshot-num", 10, "the maximum number of file snapshots allowed")
	flagS.DurationVar(&s.minFileSnapshotAge, "schema-server-min-file-snapshot-age", time.Hour, "the minimum age for file snapshots to be eligible for deletion")
	return flagS
}

func (s *server) Validate() error {
	if s.root == "" {
		return errors.New("root path must not be empty")
	}
	if s.port == 0 {
		s.port = 17920
	}
	// Short retention values (<1h) are unusual in production but required by integration tests
	// that verify the physical-removal path of tombstone GC. Accept the value as given so
	// operators (and the test harness) retain control; silent clamping previously hid the
	// test case.
	s.addr = net.JoinHostPort(s.host, strconv.FormatUint(uint64(s.port), 10))
	if s.addr == ":" {
		return errors.New("no address")
	}
	if !s.tls {
		return nil
	}
	if s.certFile == "" {
		return errors.New("invalid server cert file")
	}
	if s.keyFile == "" {
		return errors.New("invalid server key file")
	}
	return nil
}

// batchWaitSeconds converts a flush timeout to whole seconds for the index
// BatchWaitSec, rounding up with a floor of 1. A sub-second flush timeout must
// still yield a positive wait — a 0-second batch wait risks a tight flush loop.
func batchWaitSeconds(d time.Duration) int64 {
	return max(int64((d+time.Second-1)/time.Second), 1)
}

func (s *server) PreRun(ctx context.Context) error {
	s.l = logger.GetLogger("schema-server")
	var err error
	if s.root, err = banyandbpath.Get(s.root); err != nil {
		return err
	}
	s.lfs = fs.NewLocalFileSystem()

	s.watchers = newWatcherManager(s.l)
	grpcFactory := s.omr.With(schemaServerScope.SubScope("grpc"))
	sm := newServerMetrics(grpcFactory)
	s.schemaService = &schemaManagementServer{
		server:   s,
		watchers: s.watchers,
		l:        s.l,
		metrics:  sm,
	}
	s.updateService = &schemaUpdateServer{
		server:   s,
		watchers: s.watchers,
		l:        s.l,
		metrics:  sm,
	}

	if s.tls {
		var tlsErr error
		s.tlsReloader, tlsErr = pkgtls.NewReloader(s.certFile, s.keyFile, s.l)
		if tlsErr != nil {
			return errors.Wrap(tlsErr, "failed to initialize TLS reloader for server")
		}
	}

	dataDir := filepath.Join(s.root, snapshot.SchemaPropertyCatalogName, "data")
	s.snapshotDir = filepath.Join(s.root, snapshot.SchemaPropertyCatalogName, "snapshots")
	repairDir := filepath.Join(s.root, snapshot.SchemaPropertyCatalogName, "repair")

	expireToDelete := s.expireTimeout
	if s.tombstoneRetention > 0 {
		expireToDelete = s.tombstoneRetention
	}
	cfg := db.Config{
		Location:               dataDir,
		MetricsScopeName:       "schema_property",
		FlushInterval:          s.flushTimeout,
		ExpireToDeleteDuration: expireToDelete,
		Repair: db.RepairConfig{
			Enabled:            true,
			Location:           repairDir,
			BuildTreeCron:      s.repairBuildTreeCron,
			QuickBuildTreeTime: s.repairQuickBuildTreeTime,
			TreeSlotCount:      s.repairTreeSlotCount,
		},
		Index: db.IndexConfig{
			BatchWaitSec:       batchWaitSeconds(s.flushTimeout),
			WaitForPersistence: false,
		},
		Snapshot: db.SnapshotConfig{
			Location: s.snapshotDir,
			Func: func(ctx context.Context) (string, error) {
				s.snapshotMu.Lock()
				defer s.snapshotMu.Unlock()
				storage.DeleteStaleSnapshots(s.snapshotDir, s.maxFileSnapshotNum, s.minFileSnapshotAge, s.lfs)
				sn := s.snapshotName()
				snp := s.db.TakeSnapShot(ctx, sn)
				if snp.Error != "" {
					return "", fmt.Errorf("failed to take snapshot %s: %s", sn, snp.Error)
				}
				return filepath.Join(s.snapshotDir, sn, storage.DataDir), nil
			},
		},
	}

	var openErr error
	// nolint:contextcheck
	s.db, openErr = db.OpenDB(s.closer.Ctx(), cfg, s.omr, s.lfs)
	if openErr != nil {
		return errors.Wrap(openErr, "failed to open property database")
	}
	s.l.Info().Str("root", s.root).Msg("schema property database initialized")

	// Start gRPC server during PreRun so it is ready before other services try to connect.
	var opts []grpclib.ServerOption
	if s.tls {
		if s.tlsReloader != nil {
			if startErr := s.tlsReloader.Start(ctx); startErr != nil {
				return errors.Wrap(startErr, "failed to start TLS reloader for schema server")
			}
			s.l.Info().Str("certFile", s.certFile).Str("keyFile", s.keyFile).
				Msg("Started TLS file monitoring for schema server")
			tlsConfig := s.tlsReloader.GetTLSConfig()
			creds := credentials.NewTLS(tlsConfig)
			opts = []grpclib.ServerOption{grpclib.Creds(creds)}
		} else {
			creds, tlsErr := credentials.NewServerTLSFromFile(s.certFile, s.keyFile)
			if tlsErr != nil {
				return errors.Wrap(tlsErr, "failed to load TLS credentials")
			}
			opts = []grpclib.ServerOption{grpclib.Creds(creds)}
		}
	}
	grpcPanicRecoveryHandler := func(ctx context.Context, p any) (err error) {
		breadcrumbs := panicdiag.BreadcrumbsFromContext(ctx)
		stages := make([]string, len(breadcrumbs))
		for idx, bc := range breadcrumbs {
			stages[idx] = bc.Stage
		}
		s.l.Error().Interface("panic", p).Str("stack", string(debug.Stack())).Strs("breadcrumbs", stages).Msg("recovered from panic")
		return status.Errorf(codes.Internal, "%s", p)
	}
	streamChain := []grpclib.StreamServerInterceptor{
		panicdiag.BreadcrumbStreamInterceptor(),
		recovery.StreamServerInterceptor(recovery.WithRecoveryHandlerContext(grpcPanicRecoveryHandler)),
	}
	unaryChain := []grpclib.UnaryServerInterceptor{
		panicdiag.BreadcrumbUnaryInterceptor(),
		grpc_validator.UnaryServerInterceptor(),
		recovery.UnaryServerInterceptor(recovery.WithRecoveryHandlerContext(grpcPanicRecoveryHandler)),
	}
	opts = append(opts, grpclib.MaxRecvMsgSize(int(s.maxRecvMsgSize)),
		grpclib.ChainUnaryInterceptor(unaryChain...),
		grpclib.ChainStreamInterceptor(streamChain...),
	)
	s.ser = grpclib.NewServer(opts...)
	schemav1.RegisterSchemaManagementServiceServer(s.ser, s.schemaService)
	schemav1.RegisterSchemaUpdateServiceServer(s.ser, s.updateService)
	grpc_health_v1.RegisterHealthServer(s.ser, health.NewServer())

	lis, lisErr := net.Listen("tcp", s.addr)
	if lisErr != nil {
		return errors.Wrap(lisErr, "failed to listen on schema server address")
	}
	s.l.Info().Str("addr", s.addr).Msg("Listening to")
	s.closer.AddRunning()
	go func() {
		defer s.closer.Done()
		serveErr := s.ser.Serve(lis)
		if serveErr != nil {
			s.l.Error().Err(serveErr).Msg("server is interrupted")
		}
	}()
	return nil
}

// TakeSnapshot creates a new snapshot and returns the snapshot metadata.
func (s *server) TakeSnapshot(ctx context.Context) *databasev1.Snapshot {
	s.snapshotMu.Lock()
	defer s.snapshotMu.Unlock()
	storage.DeleteStaleSnapshots(s.snapshotDir, s.maxFileSnapshotNum, s.minFileSnapshotAge, s.lfs)
	sn := s.snapshotName()
	return s.db.TakeSnapShot(ctx, sn)
}

func (s *server) snapshotName() string {
	s.snapshotSeq++
	return fmt.Sprintf("%s-%08X", time.Now().UTC().Format(storage.SnapshotTimeFormat), s.snapshotSeq)
}

func (s *server) Serve() run.StopNotify {
	return s.closer.CloseNotify()
}

func (s *server) GracefulStop() {
	if s.watchers != nil {
		s.watchers.Close()
	}
	if s.tlsReloader != nil {
		s.tlsReloader.Stop()
	}
	if s.ser != nil {
		stopped := make(chan struct{})
		go func() {
			s.ser.GracefulStop()
			close(stopped)
		}()
		t := time.NewTimer(10 * time.Second)
		select {
		case <-t.C:
			s.ser.Stop()
			s.l.Info().Msg("force stopped")
		case <-stopped:
			t.Stop()
			s.l.Info().Msg("stopped gracefully")
		}
	}
	if s.db != nil {
		if closeErr := s.db.Close(); closeErr != nil {
			s.l.Error().Err(closeErr).Msg("failed to close database")
		}
	}
	s.closer.CloseThenWait()
}
