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

import (
	"fmt"
	"hash/crc32"
	"time"

	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"

	commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
	databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
	modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
	propertyv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/property/v1"
	schemav1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/schema/v1"
	"github.com/apache/skywalking-banyandb/banyand/metadata/schema"
)

// Tag key constants for schema properties.
const (
	TagKeyGroup     = "group"
	TagKeyName      = "name"
	TagKeySource    = "source"
	TagKeyKind      = "kind"
	TagKeyUpdatedAt = "updated_at"
	// TagKeyCreatedAt is the tag key for the first-appearance timestamp; survives updates unchanged.
	TagKeyCreatedAt = "created_at"
)

// BuildPropertyID returns the property ID for a schema resource.
// Format: kind.String() + "_" + group + "/" + name; for KindGroup: kind.String() + "_" + name.
func BuildPropertyID(kind schema.Kind, metadata *commonv1.Metadata) string {
	if kind == schema.KindGroup {
		return kind.String() + "_" + metadata.GetName()
	}
	return kind.String() + "_" + metadata.GetGroup() + "/" + metadata.GetName()
}

// BuildPropertyIDFromMeta builds a property ID from a TypeMeta.
func BuildPropertyIDFromMeta(meta schema.TypeMeta) string {
	if meta.Kind == schema.KindGroup {
		return meta.Kind.String() + "_" + meta.Name
	}
	return meta.Kind.String() + "_" + meta.Group + "/" + meta.Name
}

// SchemaToProperty converts a schema proto to a property for storage in the schema server.
func SchemaToProperty(kind schema.Kind, spec proto.Message) (*propertyv1.Property, error) {
	metadata, metaErr := getMetadataFromSpec(kind, spec)
	if metaErr != nil {
		return nil, metaErr
	}
	data, marshalErr := protojson.Marshal(spec)
	if marshalErr != nil {
		return nil, fmt.Errorf("failed to marshal schema to protojson: %w", marshalErr)
	}
	updatedAt, updErr := getUpdatedAtFromSpec(kind, spec)
	if updErr != nil {
		return nil, updErr
	}
	group := metadata.GetGroup()
	name := metadata.GetName()
	var updatedAtNano int64
	if updatedAt != nil {
		updatedAtNano = updatedAt.AsTime().UnixNano()
	}
	tags := []*modelv1.Tag{
		{Key: TagKeyGroup, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: group}}}},
		{Key: TagKeyName, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: name}}}},
		{Key: TagKeySource, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: string(data)}}}},
		{Key: TagKeyKind, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: kind.String()}}}},
		{Key: TagKeyUpdatedAt, Value: &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: updatedAtNano}}}},
	}
	// Only persist the created_at tag when it has been stamped (non-zero).
	if createdAt := getCreatedAtFromSpec(kind, spec); createdAt != nil {
		if createdAtNano := createdAt.AsTime().UnixNano(); createdAtNano != 0 {
			tags = append(tags, &modelv1.Tag{
				Key:   TagKeyCreatedAt,
				Value: &modelv1.TagValue{Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: createdAtNano}}},
			})
		}
	}
	return &propertyv1.Property{
		Metadata: &commonv1.Metadata{
			Name:        kind.String(),
			ModRevision: metadata.GetModRevision(),
		},
		Id:        BuildPropertyID(kind, metadata),
		Tags:      tags,
		UpdatedAt: updatedAt,
	}, nil
}

