// 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 measure

import (
	"context"
	"encoding/json"
	"fmt"
	"path"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/pkg/errors"
	"go.uber.org/multierr"

	"github.com/apache/skywalking-banyandb/api/common"
	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	"github.com/apache/skywalking-banyandb/api/validate"
	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
	"github.com/apache/skywalking-banyandb/banyand/internal/wqueue"
	"github.com/apache/skywalking-banyandb/banyand/liaison/grpc"
	"github.com/apache/skywalking-banyandb/banyand/metadata"
	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
	"github.com/apache/skywalking-banyandb/banyand/observability"
	"github.com/apache/skywalking-banyandb/banyand/protector"
	"github.com/apache/skywalking-banyandb/banyand/queue"
	"github.com/apache/skywalking-banyandb/banyand/queue/pub"
	"github.com/apache/skywalking-banyandb/pkg/logger"
	"github.com/apache/skywalking-banyandb/pkg/meter"
	resourceSchema "github.com/apache/skywalking-banyandb/pkg/schema"
	"github.com/apache/skywalking-banyandb/pkg/schema/registry"
	"github.com/apache/skywalking-banyandb/pkg/timestamp"
)

// measureRegistryKinds is the kind set the measure schemaRepo registers with
// the per-node NodeRepoRegistry. TopNAggregation is intentionally excluded —
// schemaRepo does not store TopN entries, so the cluster barrier reads TopN
// keys via the property schemaCache instead. Property kinds are similarly
// out-of-scope for the per-service repos.
const measureRegistryKinds = schema.KindGroup | schema.KindMeasure | schema.KindIndexRule | schema.KindIndexRuleBinding

const (
	// TopNSchemaName is the name of the top n result schema.
	TopNSchemaName = "_top_n_result"
	// TopNTagFamily is the tag family name of the topN result measure.
	TopNTagFamily = "_topN"
	// TopNFieldName is the field name of the topN result measure.
	TopNFieldName = "value"
)

var (
	metadataScope = measureScope.SubScope("metadata")

	topNFieldsSpec = []*databasev1.FieldSpec{{
		Name:              TopNFieldName,
		FieldType:         databasev1.FieldType_FIELD_TYPE_DATA_BINARY,
		EncodingMethod:    databasev1.EncodingMethod_ENCODING_METHOD_GORILLA,
		CompressionMethod: databasev1.CompressionMethod_COMPRESSION_METHOD_ZSTD,
	}}
	// TopNTagNames is the tag names of the topN result measure.
	TopNTagNames = []string{"name", "direction", "group", "parameters"}
)

// SchemaService allows querying schema information.
type SchemaService interface {
	Query
	Close()
}
type schemaRepo struct {
	resourceSchema.Repository
	metadata         metadata.Repo
	pipeline         queue.Client
	l                *logger.Logger
	ctx              context.Context
	cancel           context.CancelFunc
	closingGroups    map[string]struct{}
	topNProcessorMap sync.Map
	nodeID           string
	path             string
	closingGroupsMu  sync.RWMutex
	role             databasev1.Role
}

func newSchemaRepo(path string, svc *standalone, nodeLabels map[string]string, nodeID string) *schemaRepo {
	ctx, cancel := context.WithCancel(context.Background())
	sr := &schemaRepo{
		path:          path,
		l:             svc.l,
		metadata:      svc.metadata,
		pipeline:      svc.localPipeline,
		nodeID:        nodeID,
		closingGroups: make(map[string]struct{}),
		role:          databasev1.Role_ROLE_DATA,
		ctx:           ctx,
		cancel:        cancel,
	}
	sr.Repository = resourceSchema.NewRepository(
		svc.metadata,
		svc.l,
		newSupplier(path, svc, sr, nodeLabels),
		resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
	)
	sr.start()
	sr.registerWithNodeRepo()
	return sr
}

func newLiaisonSchemaRepo(path string, svc *liaison, measureDataNodeRegistry grpc.NodeRegistry, pipeline queue.Client) *schemaRepo {
	ctx, cancel := context.WithCancel(context.Background())
	sr := &schemaRepo{
		path:          path,
		l:             svc.l,
		metadata:      svc.metadata,
		pipeline:      pipeline,
		closingGroups: make(map[string]struct{}),
		role:          databasev1.Role_ROLE_LIAISON,
		ctx:           ctx,
		cancel:        cancel,
	}
	sr.Repository = resourceSchema.NewRepository(
		svc.metadata,
		svc.l,
		newQueueSupplier(path, svc, measureDataNodeRegistry),
		resourceSchema.NewMetrics(svc.omr.With(metadataScope)),
	)
	sr.start()
	sr.registerWithNodeRepo()
	return sr
}

