handler

var EvalFlag = func(evalContext models.EvalContext) *models.EvalResult {
	flag := LookupFlag(evalContext)
	return EvalFlagWithContext(flag, evalContext)
}

var EvalFlagWithContext = func(flag *entity.Flag, evalContext models.EvalContext) *models.EvalResult {
	flagID := util.SafeUint(evalContext.FlagID)
	flagKey := util.SafeString(evalContext.FlagKey)

	if flag == nil {
		emptyFlag := &entity.Flag{Model: gorm.Model{ID: flagID}, Key: flagKey}
		return BlankResult(emptyFlag, evalContext, fmt.Sprintf("flagID %v not found or deleted", flagID))
	}

	if !flag.Enabled {
		return BlankResult(flag, evalContext, fmt.Sprintf("flagID %v is not enabled", flag.ID))
	}

	if len(flag.Segments) == 0 {
		return BlankResult(flag, evalContext, fmt.Sprintf("flagID %v has no segments", flag.ID))
	}

	if evalContext.EntityID == "" {
		evalContext.EntityID = fmt.Sprintf("randomly_generated_%d", rand.Int31())
	}

	if flag.EntityType != "" {
		evalContext.EntityType = flag.EntityType
	}

	logs := []*models.SegmentDebugLog{}
	var vID int64
	var sID int64

	for _, segment := range flag.Segments {
		sID = int64(segment.ID)
		variantID, log, evalNextSegment := evalSegment(flag.ID, evalContext, segment)
		if config.Config.EvalDebugEnabled && evalContext.EnableDebug {
			logs = append(logs, log)
		}
		if variantID != nil {
			vID = int64(*variantID)
		}
		if !evalNextSegment {
			break
		}
	}
	evalResult := BlankResult(flag, evalContext, "")
	evalResult.EvalDebugLog.SegmentDebugLogs = logs
	evalResult.SegmentID = sID
	evalResult.VariantID = vID
	v := flag.FlagEvaluation.VariantsMap[util.SafeUint(vID)]
	if v != nil {
		evalResult.VariantAttachment = v.Attachment
		evalResult.VariantKey = v.Key
	}

	logEvalResult(evalResult, flag.DataRecordsEnabled)
	evalResult.DataRecordsEnabled = flag.DataRecordsEnabled
	return evalResult
}

var EvalFlagsByTags = func(evalContext models.EvalContext) []*models.EvalResult {
	cache := GetEvalCache()
	fs := cache.GetByTags(evalContext.FlagTags, evalContext.FlagTagsOperator)
	results := []*models.EvalResult{}
	for _, f := range fs {
		results = append(results, EvalFlagWithContext(f, evalContext))
	}
	return results
}

var GetEvalCache = func() *EvalCache {
	singletonEvalCacheOnce.Do(func() {
		ec := &EvalCache{
			cache:           &cacheContainer{},
			refreshTimeout:  config.Config.EvalCacheRefreshTimeout,
			refreshInterval: config.Config.EvalCacheRefreshInterval,
		}
		singletonEvalCache = ec
	})
	return singletonEvalCache
}

var LookupFlag = func(evalContext models.EvalContext) *entity.Flag {
	cache := GetEvalCache()
	flagID := util.SafeUint(evalContext.FlagID)
	flagKey := util.SafeString(evalContext.FlagKey)
	f := cache.GetByFlagKeyOrID(flagID)
	if f == nil {
		f = cache.GetByFlagKeyOrID(flagKey)
	}
	return f
}