// ToSchema converts a property back to a schema Metadata.
func ToSchema(kind schema.Kind, prop *propertyv1.Property) (schema.Metadata, error) {
	if prop == nil {
		return schema.Metadata{}, fmt.Errorf("property is nil")
	}
	msg, newErr := newMessageForKind(kind)
	if newErr != nil {
		return schema.Metadata{}, newErr
	}
	parsed := ParseTags(prop.GetTags())
	if parsed.Source == "" {
		return schema.Metadata{}, fmt.Errorf("property %s/%s has no source tag", kind.String(), prop.GetId())
	}
	if unmarshalErr := protojson.Unmarshal([]byte(parsed.Source), msg); unmarshalErr != nil {
		return schema.Metadata{}, fmt.Errorf("failed to unmarshal schema from property: %w", unmarshalErr)
	}
	if hasMetadata, ok := msg.(schema.HasMetadata); ok && hasMetadata.GetMetadata() != nil {
		hasMetadata.GetMetadata().ModRevision = prop.GetMetadata().GetModRevision()
	}
	return schema.Metadata{
		TypeMeta: schema.TypeMeta{
			Kind:        kind,
			Name:        parsed.Name,
			Group:       parsed.Group,
			ModRevision: prop.GetMetadata().GetModRevision(),
		},
		Spec: msg,
	}, nil
}

// KindFromString returns the Kind for a given kind string (e.g. "stream", "measure").
func KindFromString(kindStr string) (schema.Kind, error) {
	for _, k := range schema.AllKinds() {
		if k.String() == kindStr {
			return k, nil
		}
	}
	return 0, fmt.Errorf("unknown kind string: %s", kindStr)
}

// kindFromProtoString maps the proto-style kind string from SchemaKey
// (underscore_case for compound kinds) to the corresponding schema.Kind. The
// proto schema keeps SchemaKey.kind on a fixed set of values: "stream",
// "measure", "trace", "property", "index_rule", "index_rule_binding",
// "group", "top_n_aggregation". An unknown value returns 0 and the caller
// should treat the SchemaKey as referring to no live entry.
func kindFromProtoString(protoKind string) schema.Kind {
	switch protoKind {
	case "stream":
		return schema.KindStream
	case "measure":
		return schema.KindMeasure
	case "trace":
		return schema.KindTrace
	case "property":
		return schema.KindProperty
	case "index_rule":
		return schema.KindIndexRule
	case "index_rule_binding":
		return schema.KindIndexRuleBinding
	case "group":
		return schema.KindGroup
	case "top_n_aggregation":
		return schema.KindTopNAggregation
	default:
		return 0
	}
}

// BuildPropertyIDFromSchemaKey converts a SchemaKey from the cluster
// node-status proto into the property-ID string used internally by the
// schema cache. Returns an empty string when the kind is unrecognized so
// the caller can short-circuit instead of looking up a malformed propID.
func BuildPropertyIDFromSchemaKey(key *schemav1.SchemaKey) string {
	kind := kindFromProtoString(key.GetKind())
	if kind == 0 {
		return ""
	}
	return BuildPropertyID(kind, &commonv1.Metadata{Group: key.GetGroup(), Name: key.GetName()})
}

// ParsedTags holds pre-extracted tag values from a property.
type ParsedTags struct {
	Group     string
	Name      string
	Source    string
	Kind      string
	UpdatedAt int64
	CreatedAt int64
}

// ParseTags extracts all known tag values in a single pass.
func ParseTags(tags []*modelv1.Tag) ParsedTags {
	var pt ParsedTags
	for _, tag := range tags {
		switch tag.GetKey() {
		case TagKeyGroup:
			pt.Group = tag.GetValue().GetStr().GetValue()
		case TagKeyName:
			pt.Name = tag.GetValue().GetStr().GetValue()
		case TagKeySource:
			pt.Source = tag.GetValue().GetStr().GetValue()
		case TagKeyKind:
			pt.Kind = tag.GetValue().GetStr().GetValue()
		case TagKeyUpdatedAt:
			pt.UpdatedAt = tag.GetValue().GetInt().GetValue()
		case TagKeyCreatedAt:
			pt.CreatedAt = tag.GetValue().GetInt().GetValue()
		}
	}
	return pt
}

// basicTagKeys are the tag keys used for metadata-only projections (full sync digest).
var basicTagKeys = []string{TagKeyKind, TagKeyGroup, TagKeyName, TagKeyUpdatedAt}

