return err
		}
	}
	if k.QueueDir != "" {
		if !filepath.IsAbs(k.QueueDir) {
			return errors.New("queueDir path should be absolute")
		}
	}
	if k.Version != "" {
		if _, err := sarama.ParseKafkaVersion(k.Version); err != nil {
			return err
		}
	}
	if k.BatchSize > 1 {
		if k.QueueDir == "" {
			return errors.New("batch should be enabled only if queue dir is enabled")

func NewNATSTarget(id string, args NATSArgs, loggerOnce logger.LogOnce) (*NATSTarget, error) {
	var queueStore store.Store[event.Event]
	if args.QueueDir != "" {
		queueDir := filepath.Join(args.QueueDir, storePrefix+"-nats-"+id)
		queueStore = store.NewQueueStore[event.Event](queueDir, args.QueueLimit, event.StoreExtension)
		if err := queueStore.Open(); err != nil {

				}
				return strings.Join(brokers, config.ValueSeparator)
			}(),
		},
		config.KV{
			Key:   target.KafkaTopic,
			Value: cfg.Topic,
		},
		config.KV{
			Key:   target.KafkaQueueDir,
			Value: cfg.QueueDir,
		},
		config.KV{
			Key:   target.KafkaClientTLSCert,
			Value: cfg.TLS.ClientTLSCert,
		},
		config.KV{
			Key:   target.KafkaClientTLSKey,
			Value: cfg.TLS.ClientTLSKey,
		},
		config.KV{

func NewRedisTarget(id string, args RedisArgs, loggerOnce logger.LogOnce) (*RedisTarget, error) {
	var queueStore store.Store[event.Event]
	if args.QueueDir != "" {
		queueDir := filepath.Join(args.QueueDir, storePrefix+"-redis-"+id)
		queueStore = store.NewQueueStore[event.Event](queueDir, args.QueueLimit, event.StoreExtension)
		if err := queueStore.Open(); err != nil {

			ClientCert: clientCert,
			ClientKey:  clientKey,
			Proxy:      getCfgVal(EnvLoggerWebhookProxy, k, kv.Get(Proxy)),
			BatchSize:  batchSize,
			QueueSize:  queueSize,
			QueueDir:   getCfgVal(EnvLoggerWebhookQueueDir, k, kv.Get(QueueDir)),
			Name:       loggerTargetNamePrefix + k,
		}
	}
	return cfg, nil
}

func lookupAuditWebhookConfig(scfg config.Config, cfg Config) (Config, error) {

		}
	}
	connStr := strings.Join(params, " ")

	var queueStore store.Store[event.Event]
	if args.QueueDir != "" {
		queueDir := filepath.Join(args.QueueDir, storePrefix+"-postgresql-"+id)
		queueStore = store.NewQueueStore[event.Event](queueDir, args.QueueLimit, event.StoreExtension)
		if err := queueStore.Open(); err != nil {

	ClientCert string            `json:"clientCert"`
	ClientKey  string            `json:"clientKey"`
	BatchSize  int               `json:"batchSize"`
	QueueSize  int               `json:"queueSize"`
	QueueDir   string            `json:"queueDir"`
	Proxy      string            `json:"string"`
	Transport  http.RoundTripper `json:"-"`

	// Custom logger
	LogOnceIf func(ctx context.Context, err error, id string, errKind ...interface{}) `json:"-"`

	return nil
}

func (h *Target) initQueueStore(ctx context.Context) (err error) {
	var queueStore store.Store[interface{}]
	queueDir := filepath.Join(h.kconfig.QueueDir, h.Name())
	queueStore = store.NewQueueStore[interface{}](queueDir, uint64(h.kconfig.QueueSize), kafkaLoggerExtension)
	if err = queueStore.Open(); err != nil {
		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
	}

		if err != nil {
			return nil, err
		}

		kafkaArgs := target.KafkaArgs{
			Enable:     enabled,
			Brokers:    brokers,
			Topic:      env.Get(topicEnv, kv.Get(target.KafkaTopic)),
			QueueDir:   env.Get(queueDirEnv, kv.Get(target.KafkaQueueDir)),
			QueueLimit: queueLimit,
			Version:    env.Get(versionEnv, kv.Get(target.KafkaVersion)),
			BatchSize:  uint32(batchSize),
		}

		},
		config.HelpKV{
			Key:         QueueSize,
			Description: "configure channel queue size for webhook targets",
			Optional:    true,
			Type:        "number",
		},
		config.HelpKV{
			Key:         QueueDir,
			Description: `staging dir for undelivered logger messages e.g. '/home/logger-events'`,
			Optional:    true,
			Type:        "string",
		},
		config.HelpKV{
			Key:         Proxy,