// 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 sidx provides interface definitions for the Secondary Index File System,
// enabling efficient secondary indexing with user-controlled int64 ordering keys.
package sidx

import (
	"context"
	"fmt"
	"sync/atomic"

	"github.com/apache/skywalking-banyandb/api/common"
	"github.com/apache/skywalking-banyandb/banyand/queue"
	"github.com/apache/skywalking-banyandb/pkg/index"
	pbv1 "github.com/apache/skywalking-banyandb/pkg/pb/v1"
	"github.com/apache/skywalking-banyandb/pkg/query/model"
)

// SnapshotTransactionSupport defines snapshot and transaction-related operations
// for coordinated snapshot updates. It is embedded by SIDX so that callers
// (e.g. trace) can participate in atomic snapshot transactions.
type SnapshotTransactionSupport interface {
	// PrepareMemPart prepares a transition for introducing a memory part.
	PrepareMemPart(partID uint64, mp *MemPart) func(cur *Snapshot) *Snapshot
	// PrepareFilePart prepares a transition for introducing a file-backed part.
	PrepareFilePart(partID uint64, partPath string) func(cur *Snapshot) *Snapshot
	// PrepareFlushed prepares a transition for introducing flushed parts.
	PrepareFlushed(intro *FlusherIntroduction) func(cur *Snapshot) *Snapshot
	// PrepareMerged prepares a transition for introducing merged parts.
	PrepareMerged(intro *MergerIntroduction) func(cur *Snapshot) *Snapshot
	// PrepareSynced prepares a transition for removing synced parts.
	PrepareSynced(partIDsToSync map[uint64]struct{}) func(cur *Snapshot) *Snapshot
	// CurrentSnapshot returns the current snapshot with incremented reference count.
	CurrentSnapshot() *Snapshot
	// ReplaceSnapshot atomically replaces the current snapshot with next.
	ReplaceSnapshot(next *Snapshot)
}

// SIDX defines the main secondary index interface with user-controlled ordering.
// The core principle is that int64 keys are provided by users and treated as
// opaque ordering values by sidx - the system only performs numerical comparisons
// without interpreting the semantic meaning of keys.
type SIDX interface {
	SnapshotTransactionSupport
	// IntroduceMemPart introduces a memPart to the SIDX instance.
	IntroduceMemPart(partID uint64, mp *MemPart)
	// IntroduceFlushed introduces a flushed map to the SIDX instance.
	IntroduceFlushed(nextIntroduction *FlusherIntroduction)
	// IntroduceMerged introduces a merged map and a new part to the SIDX instance.
	IntroduceMerged(nextIntroduction *MergerIntroduction) func()
	// ConvertToMemPart converts a write request to a memPart.
	// minTimestamp and maxTimestamp are optional; when provided they are stored in manifest.json for part selection.
	ConvertToMemPart(reqs []WriteRequest, segmentID int64, minTimestamp, maxTimestamp *int64) (*MemPart, error)
	// StreamingQuery executes the query and streams batched QueryResponse objects.
	// The returned QueryResponse channel contains ordered batches limited by req.MaxBatchSize
	// unique Data elements (when positive). The error channel delivers any fatal execution error.
	StreamingQuery(ctx context.Context, req QueryRequest) (<-chan *QueryResponse, <-chan error)
	// QuerySync executes the query synchronously and returns all matching responses.
	QuerySync(ctx context.Context, req QueryRequest) ([]*QueryResponse, error)
	// ScanQuery executes a synchronous full scan query without requiring series IDs.
	// It scans all blocks in parts sequentially and applies filters to each row.
	// Returns a slice of QueryResponse objects containing all matching results.
	// This is a synchronous operation suitable for dump/debug tools.
	ScanQuery(ctx context.Context, req ScanQueryRequest) ([]*QueryResponse, error)
	// Stats returns current system statistics and performance metrics.
	Stats(ctx context.Context) (*Stats, error)
	// Close gracefully shuts down the SIDX instance, ensuring all data is persisted.
	Close() error
	// Flush flushes the SIDX instance to disk.
	Flush(partIDsToFlush map[uint64]struct{}) (*FlusherIntroduction, error)
	// Merge merges the specified parts into a new part. keep is an optional
	// opaque per-element predicate: when non-nil, an element is retained only if
	// keep(data) returns true; nil means retain everything (the lossless bulk
	// merge path). sidx never decodes data — the caller owns the encoding.
	Merge(closeCh <-chan struct{}, partIDstoMerge map[uint64]struct{}, newPartID uint64, keep func([]byte) bool) (*MergerIntroduction, error)
	// StreamingParts returns the streaming parts.
	StreamingParts(partIDsToSync map[uint64]struct{}, group string, shardID uint32, name string) ([]queue.StreamingPartData, []func())
	// PartPaths returns filesystem paths for the requested partIDs keyed by partID.
	// Missing partIDs are omitted from the returned map.
	PartPaths(partIDs map[uint64]struct{}) map[uint64]string
	// IntroduceSynced introduces a synced map to the SIDX instance.
	IntroduceSynced(partIDsToSync map[uint64]struct{}) func()
	// TakeFileSnapshot creates a snapshot of the SIDX files at the specified destination path.
	TakeFileSnapshot(dst string) error
}