// markGroupClosing marks a group as closing to prevent new processors from being created.
func (sr *schemaRepo) markGroupClosing(group string) {
	sr.closingGroupsMu.Lock()
	if sr.closingGroups == nil {
		sr.closingGroups = make(map[string]struct{})
	}
	sr.closingGroups[group] = struct{}{}
	sr.closingGroupsMu.Unlock()
}

// unmarkGroupClosing removes the closing mark of a group.
func (sr *schemaRepo) unmarkGroupClosing(group string) {
	sr.closingGroupsMu.Lock()
	delete(sr.closingGroups, group)
	sr.closingGroupsMu.Unlock()
}

// isGroupClosing returns true if the group is currently being closed.
func (sr *schemaRepo) isGroupClosing(group string) bool {
	sr.closingGroupsMu.RLock()
	_, ok := sr.closingGroups[group]
	sr.closingGroupsMu.RUnlock()
	return ok
}

func (sr *schemaRepo) start() {
	sr.Watcher()
	sr.metadata.
		RegisterHandler("measure", schema.KindGroup|schema.KindMeasure|schema.KindIndexRuleBinding|schema.KindIndexRule|schema.KindTopNAggregation,
			sr)
}

// registerWithNodeRepo joins this schemaRepo to the per-node aggregator so the
// cluster barrier and NodeSchemaStatusService route Group/Measure/IndexRule/
// IndexRuleBinding lookups through the same cache the executor consults via
// LoadGroup / LoadResource. Idempotent and safe to call from every measure
// constructor (standalone / data / liaison).
func (sr *schemaRepo) registerWithNodeRepo() {
	metaSvc, ok := sr.metadata.(metadata.Service)
	if !ok {
		return
	}
	registry.MaybeRegister(metaSvc.NodeRepoRegistry(), measureRegistryKinds, sr.Repository)
}

func (sr *schemaRepo) Measure(metadata *commonv1.Metadata) (Measure, error) {
	sm, ok := sr.loadMeasure(metadata)
	if !ok {
		return nil, errors.WithStack(ErrMeasureNotExist)
	}
	return sm, nil
}

func (sr *schemaRepo) GetRemovalSegmentsTimeRange(group string) *timestamp.TimeRange {
	g, ok := sr.LoadGroup(group)
	if !ok {
		return nil
	}
	db := g.SupplyTSDB()
	if db == nil {
		return nil
	}
	return db.(storage.TSDB[*tsTable, option]).GetExpiredSegmentsTimeRange()
}

func (sr *schemaRepo) OnInit(kinds []schema.Kind) (bool, []int64) {
	if len(kinds) != 5 {
		logger.Panicf("unexpected kinds: %v", kinds)
		return false, nil
	}
	groupNames, revs := sr.Repository.Init(schema.KindMeasure)
	for i := range groupNames {
		sr.createTopNResultMeasure(sr.ctx, sr.metadata.MeasureRegistry(), groupNames[i])
	}
	return true, revs
}

func (sr *schemaRepo) OnAddOrUpdate(metadata schema.Metadata) {
	switch metadata.Kind {
	case schema.KindGroup:
		g := metadata.Spec.(*commonv1.Group)
		if g.Catalog != commonv1.Catalog_CATALOG_MEASURE {
			return
		}
		if err := validate.GroupForNonProperty(g); err != nil {
			sr.l.Warn().Err(err).Msg("group is ignored")
			return
		}
		sr.SendMetadataEvent(resourceSchema.MetadataEvent{
			Typ:      resourceSchema.EventAddOrUpdate,
			Kind:     resourceSchema.EventKindGroup,
			Metadata: g,
		})
		sr.createTopNResultMeasure(sr.ctx, sr.metadata.MeasureRegistry(), g.Metadata.Name)
	case schema.KindMeasure:
		m := metadata.Spec.(*databasev1.Measure)
		if err := validate.Measure(m); err != nil {
			sr.l.Warn().Err(err).Msg("measure is ignored")
			return
		}
		if subsetWarn := validate.CheckShardingKeySubset(m); subsetWarn != nil {
			sr.l.Warn().Err(subsetWarn).Str("measure", m.GetMetadata().GetName()).Msg("sharding key is not a subset of entity tags")
		}
		sr.SendMetadataEvent(resourceSchema.MetadataEvent{
			Typ:      resourceSchema.EventAddOrUpdate,
			Kind:     resourceSchema.EventKindResource,
			Metadata: m,
		})
	case schema.KindIndexRuleBinding:
		if irb, ok := metadata.Spec.(*databasev1.IndexRuleBinding); ok {
			if err := validate.IndexRuleBinding(irb); err != nil {
				sr.l.Warn().Err(err).Msg("index rule binding is ignored")
				return
			}
			if irb.GetSubject().Catalog == commonv1.Catalog_CATALOG_MEASURE {
				sr.SendMetadataEvent(resourceSchema.MetadataEvent{
					Typ:      resourceSchema.EventAddOrUpdate,
					Kind:     resourceSchema.EventKindIndexRuleBinding,
					Metadata: irb,
				})
			}
		}
	case schema.KindIndexRule:
		if ir, ok := metadata.Spec.(*databasev1.IndexRule); ok {
			if err := validate.IndexRule(metadata.Spec.(*databasev1.IndexRule)); err != nil {
				sr.l.Warn().Err(err).Msg("index rule is ignored")
				return
			}
			sr.SendMetadataEvent(resourceSchema.MetadataEvent{
				Typ:      resourceSchema.EventAddOrUpdate,
				Kind:     resourceSchema.EventKindIndexRule,
				Metadata: ir,
			})
		}
	case schema.KindTopNAggregation:
		if sr.pipeline == nil {
			return
		}
		topNSchema := metadata.Spec.(*databasev1.TopNAggregation)
		// Skip creating/registering processors if the group is closing
		if sr.isGroupClosing(topNSchema.GetMetadata().GetGroup()) || (topNSchema.SourceMeasure != nil && sr.isGroupClosing(topNSchema.SourceMeasure.GetGroup())) {
			sr.l.Debug().Str("group", topNSchema.GetMetadata().GetGroup()).
				Str("topN", topNSchema.GetMetadata().GetName()).
				Msg("skip TopNAggregation registration: group is closing")
			return
		}
		if err := validate.TopNAggregation(topNSchema); err != nil {
			sr.l.Warn().Err(err).Msg("topNAggregation is ignored")
			return
		}
		manager := sr.getSteamingManager(sr.ctx, topNSchema.SourceMeasure, sr.pipeline)
		if manager == nil {
			// group is closing; skip registering
			return
		}
		manager.register(sr.ctx, topNSchema)
	default:
	}
}