var NewKafkaRecorder = func() DataRecorder {
	if config.Config.RecorderKafkaVerbose {
		sarama.Logger = logrus.StandardLogger()
	}

	cfg := sarama.NewConfig()

	tlscfg := createTLSConfiguration(
		config.Config.RecorderKafkaCertFile,
		config.Config.RecorderKafkaKeyFile,
		config.Config.RecorderKafkaCAFile,
		config.Config.RecorderKafkaVerifySSL,
		config.Config.RecorderKafkaSimpleSSL,
	)
	if tlscfg != nil {
		cfg.Net.TLS.Enable = true
		cfg.Net.TLS.Config = tlscfg
	}

	if config.Config.RecorderKafkaSASLUsername != "" && config.Config.RecorderKafkaSASLPassword != "" {
		cfg.Net.SASL.Enable = true
		cfg.Net.SASL.User = config.Config.RecorderKafkaSASLUsername
		cfg.Net.SASL.Password = config.Config.RecorderKafkaSASLPassword
	}

	cfg.Net.MaxOpenRequests = config.Config.RecorderKafkaMaxOpenReqs

	cfg.Producer.Compression = sarama.CompressionCodec(config.Config.RecorderKafkaCompressionCodec)
	cfg.Producer.RequiredAcks = sarama.RequiredAcks(config.Config.RecorderKafkaRequiredAcks)
	cfg.Producer.Idempotent = config.Config.RecorderKafkaIdempotent
	cfg.Producer.Retry.Max = config.Config.RecorderKafkaRetryMax
	cfg.Producer.Flush.Frequency = config.Config.RecorderKafkaFlushFrequency
	cfg.Version = mustParseKafkaVersion(config.Config.RecorderKafkaVersion)

	if cfg.Producer.Idempotent {
		cfg.Producer.RequiredAcks = sarama.WaitForAll
		cfg.Net.MaxOpenRequests = 1
		if !cfg.Version.IsAtLeast(sarama.V0_11_0_0) {
			cfg.Version = sarama.V0_11_0_0
		}
		logrus.Info("Idempotent producer enabled: set RequiredAcks=WaitForAll, MaxOpenRequests=1, Version>=0.11.0.0")
	}

	brokerList := strings.Split(config.Config.RecorderKafkaBrokers, ",")
	producer, err := saramaNewAsyncProducer(brokerList, cfg)
	if err != nil {
		logrus.WithField("kafka_error", err).Fatal("Failed to start Sarama producer:")
	}

	var encryptor dataRecordEncryptor
	if config.Config.RecorderKafkaEncrypted && config.Config.RecorderKafkaEncryptionKey != "" {
		encryptor = newSimpleboxEncryptor(config.Config.RecorderKafkaEncryptionKey)
	}

	bufSize := config.Config.RecorderKafkaBufferSize
	if bufSize < 1 {
		bufSize = 10000
		logrus.Warn("RecorderKafkaBufferSize < 1, using default 10000")
	}

	workerCount := config.Config.RecorderKafkaWorkerCount
	if workerCount < 1 {
		workerCount = 4
		logrus.Warn("RecorderKafkaWorkerCount < 1, using default 4")
	}

	recorder := &kafkaRecorder{
		topic:               config.Config.RecorderKafkaTopic,
		partitionKeyEnabled: config.Config.RecorderKafkaPartitionKeyEnabled,
		producer:            producer,
		recordCh:            make(chan models.EvalResult, bufSize),
		options: DataRecordFrameOptions{
			Encrypted:       config.Config.RecorderKafkaEncrypted,
			Encryptor:       encryptor,
			FrameOutputMode: config.Config.RecorderFrameOutputMode,
		},
	}

	recorder.errWg.Add(1)
	go func() {
		defer recorder.errWg.Done()
		for err := range producer.Errors() {
			logrus.WithField("kafka_error", err).Error("failed to write access log entry")
			if config.Global.Prometheus.RecorderErrors != nil {
				config.Global.Prometheus.RecorderErrors.Inc()
			}
		}
	}()

	if config.Config.PrometheusEnabled {
		promauto.NewGaugeFunc(prometheus.GaugeOpts{
			Name: "flagr_recorder_buffer_usage",
			Help: "Current number of records in the async buffer",
		}, func() float64 {
			return float64(len(recorder.recordCh))
		})
	}

	for i := 0; i < workerCount; i++ {
		recorder.startWorker()
	}

	return recorder
}

var NewKinesisRecorder = func() DataRecorder {
	cfg, err := config.LoadDefaultConfig(context.TODO())
	if err != nil {
		logrus.WithField("kinesis_error", err).Fatal("error creating aws session")
	}

	client := kinesis.NewFromConfig(cfg)

	p := newKinesisProducer(&producer.Config{
		StreamName:          flagrConfig.Config.RecorderKinesisStreamName,
		Client:              client,
		BacklogCount:        flagrConfig.Config.RecorderKinesisBacklogCount,
		MaxConnections:      flagrConfig.Config.RecorderKinesisMaxConnections,
		FlushInterval:       flagrConfig.Config.RecorderKinesisFlushInterval,
		BatchSize:           flagrConfig.Config.RecorderKinesisBatchSize,
		BatchCount:          flagrConfig.Config.RecorderKinesisBatchCount,
		AggregateBatchCount: flagrConfig.Config.RecorderKinesisAggregateBatchCount,
		AggregateBatchSize:  flagrConfig.Config.RecorderKinesisAggregateBatchSize,
		Verbose:             flagrConfig.Config.RecorderKinesisVerbose,
		Logger:              &kplogrus.Logger{Logger: logrus.StandardLogger()},
	})

	p.Start()

	go func() {
		for err := range p.NotifyFailures() {
			logrus.WithField("kinesis_error", err).Error("error pushing to kinesis")
		}
	}()

	return &kinesisRecorder{
		producer: p,
		options: DataRecordFrameOptions{
			Encrypted:       false,
			FrameOutputMode: flagrConfig.Config.RecorderFrameOutputMode,
		},
	}
}

var NewPubsubRecorder = func() DataRecorder {
	client, err := pubsubClient()
	if err != nil {
		logrus.WithField("pubsub_error", err).Fatal("error getting pubsub client")
	}

	return &pubsubRecorder{
		producer:  client,
		publisher: client.Publisher(config.Config.RecorderPubsubTopicName),
		options: DataRecordFrameOptions{
			Encrypted:       false,
			FrameOutputMode: config.Config.RecorderFrameOutputMode,
		},
	}
}