// WriteRequest contains data for a single write operation within a batch.
// The user provides the ordering key as an int64 value that sidx treats opaquely.
type WriteRequest struct {
	Data     []byte
	Tags     []Tag
	SeriesID common.SeriesID
	Key      int64
}

// QueryRequest specifies parameters for a query operation, following StreamQueryOptions pattern.
type QueryRequest struct {
	Filter         index.Filter
	TagFilter      model.TagFilterMatcher
	Order          *index.OrderBy
	MinKey         *int64
	MaxKey         *int64
	SchemaTagTypes map[string]pbv1.ValueType
	MinTimestamp   *int64
	MaxTimestamp   *int64
	SeriesIDs      []common.SeriesID
	TagProjection  []model.TagProjection
	MaxBatchSize   int
}

// ScanProgressFunc is a callback for reporting scan progress.
// It receives the current part number (1-based), total parts, and rows found so far.
type ScanProgressFunc func(currentPart, totalParts int, rowsFound int)

// ScanQueryRequest specifies parameters for a full-scan query operation.
// Unlike QueryRequest, this does not require SeriesIDs and does not support Filter
// (all blocks are scanned, none are skipped).
//
//nolint:govet // struct layout optimized for readability; 64 bytes is acceptable
type ScanQueryRequest struct {
	TagFilter      model.TagFilterMatcher
	SchemaTagTypes map[string]pbv1.ValueType
	OnProgress     ScanProgressFunc
	MinKey         *int64
	MaxKey         *int64
	MinTimestamp   *int64
	MaxTimestamp   *int64
	TagProjection  []model.TagProjection
	MaxBatchSize   int
}

// QueryResponse contains a batch of query results and execution metadata.
// This follows BanyanDB result patterns with parallel arrays for efficiency.
// Uses individual tag-based strategy (like trace module) rather than tag-family approach (like stream module).
type QueryResponse struct {
	Error    error
	Keys     []int64
	Data     [][]byte
	SIDs     []common.SeriesID
	PartIDs  []uint64
	Tags     map[string][]string // Projected tags: tag name -> values for each row
	Metadata ResponseMetadata
}

// Len returns the number of results in the QueryResponse.
func (qr *QueryResponse) Len() int {
	return len(qr.Keys)
}

// Reset resets the QueryResponse to its zero state for reuse.
func (qr *QueryResponse) Reset() {
	qr.Error = nil
	qr.Keys = qr.Keys[:0]
	qr.Data = qr.Data[:0]
	qr.SIDs = qr.SIDs[:0]
	qr.PartIDs = qr.PartIDs[:0]
	qr.Metadata = ResponseMetadata{}
}

// Validate validates a QueryResponse for correctness.
func (qr *QueryResponse) Validate() error {
	keysLen := len(qr.Keys)
	dataLen := len(qr.Data)
	sidsLen := len(qr.SIDs)
	partIDsLen := len(qr.PartIDs)

	if keysLen != dataLen {
		return fmt.Errorf("inconsistent array lengths: keys=%d, data=%d", keysLen, dataLen)
	}
	if keysLen != sidsLen {
		return fmt.Errorf("inconsistent array lengths: keys=%d, sids=%d", keysLen, sidsLen)
	}
	if keysLen != partIDsLen {
		return fmt.Errorf("inconsistent array lengths: keys=%d, partIDs=%d", keysLen, partIDsLen)
	}

	return nil
}