func (sr *schemaRepo) OnDelete(metadata schema.Metadata) {
	switch metadata.Kind {
	case schema.KindGroup:
		g := metadata.Spec.(*commonv1.Group)
		if g.Catalog != commonv1.Catalog_CATALOG_MEASURE {
			return
		}
		// Mark group as closing to prevent new processors from being created during deletion
		sr.markGroupClosing(g.Metadata.Name)
		sr.SendMetadataEvent(resourceSchema.MetadataEvent{
			Typ:      resourceSchema.EventDelete,
			Kind:     resourceSchema.EventKindGroup,
			Metadata: g,
		})
		sr.stopAllProcessorsWithGroupPrefix(g.Metadata.Name)
		// Deletion completed; allow future re-creation
		sr.unmarkGroupClosing(g.Metadata.Name)
	case schema.KindMeasure:
		m := metadata.Spec.(*databasev1.Measure)
		sr.SendMetadataEvent(resourceSchema.MetadataEvent{
			Typ:            resourceSchema.EventDelete,
			Kind:           resourceSchema.EventKindResource,
			DeleteRevision: m.GetMetadata().GetModRevision(),
			Metadata:       m,
		})
		sr.stopSteamingManager(m.GetMetadata())
	case schema.KindIndexRuleBinding:
		if binding, ok := metadata.Spec.(*databasev1.IndexRuleBinding); ok {
			if binding.GetSubject().Catalog == commonv1.Catalog_CATALOG_MEASURE {
				sr.SendMetadataEvent(resourceSchema.MetadataEvent{
					Typ:      resourceSchema.EventDelete,
					Kind:     resourceSchema.EventKindIndexRuleBinding,
					Metadata: metadata.Spec.(*databasev1.IndexRuleBinding),
				})
			}
		}

	case schema.KindIndexRule:
		if rule, ok := metadata.Spec.(*databasev1.IndexRule); ok {
			sr.SendMetadataEvent(resourceSchema.MetadataEvent{
				Typ:      resourceSchema.EventDelete,
				Kind:     resourceSchema.EventKindIndexRule,
				Metadata: rule,
			})
		}
	case schema.KindTopNAggregation:
		if sr.pipeline != nil {
			topNAggregation := metadata.Spec.(*databasev1.TopNAggregation)
			sr.removeTopNAggregation(topNAggregation.SourceMeasure, topNAggregation)
		}
	default:
	}
}

func (sr *schemaRepo) Close() {
	sr.cancel()
	// Close the Repository first to stop the event watcher, ensuring no new
	// topN processors are created while we are draining the existing ones.
	sr.Repository.Close()
	var err error
	sr.topNProcessorMap.Range(func(_, val any) bool {
		manager := val.(*topNProcessorManager)
		err = multierr.Append(err, manager.Close())
		return true
	})
	if err != nil {
		sr.l.Error().Err(err).Msg("faced error when closing schema repository")
	}
}

