package kas

import (
	"context"
	"encoding/binary"
	"time"

	"github.com/redis/rueidis"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/nettool"
	"gitlab.com/gitlab-org/cluster-integration/gitlab-agent/v18/internal/tool/redistool"
	"go.opentelemetry.io/otel/attribute"
	otelmetric "go.opentelemetry.io/otel/metric"
)

const (
	setUnsetTimeout                             = 5 * time.Second
	acceptedConnectionsMetricName               = "accepted_connections"
	closedConnectionsMetricName                 = "closed_connections"
	nameAttr                      attribute.Key = "name"
)

var (
	_ nettool.NewConnectionSet = (*connectionSetFactory)(nil).newConnectionSet
	_ nettool.ConnectionSet    = (*connectionSet)(nil)
)

type connectionSetFactory struct {
	instanceID int64
	redis      rueidis.Client
	keyPrefix  string
	m          otelmetric.Meter
}

func (f *connectionSetFactory) connID2RedisKey(connID int64) string {
	key := make([]byte, 0, 8+8)                                       // int64 + int64. Don't need a separator since widths are fixed.
	key = binary.LittleEndian.AppendUint64(key, uint64(f.instanceID)) //nolint: gosec
	key = binary.LittleEndian.AppendUint64(key, uint64(connID))       //nolint: gosec
	return rueidis.BinaryString(key)
}

func (f *connectionSetFactory) newConnectionSet(name string, ttl time.Duration) (nettool.ConnectionSet, error) {
	acceptedCounter, err := f.m.Int64Counter(
		acceptedConnectionsMetricName,
		otelmetric.WithDescription("Number of accepted connections"),
	)
	if err != nil {
		return nil, err
	}
	closedCounter, err := f.m.Int64Counter(
		closedConnectionsMetricName,
		otelmetric.WithDescription("Number of closed connections"),
	)
	if err != nil {
		return nil, err
	}

	// We always use the same key because we only use a single key. Precompute it once.
	key := f.keyPrefix + ":listener_conns:" + name
	cs := &connectionSet{
		acceptedCounter: acceptedCounter,
		closedCounter:   closedCounter,
		counterAddOpts: []otelmetric.AddOption{
			otelmetric.WithAttributeSet(attribute.NewSet(
				nameAttr.String(name),
			)),
		}, // allocate slice once
		key:  key,
		keys: []string{key}, // allocate slice once
		ttl:  ttl,
	}
	cs.set, err = redistool.NewRedisExpiringHash(
		"listener_conns_"+name,
		f.redis,
		cs.k1,
		f.connID2RedisKey,
		ttl,
		f.m,
		false,
	)
	if err != nil {
		return nil, err
	}
	return cs, nil
}

type connectionSet struct {
	set             redistool.ExpiringHash[string, int64]
	acceptedCounter otelmetric.Int64Counter
	closedCounter   otelmetric.Int64Counter
	counterAddOpts  []otelmetric.AddOption
	key             string
	keys            []string
	ttl             time.Duration
}

func (c *connectionSet) Set(connID int64) error {
	c.acceptedCounter.Add(context.Background(), 1, c.counterAddOpts...)
	ctx, cancel := context.WithTimeout(context.Background(), setUnsetTimeout)
	defer cancel()
	return c.set.SetEX(ctx, "" /*unused*/, connID, nil, time.Now().Add(c.ttl))
}

func (c *connectionSet) Unset(connID int64) error {
	c.closedCounter.Add(context.Background(), 1, c.counterAddOpts...)
	ctx, cancel := context.WithTimeout(context.Background(), setUnsetTimeout)
	defer cancel()
	return c.set.DelEX(ctx, "" /*unused*/, connID)
}

func (c *connectionSet) GC(ctx context.Context) error {
	_, err := c.set.GCFor(c.keys)(ctx)
	return err
}

func (c *connectionSet) k1(_ string) string {
	return c.key
}