// buildRevisionCriteria builds a Criteria for updated_at > sinceRevision.
// Returns nil if sinceRevision <= 0.
func buildRevisionCriteria(sinceRevision int64) *modelv1.Criteria {
	if sinceRevision <= 0 {
		return nil
	}
	return &modelv1.Criteria{
		Exp: &modelv1.Criteria_Condition{
			Condition: &modelv1.Condition{
				Name: TagKeyUpdatedAt,
				Op:   modelv1.Condition_BINARY_OP_GT,
				Value: &modelv1.TagValue{
					Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: sinceRevision}},
				},
			},
		},
	}
}

// BuildSchemaQuery builds the property QueryRequest the schema server issues
// to look up schema docs of one kind, optionally narrowed to a group, a name
// and a minimum mod revision.
func BuildSchemaQuery(kind schema.Kind, group, name string, sinceRevision int64) *propertyv1.QueryRequest {
	query := &propertyv1.QueryRequest{
		Groups: []string{schema.SchemaGroup},
		Name:   kind.String(),
	}
	if name != "" {
		// for the group case, the group value could be empty
		metadata := &commonv1.Metadata{Group: group, Name: name}
		query.Ids = []string{BuildPropertyID(kind, metadata)}
		return query
	}
	var conditions []*modelv1.Criteria
	if group != "" {
		conditions = append(conditions, &modelv1.Criteria{
			Exp: &modelv1.Criteria_Condition{
				Condition: &modelv1.Condition{
					Name: TagKeyGroup,
					Op:   modelv1.Condition_BINARY_OP_EQ,
					Value: &modelv1.TagValue{
						Value: &modelv1.TagValue_Str{Str: &modelv1.Str{Value: group}},
					},
				},
			},
		})
	}
	if sinceRevision > 0 {
		conditions = append(conditions, &modelv1.Criteria{
			Exp: &modelv1.Criteria_Condition{
				Condition: &modelv1.Condition{
					Name: TagKeyUpdatedAt,
					Op:   modelv1.Condition_BINARY_OP_GT,
					Value: &modelv1.TagValue{
						Value: &modelv1.TagValue_Int{Int: &modelv1.Int{Value: sinceRevision}},
					},
				},
			},
		})
	}
	switch len(conditions) {
	case 1:
		query.Criteria = conditions[0]
	case 2:
		query.Criteria = &modelv1.Criteria{
			Exp: &modelv1.Criteria_Le{Le: &modelv1.LogicalExpression{
				Op:    modelv1.LogicalExpression_LOGICAL_OP_AND,
				Left:  conditions[0],
				Right: conditions[1],
			}},
		}
	}
	return query
}

// ParsePropID parses a property ID into kind, group, and name.
// Format: "<kind>_<group>/<name>" or "<kind>_<name>" (for KindGroup).
func ParsePropID(propID string) (kind schema.Kind, group, name string, err error) {
	underscoreIdx := -1
	for idx := range propID {
		if propID[idx] == '_' {
			underscoreIdx = idx
			break
		}
	}
	if underscoreIdx < 0 {
		return 0, "", "", fmt.Errorf("invalid propID format: %s", propID)
	}
	kindStr := propID[:underscoreIdx]
	rest := propID[underscoreIdx+1:]
	kind, err = KindFromString(kindStr)
	if err != nil {
		return 0, "", "", err
	}
	if kind == schema.KindGroup {
		return kind, "", rest, nil
	}
	slashIdx := -1
	for idx := range rest {
		if rest[idx] == '/' {
			slashIdx = idx
			break
		}
	}
	if slashIdx < 0 {
		return 0, "", "", fmt.Errorf("invalid propID format (missing group/name separator): %s", propID)
	}
	return kind, rest[:slashIdx], rest[slashIdx+1:], nil
}