// CopyFrom copies the QueryResponse from other to qr.
func (qr *QueryResponse) CopyFrom(other *QueryResponse) {
	qr.Error = other.Error

	qr.Keys = append(qr.Keys[:0], other.Keys...)
	qr.SIDs = append(qr.SIDs[:0], other.SIDs...)
	qr.PartIDs = append(qr.PartIDs[:0], other.PartIDs...)

	if cap(qr.Data) < len(other.Data) {
		qr.Data = make([][]byte, len(other.Data))
	} else {
		qr.Data = qr.Data[:len(other.Data)]
	}
	for i, data := range other.Data {
		qr.Data[i] = append(qr.Data[i][:0], data...)
	}

	qr.Metadata = other.Metadata
}

// Stats contains system statistics and performance metrics.
type Stats struct {
	MemoryUsageBytes int64
	DiskUsageBytes   int64
	ElementCount     int64
	PartCount        int64
	QueryCount       atomic.Int64
	WriteCount       atomic.Int64
	LastFlushTime    int64
	LastMergeTime    int64
}

// ResponseMetadata provides query execution information for monitoring and debugging.
type ResponseMetadata struct {
	Warnings         []string
	ExecutionTimeMs  int64
	ElementsScanned  int64
	ElementsFiltered int64
	PartsAccessed    int
	BlocksScanned    int
	CacheHitRatio    float64
	TruncatedResults bool
}

// Validate validates ResponseMetadata for correctness.
func (rm *ResponseMetadata) Validate() error {
	if rm.ExecutionTimeMs < 0 {
		return fmt.Errorf("executionTimeMs cannot be negative")
	}
	if rm.ElementsScanned < 0 {
		return fmt.Errorf("elementsScanned cannot be negative")
	}
	if rm.ElementsFiltered < 0 {
		return fmt.Errorf("elementsFiltered cannot be negative")
	}
	if rm.ElementsFiltered > rm.ElementsScanned {
		return fmt.Errorf("elementsFiltered (%d) cannot exceed elementsScanned (%d)", rm.ElementsFiltered, rm.ElementsScanned)
	}
	if rm.PartsAccessed < 0 {
		return fmt.Errorf("partsAccessed cannot be negative")
	}
	if rm.BlocksScanned < 0 {
		return fmt.Errorf("blocksScanned cannot be negative")
	}
	if rm.CacheHitRatio < 0.0 || rm.CacheHitRatio > 1.0 {
		return fmt.Errorf("cacheHitRatio must be between 0.0 and 1.0, got %f", rm.CacheHitRatio)
	}
	return nil
}

// Tag represents an individual tag for WriteRequest.
// This is an exported type that can be used outside the package.
type Tag struct {
	Name      string
	Value     []byte
	ValueArr  [][]byte
	ValueType pbv1.ValueType
}

// Reset resets the Tag to its zero state for reuse.
func (t *Tag) Reset() {
	t.Name = ""
	t.Value = nil
	t.ValueArr = nil
	t.ValueType = pbv1.ValueTypeUnknown
}

// Size returns the size of the tag in bytes.
func (t *Tag) Size() int {
	size := len(t.Name) + 1 // +1 for valueType
	if t.ValueArr != nil {
		for _, v := range t.ValueArr {
			size += len(v)
		}
	} else {
		size += len(t.Value)
	}
	return size
}

// Copy creates a deep copy of the Tag.
func (t *Tag) Copy() Tag {
	var valueCopy []byte
	if t.Value != nil {
		valueCopy = make([]byte, len(t.Value))
		copy(valueCopy, t.Value)
	}
	var valueArrCopy [][]byte
	if t.ValueArr != nil {
		valueArrCopy = make([][]byte, len(t.ValueArr))
		for i, v := range t.ValueArr {
			valueArrCopy[i] = make([]byte, len(v))
			copy(valueArrCopy[i], v)
		}
	}
	return Tag{
		Name:      t.Name,
		Value:     valueCopy,
		ValueArr:  valueArrCopy,
		ValueType: t.ValueType,
	}
}

