package com.pingcap.kb.integration.config;

import java.time.Duration;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.stream.StreamReceiver;
import org.springframework.integration.channel.RendezvousChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.redis.inbound.ReactiveRedisStreamMessageProducer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;

@Slf4j
@ConfigurationProperties(prefix = "kb.redis.stream")
@Configuration
public class RedisIntegrationConfiguration {

  @Getter
  @Setter
  private String key;

  @Getter
  @Setter
  private String consumerGroup;

  @Getter
  @Setter
  private String consumerName;


  @Bean("kb.event")
  MessageChannel kbEventChannel() {
    return new RendezvousChannel();
  }

  @Bean
  IntegrationFlow eventFlow(
      @Qualifier("kb.event") MessageChannel kbEventChannel) {
    return IntegrationFlows
        .from(kbEventChannel)
        .handle("MqEndpoints", "receive")
        .get();
  }

  @Bean
  public ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer(
      ReactiveRedisConnectionFactory reactiveRedisConnectionFactory,
      @Qualifier("kb.event") MessageChannel targetChannel) {
    ReactiveRedisStreamMessageProducer messageProducer =
        new ReactiveRedisStreamMessageProducer(reactiveRedisConnectionFactory, getKey());
    messageProducer.setStreamReceiverOptions(
        StreamReceiver.StreamReceiverOptions.builder()
            .pollTimeout(Duration.ofMillis(100))
            .build());
    messageProducer.setAutoStartup(true);
    messageProducer.setAutoAck(false);
    messageProducer.setCreateConsumerGroup(true);
    messageProducer.setConsumerGroup(getConsumerGroup());
    messageProducer.setConsumerName(getConsumerName());
    messageProducer.setOutputChannel(targetChannel);
    messageProducer.setReadOffset(ReadOffset.lastConsumed());
    messageProducer.setExtractPayload(true);
    return messageProducer;
  }
}