// buildDeleteRequest builds a DeleteSchema request. The caller supplies deleteTime
// (unix nanos) so the value persisted as the tombstone on the server matches the
// value broadcastDelete returns to its caller — otherwise two separate time.Now()
// snapshots drift, breaking the design contract that delete_time is the
// authoritative tombstone timestamp.
func buildDeleteRequest(kind schema.Kind, group, name string, deleteTime int64) *schemav1.DeleteSchemaRequest {
	metadata := &commonv1.Metadata{Group: group, Name: name}
	return &schemav1.DeleteSchemaRequest{
		Delete: &propertyv1.DeleteRequest{
			Group: schema.SchemaGroup,
			Name:  kind.String(),
			Id:    BuildPropertyID(kind, metadata),
		},
		UpdateAt: timestamppb.New(time.Unix(0, deleteTime)),
	}
}

func getUpdatedAtFromSpec(kind schema.Kind, spec proto.Message) (*timestamppb.Timestamp, error) {
	var ts *timestamppb.Timestamp
	switch kind {
	case schema.KindGroup:
		if g, ok := spec.(*commonv1.Group); ok {
			ts = g.GetUpdatedAt()
		}
	case schema.KindStream:
		if s, ok := spec.(*databasev1.Stream); ok {
			ts = s.GetUpdatedAt()
		}
	case schema.KindMeasure:
		if m, ok := spec.(*databasev1.Measure); ok {
			ts = m.GetUpdatedAt()
		}
	case schema.KindTrace:
		if t, ok := spec.(*databasev1.Trace); ok {
			ts = t.GetUpdatedAt()
		}
	case schema.KindIndexRule:
		if ir, ok := spec.(*databasev1.IndexRule); ok {
			ts = ir.GetUpdatedAt()
		}
	case schema.KindIndexRuleBinding:
		if irb, ok := spec.(*databasev1.IndexRuleBinding); ok {
			ts = irb.GetUpdatedAt()
		}
	case schema.KindTopNAggregation:
		if t, ok := spec.(*databasev1.TopNAggregation); ok {
			ts = t.GetUpdatedAt()
		}
	case schema.KindProperty:
		if p, ok := spec.(*databasev1.Property); ok {
			ts = p.GetUpdatedAt()
		}
	case schema.KindNode:
		// Node does not have an UpdatedAt field.
	default:
		return nil, fmt.Errorf("unsupported kind: %s", kind)
	}
	if ts != nil {
		return ts, nil
	}
	return timestamppb.Now(), nil
}

// getCreatedAtFromSpec returns the created_at timestamp from a schema spec, or nil if not present.
func getCreatedAtFromSpec(kind schema.Kind, spec proto.Message) *timestamppb.Timestamp {
	switch kind {
	case schema.KindGroup:
		if g, ok := spec.(*commonv1.Group); ok {
			return g.GetCreatedAt()
		}
	case schema.KindStream:
		if s, ok := spec.(*databasev1.Stream); ok {
			return s.GetCreatedAt()
		}
	case schema.KindMeasure:
		if m, ok := spec.(*databasev1.Measure); ok {
			return m.GetCreatedAt()
		}
	case schema.KindTrace:
		if t, ok := spec.(*databasev1.Trace); ok {
			return t.GetCreatedAt()
		}
	case schema.KindIndexRule:
		if ir, ok := spec.(*databasev1.IndexRule); ok {
			return ir.GetCreatedAt()
		}
	case schema.KindIndexRuleBinding:
		if irb, ok := spec.(*databasev1.IndexRuleBinding); ok {
			return irb.GetCreatedAt()
		}
	case schema.KindTopNAggregation:
		if t, ok := spec.(*databasev1.TopNAggregation); ok {
			return t.GetCreatedAt()
		}
	case schema.KindProperty:
		if p, ok := spec.(*databasev1.Property); ok {
			return p.GetCreatedAt()
		}
	case schema.KindNode, schema.KindMask:
		// Node and Mask do not have a CreatedAt field.
	}
	return nil
}