func (sr *schemaRepo) loadMeasure(metadata *commonv1.Metadata) (*measure, bool) {
	r, ok := sr.LoadResource(metadata)
	if !ok {
		return nil, false
	}
	s, ok := r.Delegated().(*measure)
	return s, ok
}

func (sr *schemaRepo) loadTSDB(groupName string) (storage.TSDB[*tsTable, option], error) {
	if sr == nil {
		return nil, fmt.Errorf("schemaRepo is nil")
	}
	g, ok := sr.LoadGroup(groupName)
	if !ok {
		return nil, fmt.Errorf("group %s not found", groupName)
	}
	db := g.SupplyTSDB()
	if db == nil {
		return nil, fmt.Errorf("group %s not found", groupName)
	}
	return db.(storage.TSDB[*tsTable, option]), nil
}

func (sr *schemaRepo) loadQueue(groupName string) (*wqueue.Queue[*tsTable, option], error) {
	g, ok := sr.LoadGroup(groupName)
	if !ok {
		return nil, fmt.Errorf("group %s not found", groupName)
	}
	db := g.SupplyTSDB()
	if db == nil {
		return nil, fmt.Errorf("queue for group %s not found", groupName)
	}
	return db.(*wqueue.Queue[*tsTable, option]), nil
}

// CollectDataInfo collects data info for a specific group.
func (sr *schemaRepo) CollectDataInfo(ctx context.Context, group string) (*databasev1.DataInfo, error) {
	if sr.nodeID == "" {
		return nil, nil
	}
	node, nodeErr := sr.metadata.NodeRegistry().GetNode(ctx, sr.nodeID)
	if nodeErr != nil {
		return nil, nodeErr
	}
	tsdb, tsdbErr := sr.loadTSDB(group)
	if tsdbErr != nil {
		return nil, tsdbErr
	}
	if tsdb == nil {
		return nil, nil
	}
	// reopenClosed=false: do NOT reopen closed segments, so inspecting cold
	// segments never reopens their writable index (no exclusive-lock churn).
	// Closed segments are reported from their on-disk files.
	segments, segmentsErr := tsdb.SelectSegments(timestamp.TimeRange{
		Start: time.Unix(0, 0),
		End:   time.Unix(0, timestamp.MaxNanoTime),
	}, false)
	if segmentsErr != nil {
		return nil, segmentsErr
	}
	var segmentInfoList []*databasev1.SegmentInfo
	var totalDataSize int64
	for _, segment := range segments {
		timeRange := segment.GetTimeRange()
		tables, _ := segment.Tables()
		var shardInfoList []*databasev1.ShardInfo
		if len(tables) > 0 {
			for shardIdx, table := range tables {
				shardInfo := sr.collectShardInfo(table, uint32(shardIdx))
				shardInfoList = append(shardInfoList, shardInfo)
				totalDataSize += shardInfo.DataSizeBytes
			}
		} else {
			// No live shard tables (a closed segment, or a brand-new open one
			// with no data yet): read shard part stats from disk without
			// reopening. Either way the result is the on-disk shard stats.
			closedShards, closedSize := storage.CollectClosedShardInfo(segment.Location())
			shardInfoList = closedShards
			totalDataSize += closedSize
		}
		seriesIndexInfo := sr.collectSeriesIndexInfo(segment)
		totalDataSize += seriesIndexInfo.DataSizeBytes
		segmentInfo := &databasev1.SegmentInfo{
			SegmentId:       fmt.Sprintf("%d-%d", timeRange.Start.UnixNano(), timeRange.End.UnixNano()),
			TimeRangeStart:  timeRange.Start.Format(time.RFC3339Nano),
			TimeRangeEnd:    timeRange.End.Format(time.RFC3339Nano),
			ShardInfo:       shardInfoList,
			SeriesIndexInfo: seriesIndexInfo,
		}
		segmentInfoList = append(segmentInfoList, segmentInfo)
		segment.DecRef()
	}
	dataInfo := &databasev1.DataInfo{
		Node:          node,
		SegmentInfo:   segmentInfoList,
		DataSizeBytes: totalDataSize,
	}
	return dataInfo, nil
}

func (sr *schemaRepo) collectSeriesIndexInfo(segment storage.Segment[*tsTable, option]) *databasev1.SeriesIndexInfo {
	// SeriesIndexStats reads from the live index when the segment is open and
	// from disk (read-only) when it is closed, so inspecting a cold segment
	// never reopens its writable index.
	dataCount, dataSizeBytes := segment.SeriesIndexStats()
	return &databasev1.SeriesIndexInfo{
		DataCount:     dataCount,
		DataSizeBytes: dataSizeBytes,
	}
}