// Validate validates a WriteRequest for correctness.
func (wr WriteRequest) Validate() error {
	if wr.SeriesID == 0 {
		return fmt.Errorf("seriesID cannot be zero")
	}
	if wr.Data == nil {
		return fmt.Errorf("data cannot be nil")
	}
	if len(wr.Data) == 0 {
		return fmt.Errorf("data cannot be empty")
	}
	for i, tag := range wr.Tags {
		if tag.Name == "" {
			return fmt.Errorf("tag[%d] name cannot be empty", i)
		}
	}
	return nil
}

// Validate validates a QueryRequest for correctness.
func (qr QueryRequest) Validate() error {
	if len(qr.SeriesIDs) == 0 {
		return fmt.Errorf("at least one SeriesID is required")
	}
	if qr.MaxBatchSize < 0 {
		return fmt.Errorf("maxBatchSize cannot be negative")
	}
	if qr.MinKey != nil && qr.MaxKey != nil && *qr.MinKey > *qr.MaxKey {
		return fmt.Errorf("MinKey cannot be greater than MaxKey")
	}
	if qr.MinTimestamp != nil && qr.MaxTimestamp != nil && *qr.MinTimestamp > *qr.MaxTimestamp {
		return fmt.Errorf("MinTimestamp cannot be greater than MaxTimestamp")
	}
	return nil
}

// Validate validates a ScanQueryRequest for correctness.
func (sqr ScanQueryRequest) Validate() error {
	if sqr.MaxBatchSize < 0 {
		return fmt.Errorf("maxBatchSize cannot be negative")
	}
	if sqr.MinKey != nil && sqr.MaxKey != nil && *sqr.MinKey > *sqr.MaxKey {
		return fmt.Errorf("MinKey cannot be greater than MaxKey")
	}
	if sqr.MinTimestamp != nil && sqr.MaxTimestamp != nil && *sqr.MinTimestamp > *sqr.MaxTimestamp {
		return fmt.Errorf("MinTimestamp cannot be greater than MaxTimestamp")
	}
	return nil
}

// Reset resets the QueryRequest to its zero state.
func (qr *QueryRequest) Reset() {
	qr.SeriesIDs = nil
	qr.Filter = nil
	qr.Order = nil
	qr.TagProjection = nil
	qr.MaxBatchSize = 0
	qr.MinKey = nil
	qr.MaxKey = nil
	qr.MinTimestamp = nil
	qr.MaxTimestamp = nil
}

// CopyFrom copies the QueryRequest from other to qr.
func (qr *QueryRequest) CopyFrom(other *QueryRequest) {
	if other.SeriesIDs != nil {
		qr.SeriesIDs = make([]common.SeriesID, len(other.SeriesIDs))
		copy(qr.SeriesIDs, other.SeriesIDs)
	} else {
		qr.SeriesIDs = nil
	}

	qr.Filter = other.Filter
	qr.Order = other.Order

	if other.TagProjection != nil {
		qr.TagProjection = make([]model.TagProjection, len(other.TagProjection))
		copy(qr.TagProjection, other.TagProjection)
	} else {
		qr.TagProjection = nil
	}

	qr.MaxBatchSize = other.MaxBatchSize

	if other.MinKey != nil {
		minKey := *other.MinKey
		qr.MinKey = &minKey
	} else {
		qr.MinKey = nil
	}

	if other.MaxKey != nil {
		maxKey := *other.MaxKey
		qr.MaxKey = &maxKey
	} else {
		qr.MaxKey = nil
	}
	if other.MinTimestamp != nil {
		minTS := *other.MinTimestamp
		qr.MinTimestamp = &minTS
	} else {
		qr.MinTimestamp = nil
	}
	if other.MaxTimestamp != nil {
		maxTS := *other.MaxTimestamp
		qr.MaxTimestamp = &maxTS
	} else {
		qr.MaxTimestamp = nil
	}
}