// setCreatedAtOnSpec sets the created_at timestamp on a schema spec in-place.
func setCreatedAtOnSpec(kind schema.Kind, spec proto.Message, ts *timestamppb.Timestamp) {
	switch kind {
	case schema.KindGroup:
		if g, ok := spec.(*commonv1.Group); ok {
			g.CreatedAt = ts
		}
	case schema.KindStream:
		if s, ok := spec.(*databasev1.Stream); ok {
			s.CreatedAt = ts
		}
	case schema.KindMeasure:
		if m, ok := spec.(*databasev1.Measure); ok {
			m.CreatedAt = ts
		}
	case schema.KindTrace:
		if t, ok := spec.(*databasev1.Trace); ok {
			t.CreatedAt = ts
		}
	case schema.KindIndexRule:
		if ir, ok := spec.(*databasev1.IndexRule); ok {
			ir.CreatedAt = ts
		}
	case schema.KindIndexRuleBinding:
		if irb, ok := spec.(*databasev1.IndexRuleBinding); ok {
			irb.CreatedAt = ts
		}
	case schema.KindTopNAggregation:
		if t, ok := spec.(*databasev1.TopNAggregation); ok {
			t.CreatedAt = ts
		}
	case schema.KindProperty:
		if p, ok := spec.(*databasev1.Property); ok {
			p.CreatedAt = ts
		}
	case schema.KindNode, schema.KindMask:
		// Node and Mask do not have a CreatedAt field.
	}
}

func getMetadataFromSpec(kind schema.Kind, spec proto.Message) (*commonv1.Metadata, error) {
	switch kind {
	case schema.KindGroup:
		if g, ok := spec.(*commonv1.Group); ok {
			return g.GetMetadata(), nil
		}
	case schema.KindStream:
		if s, ok := spec.(*databasev1.Stream); ok {
			return s.GetMetadata(), nil
		}
	case schema.KindMeasure:
		if m, ok := spec.(*databasev1.Measure); ok {
			return m.GetMetadata(), nil
		}
	case schema.KindTrace:
		if t, ok := spec.(*databasev1.Trace); ok {
			return t.GetMetadata(), nil
		}
	case schema.KindIndexRule:
		if ir, ok := spec.(*databasev1.IndexRule); ok {
			return ir.GetMetadata(), nil
		}
	case schema.KindIndexRuleBinding:
		if irb, ok := spec.(*databasev1.IndexRuleBinding); ok {
			return irb.GetMetadata(), nil
		}
	case schema.KindTopNAggregation:
		if t, ok := spec.(*databasev1.TopNAggregation); ok {
			return t.GetMetadata(), nil
		}
	case schema.KindNode:
		if n, ok := spec.(*databasev1.Node); ok {
			return n.GetMetadata(), nil
		}
	case schema.KindProperty:
		if p, ok := spec.(*databasev1.Property); ok {
			return p.GetMetadata(), nil
		}
	case schema.KindMask:
		return nil, fmt.Errorf("unsupported kind: %s", kind)
	}
	return nil, fmt.Errorf("unsupported kind: %s", kind)
}

func newMessageForKind(kind schema.Kind) (proto.Message, error) {
	switch kind {
	case schema.KindGroup:
		return &commonv1.Group{}, nil
	case schema.KindStream:
		return &databasev1.Stream{}, nil
	case schema.KindMeasure:
		return &databasev1.Measure{}, nil
	case schema.KindTrace:
		return &databasev1.Trace{}, nil
	case schema.KindIndexRuleBinding:
		return &databasev1.IndexRuleBinding{}, nil
	case schema.KindIndexRule:
		return &databasev1.IndexRule{}, nil
	case schema.KindTopNAggregation:
		return &databasev1.TopNAggregation{}, nil
	case schema.KindNode:
		return &databasev1.Node{}, nil
	case schema.KindProperty:
		return &databasev1.Property{}, nil
	default:
		return nil, schema.ErrUnsupportedEntityType
	}
}

// generateCRC32ID generates a CRC32 ID for an index rule.
func generateCRC32ID(group, name string) uint32 {
	buf := []byte(group)
	buf = append(buf, name...)
	return crc32.ChecksumIEEE(buf)
}