func (sr *schemaRepo) collectShardInfo(table any, shardID uint32) *databasev1.ShardInfo {
	tst, ok := table.(*tsTable)
	if !ok {
		return &databasev1.ShardInfo{
			ShardId: shardID,
		}
	}
	snapshot := tst.currentSnapshot()
	if snapshot == nil {
		return &databasev1.ShardInfo{
			ShardId: shardID,
		}
	}
	defer snapshot.decRef()
	var totalCount, compressedSize, partCount, filePartCount uint64
	for _, pw := range snapshot.parts {
		if pw.p != nil {
			totalCount += pw.p.partMetadata.TotalCount
			compressedSize += pw.p.partMetadata.CompressedSizeBytes
			partCount++
			filePartCount++
		} else if pw.mp != nil {
			totalCount += pw.mp.partMetadata.TotalCount
			compressedSize += pw.mp.partMetadata.CompressedSizeBytes
			partCount++
		}
	}
	return &databasev1.ShardInfo{
		ShardId:           shardID,
		DataCount:         int64(totalCount),
		DataSizeBytes:     int64(compressedSize),
		PartCount:         int64(partCount),
		InvertedIndexInfo: &databasev1.InvertedIndexInfo{},
		SidxInfo:          &databasev1.SIDXInfo{},
		FilePartCount:     int64(filePartCount),
	}
}

func (sr *schemaRepo) collectPendingWriteInfo(groupName string) (int64, error) {
	if sr == nil || sr.Repository == nil {
		return 0, fmt.Errorf("schema repository is not initialized")
	}
	if sr.role == databasev1.Role_ROLE_LIAISON {
		queue, queueErr := sr.loadQueue(groupName)
		if queueErr != nil {
			return 0, fmt.Errorf("failed to load queue: %w", queueErr)
		}
		if queue == nil {
			return 0, nil
		}
		var pendingWriteCount int64
		for _, sq := range queue.SubQueues() {
			if sq != nil {
				pendingWriteCount += sq.getPendingDataCount()
			}
		}
		return pendingWriteCount, nil
	}
	// Standalone mode
	tsdb, tsdbErr := sr.loadTSDB(groupName)
	if tsdbErr != nil {
		return 0, fmt.Errorf("failed to load TSDB: %w", tsdbErr)
	}
	if tsdb == nil {
		return 0, fmt.Errorf("TSDB is nil for group %s", groupName)
	}
	segments, segmentsErr := tsdb.SelectSegments(timestamp.TimeRange{
		Start: time.Unix(0, 0),
		End:   time.Unix(0, timestamp.MaxNanoTime),
	}, false)
	if segmentsErr != nil {
		return 0, fmt.Errorf("failed to select segments: %w", segmentsErr)
	}
	var pendingWriteCount int64
	for _, segment := range segments {
		tables, _ := segment.Tables()
		for _, tst := range tables {
			pendingWriteCount += tst.getPendingDataCount()
		}
		segment.DecRef()
	}
	return pendingWriteCount, nil
}

func (sr *schemaRepo) collectPendingSyncInfo(groupName string) (partCount int64, totalSizeBytes int64, err error) {
	if sr == nil || sr.Repository == nil {
		return 0, 0, fmt.Errorf("schema repository is not initialized")
	}
	// Only liaison nodes collect pending sync info
	if sr.role != databasev1.Role_ROLE_LIAISON {
		return 0, 0, nil
	}
	queue, queueErr := sr.loadQueue(groupName)
	if queueErr != nil {
		return 0, 0, fmt.Errorf("failed to load queue: %w", queueErr)
	}
	if queue == nil {
		return 0, 0, nil
	}
	for _, sq := range queue.SubQueues() {
		if sq != nil {
			snapshot := sq.currentSnapshot()
			if snapshot != nil {
				for _, pw := range snapshot.parts {
					if pw.mp == nil && pw.p != nil && pw.p.partMetadata.TotalCount > 0 {
						partCount++
						totalSizeBytes += int64(pw.p.partMetadata.CompressedSizeBytes)
					}
				}
				snapshot.decRef()
			}
		}
	}
	return partCount, totalSizeBytes, nil
}

func (sr *schemaRepo) createTopNResultMeasure(ctx context.Context, measureSchemaRegistry schema.Measure, group string) {
	md := GetTopNSchemaMetadata(group)
	operation := func() error {
		m, err := measureSchemaRegistry.GetMeasure(ctx, md)
		if err != nil && !errors.Is(err, schema.ErrGRPCResourceNotFound) {
			return errors.WithMessagef(err, "fail to get %s", md)
		}
		if m != nil {
			return nil
		}

		m = GetTopNSchema(md)
		if _, innerErr := measureSchemaRegistry.CreateMeasure(ctx, m); innerErr != nil {
			if !errors.Is(innerErr, schema.ErrGRPCAlreadyExists) {
				return errors.WithMessagef(innerErr, "fail to create new topN measure %s", m)
			}
		}
		return nil
	}

	backoffStrategy := backoff.NewExponentialBackOff()
	backoffStrategy.MaxElapsedTime = 0 // never stop until topN measure has been created

	if err := backoff.Retry(operation, backoff.WithContext(backoffStrategy, ctx)); err != nil {
		if !errors.Is(err, context.Canceled) {
			logger.Panicf("fail to create topN measure %s: %v", md, err)
		}
	}
}

func (sr *schemaRepo) stopAllProcessorsWithGroupPrefix(groupName string) {
	groupPrefix := groupName + "/"
	totalClosed := 0
	for {
		closedThisRound := 0
		sr.topNProcessorMap.Range(func(key, val any) bool {
			keyStr := key.(string)
			if strings.HasPrefix(keyStr, groupPrefix) {
				manager := val.(*topNProcessorManager)
				if err := manager.Close(); err != nil {
					sr.l.Error().Err(err).Str("key", keyStr).Msg("failed to close topN processor manager")
				} else {
					sr.topNProcessorMap.Delete(keyStr)
					closedThisRound++
				}
			}
			return true
		})
		if closedThisRound == 0 {
			break
		}
		totalClosed += closedThisRound
		// continue to ensure no late-created managers remain
	}
	if totalClosed > 0 {
		sr.l.Info().Str("groupName", groupName).Int("count", totalClosed).Msg("stopped topN processors for group")
	}
}

var _ resourceSchema.ResourceSupplier = (*supplier)(nil)

type supplier struct {
	metadata            metadata.Repo
	omr                 observability.MetricsRegistry
	c                   storage.Cache
	pm                  protector.Memory
	l                   *logger.Logger
	schemaRepo          *schemaRepo
	nodeLabels          map[string]string
	queryMetrics        atomic.Pointer[queryMetrics]
	queryMetricsFactory atomic.Value
	path                string
	option              option
}

func newSupplier(path string, svc *standalone, sr *schemaRepo, nodeLabels map[string]string) *supplier {
	if svc.pm == nil {
		svc.l.Panic().Msg("CRITICAL: svc.pm is nil in newSupplier")
	}
	opt := svc.option
	opt.protector = svc.pm

	if opt.protector == nil {
		svc.l.Panic().Msg("CRITICAL: opt.protector is still nil after assignment")
	}

	return &supplier{
		path:       path,
		metadata:   svc.metadata,
		l:          svc.l,
		c:          svc.c,
		option:     opt,
		omr:        svc.omr,
		pm:         svc.pm,
		schemaRepo: sr,
		nodeLabels: nodeLabels,
	}
}

func (s *supplier) OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error) {
	measureSchema := spec.Schema().(*databasev1.Measure)
	return openMeasure(measureSpec{
		schema: measureSchema,
	}, s.l, s.c, s.pm, s.schemaRepo, s.queryMetrics.Load(), s.option.vectorized)
}

func (s *supplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	return s.metadata.MeasureRegistry().GetMeasure(ctx, md)
}

func (s *supplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
	name := groupSchema.Metadata.Name
	p := common.Position{
		Module:   "measure",
		Database: name,
	}
	metrics, factory := s.newMetrics(p)
	ro := groupSchema.ResourceOpts
	if ro == nil {
		return nil, fmt.Errorf("no resource opts in group %s", name)
	}
	res, err := pub.ResolveStage(s.l, name, ro, s.nodeLabels)
	if err != nil {
		return nil, err
	}
	group := groupSchema.Metadata.Name
	opts := storage.TSDBOpts[*tsTable, option]{
		ShardNum:                       res.ResourceOpts.ShardNum,
		Location:                       path.Join(s.path, group),
		TSTableCreator:                 newTSTable,
		TableMetrics:                   metrics,
		SegmentInterval:                storage.MustToIntervalRule(res.ResourceOpts.SegmentInterval),
		TTL:                            storage.MustToIntervalRule(res.ResourceOpts.Ttl),
		Option:                         s.option,
		SeriesIndexFlushTimeoutSeconds: s.option.flushTimeout.Nanoseconds() / int64(time.Second),
		SeriesIndexCacheMaxBytes:       int(s.option.seriesCacheMaxSize),
		StorageMetricsFactory:          factory,
		SegmentIdleTimeout:             res.SegmentIdleTimeout,
		DisableRetention:               res.DisableRetention,
		DisableRotation:                res.DisableRotation,
		MemoryLimit:                    s.pm.GetLimit(),
	}
	return storage.OpenTSDB(
		common.SetPosition(context.Background(), func(_ common.Position) common.Position {
			return p
		}),
		opts, s.c, group)
}

// ResolveResourceOpts returns the stage-resolved ResourceOpts for this node so that a
// group UpdateOptions applies the matched stage's interval/ttl/shardNum instead of the
// group default. On resolution error it keeps the group default rather than failing the
// update.
func (s *supplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
	return pub.ResolveResourceOptsForUpdate(s.l, groupSchema, s.nodeLabels)
}

type queueSupplier struct {
	metadata                metadata.Repo
	omr                     observability.MetricsRegistry
	pm                      protector.Memory
	measureDataNodeRegistry grpc.NodeRegistry
	l                       *logger.Logger
	schemaRepo              *schemaRepo
	path                    string
	option                  option
}

func newQueueSupplier(path string, svc *liaison, measureDataNodeRegistry grpc.NodeRegistry) *queueSupplier {
	opt := svc.option
	opt.protector = svc.pm

	if opt.protector == nil {
		svc.l.Panic().Msg("CRITICAL: opt.protector is still nil after assignment")
	}
	return &queueSupplier{
		metadata:                svc.metadata,
		omr:                     svc.omr,
		pm:                      svc.pm,
		measureDataNodeRegistry: measureDataNodeRegistry,
		l:                       svc.l,
		path:                    path,
		option:                  opt,
	}
}

func (s *queueSupplier) OpenResource(spec resourceSchema.Resource) (resourceSchema.IndexListener, error) {
	measureSchema := spec.Schema().(*databasev1.Measure)
	return openMeasure(measureSpec{
		schema: measureSchema,
	}, s.l, nil, s.pm, s.schemaRepo, nil, s.option.vectorized)
}

func (s *queueSupplier) ResourceSchema(md *commonv1.Metadata) (resourceSchema.ResourceSchema, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	return s.metadata.MeasureRegistry().GetMeasure(ctx, md)
}

// ResolveResourceOpts returns the group opts unchanged: the liaison write queue has no
// lifecycle stages to resolve.
func (s *queueSupplier) ResolveResourceOpts(groupSchema *commonv1.Group) *commonv1.ResourceOpts {
	return groupSchema.GetResourceOpts()
}

func (s *queueSupplier) OpenDB(groupSchema *commonv1.Group) (resourceSchema.DB, error) {
	name := groupSchema.Metadata.Name
	p := common.Position{
		Module:   "measure",
		Database: name,
	}
	ro := groupSchema.ResourceOpts
	if ro == nil {
		return nil, fmt.Errorf("no resource opts in group %s", name)
	}
	shardNum := ro.ShardNum
	group := groupSchema.Metadata.Name
	metrics, metricsFactory := s.newMetrics(p)
	opts := wqueue.Opts[*tsTable, option]{
		Group:           group,
		ShardNum:        shardNum,
		SegmentInterval: storage.MustToIntervalRule(ro.SegmentInterval),
		Location:        path.Join(s.path, group),
		Option:          s.option,
		Metrics:         metrics,
		MetricsFactory:  metricsFactory,
		SubQueueCreator: newWriteQueue,
		GetNodes: func(shardID common.ShardID) []string {
			copies := ro.Replicas + 1
			nodes, err := s.measureDataNodeRegistry.LocateAll(group, uint32(shardID), int(copies))
			if err != nil {
				s.l.Error().Err(err).Str("group", group).Uint32("shard", uint32(shardID)).Msg("failed to locate nodes")
				return nil
			}
			return nodes
		},
	}
	return wqueue.Open(
		common.SetPosition(context.Background(), func(_ common.Position) common.Position {
			return p
		}),
		opts, group,
	)
}

func (s *queueSupplier) newMetrics(p common.Position) (storage.Metrics, observability.Factory) {
	factory := s.omr.With(measureScope.ConstLabels(meter.ToLabelPairs(common.DBLabelNames(), p.DBLabelValues())))
	return &metrics{
		totalWritten:               factory.NewCounter("total_written"),
		totalBatch:                 factory.NewCounter("total_batch"),
		totalBatchIntroLatency:     factory.NewCounter("total_batch_intro_time"),
		totalIntroduceLoopStarted:  factory.NewCounter("total_introduce_loop_started", "phase"),
		totalIntroduceLoopFinished: factory.NewCounter("total_introduce_loop_finished", "phase"),
		totalFlushLoopStarted:      factory.NewCounter("total_flush_loop_started"),
		totalFlushLoopFinished:     factory.NewCounter("total_flush_loop_finished"),
		totalFlushLoopErr:          factory.NewCounter("total_flush_loop_err"),
		totalMergeLoopStarted:      factory.NewCounter("total_merge_loop_started"),
		totalMergeLoopFinished:     factory.NewCounter("total_merge_loop_finished"),
		totalMergeLoopErr:          factory.NewCounter("total_merge_loop_err"),
		totalSyncLoopStarted:       factory.NewCounter("total_sync_loop_started"),
		totalSyncLoopFinished:      factory.NewCounter("total_sync_loop_finished"),
		totalSyncLoopErr:           factory.NewCounter("total_sync_loop_err"),
		totalSyncLoopLatency:       factory.NewCounter("total_sync_loop_latency"),
		totalSyncLoopBytes:         factory.NewCounter("total_sync_loop_bytes"),
		totalFlushLoopProgress:     factory.NewCounter("total_flush_loop_progress"),
		totalFlushed:               factory.NewCounter("total_flushed"),
		totalFlushedMemParts:       factory.NewCounter("total_flushed_mem_parts"),
		totalFlushPauseCompleted:   factory.NewCounter("total_flush_pause_completed"),
		totalFlushPauseBreak:       factory.NewCounter("total_flush_pause_break"),
		totalFlushIntroLatency:     factory.NewCounter("total_flush_intro_latency"),
		totalFlushLatency:          factory.NewCounter("total_flush_latency"),
		totalMergedParts:           factory.NewCounter("total_merged_parts", "type"),
		totalMergeLatency:          factory.NewCounter("total_merge_latency", "type"),
		totalMerged:                factory.NewCounter("total_merged", "type"),
		tbMetrics: tbMetrics{
			totalMemParts:                  factory.NewGauge("total_mem_part", common.ShardLabelNames()...),
			totalMemElements:               factory.NewGauge("total_mem_elements", common.ShardLabelNames()...),
			totalMemBlocks:                 factory.NewGauge("total_mem_blocks", common.ShardLabelNames()...),
			totalMemPartBytes:              factory.NewGauge("total_mem_part_bytes", common.ShardLabelNames()...),
			totalMemPartUncompressedBytes:  factory.NewGauge("total_mem_part_uncompressed_bytes", common.ShardLabelNames()...),
			totalFileParts:                 factory.NewGauge("total_file_parts", common.ShardLabelNames()...),
			totalFileElements:              factory.NewGauge("total_file_elements", common.ShardLabelNames()...),
			totalFileBlocks:                factory.NewGauge("total_file_blocks", common.ShardLabelNames()...),
			totalFilePartBytes:             factory.NewGauge("total_file_part_bytes", common.ShardLabelNames()...),
			totalFilePartUncompressedBytes: factory.NewGauge("total_file_part_uncompressed_bytes", common.ShardLabelNames()...),
			pendingDataCount:               factory.NewGauge("pending_data_count", common.ShardLabelNames()...),
		},
	}, factory
}

// GetTopNSchema returns the schema of the topN result measure.
func GetTopNSchema(md *commonv1.Metadata) *databasev1.Measure {
	return &databasev1.Measure{
		Metadata: md,
		TagFamilies: []*databasev1.TagFamilySpec{
			{
				Name: TopNTagFamily,
				Tags: []*databasev1.TagSpec{
					{Name: TopNTagNames[0], Type: databasev1.TagType_TAG_TYPE_STRING},
					{Name: TopNTagNames[1], Type: databasev1.TagType_TAG_TYPE_INT},
					{Name: TopNTagNames[2], Type: databasev1.TagType_TAG_TYPE_STRING},
					{Name: TopNTagNames[3], Type: databasev1.TagType_TAG_TYPE_STRING},
				},
			},
		},
		Fields: topNFieldsSpec,
		Entity: &databasev1.Entity{
			TagNames: TopNTagNames,
		},
	}
}

// GetTopNSchemaMetadata returns the metadata of the topN result measure.
func GetTopNSchemaMetadata(group string) *commonv1.Metadata {
	return &commonv1.Metadata{
		Name:  TopNSchemaName,
		Group: group,
	}
}

func getKey(metadata *commonv1.Metadata) string {
	return path.Join(metadata.GetGroup(), metadata.GetName())
}

// TopNParameters defines the structure for the "parameters" tag value.
type TopNParameters struct {
	// Limit defines the number of top items to be kept.
	Limit int64 `json:"limit"`
	// FieldType indicates whether the topN values are int or float.
	FieldType databasev1.FieldType `json:"field_type"`
}

// String implements the fmt.Stringer interface.
func (p *TopNParameters) String() string {
	if p == nil {
		return ""
	}
	b, marshalErr := json.Marshal(p)
	if marshalErr != nil {
		return ""
	}
	return string(b)
}

// ParseTopNParameters decodes the JSON metadata.
func ParseTopNParameters(val string) (*TopNParameters, error) {
	if val == "" {
		return &TopNParameters{}, nil
	}
	var params TopNParameters
	if unmarshalErr := json.Unmarshal([]byte(val), &params); unmarshalErr != nil {
		return nil, fmt.Errorf("invalid TopNParameters %q: %w", val, unmarshalErr)
	}
	return &params, nil
}
