content: package pl.allegro.tech.hermes.management.config.storage;
  
  import com.fasterxml.jackson.databind.ObjectMapper;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.beans.factory.annotation.Qualifier;
  import org.springframework.boot.context.properties.EnableConfigurationProperties;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Configuration;
  import org.springframework.context.annotation.Primary;
  import pl.allegro.tech.hermes.domain.CredentialsRepository;
  import pl.allegro.tech.hermes.domain.group.GroupRepository;
  import pl.allegro.tech.hermes.domain.oauth.OAuthProviderRepository;
  import pl.allegro.tech.hermes.domain.subscription.SubscriptionRepository;
  import pl.allegro.tech.hermes.domain.topic.TopicRepository;
  import pl.allegro.tech.hermes.domain.topic.preview.MessagePreviewRepository;
  import pl.allegro.tech.hermes.domain.workload.constraints.WorkloadConstraintsRepository;
  import pl.allegro.tech.hermes.infrastructure.dc.DatacenterNameProvider;
  import pl.allegro.tech.hermes.infrastructure.dc.DcNameSource;
  import pl.allegro.tech.hermes.infrastructure.dc.DefaultDatacenterNameProvider;
  import pl.allegro.tech.hermes.infrastructure.dc.EnvironmentVariableDatacenterNameProvider;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperCredentialsRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperGroupRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperMessagePreviewRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperOAuthProviderRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperPaths;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperSubscriptionRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperTopicRepository;
  import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperWorkloadConstraintsRepository;
  import pl.allegro.tech.hermes.management.domain.dc.MultiDatacenterRepositoryCommandExecutor;
  import pl.allegro.tech.hermes.management.domain.detection.InactiveTopicsRepository;
  import pl.allegro.tech.hermes.management.domain.mode.ModeService;
  import pl.allegro.tech.hermes.management.domain.readiness.DatacenterReadinessRepository;
  import pl.allegro.tech.hermes.management.domain.retransmit.OfflineRetransmissionRepository;
  import pl.allegro.tech.hermes.management.infrastructure.detection.ZookeeperInactiveTopicsRepository;
  import pl.allegro.tech.hermes.management.infrastructure.metrics.SummedSharedCounter;
  import pl.allegro.tech.hermes.management.infrastructure.readiness.ZookeeperDatacenterReadinessRepository;
  import pl.allegro.tech.hermes.management.infrastructure.retransmit.ZookeeperOfflineRetransmissionRepository;
  import pl.allegro.tech.hermes.management.infrastructure.zookeeper.ZookeeperClient;
  import pl.allegro.tech.hermes.management.infrastructure.zookeeper.ZookeeperClientManager;
  import pl.allegro.tech.hermes.management.infrastructure.zookeeper.ZookeeperRepositoryManager;
  
  @Configuration
  @EnableConfigurationProperties(StorageClustersProperties.class)
  public class StorageConfiguration {
  
    @Autowired StorageClustersProperties storageClustersProperties;
  
    @Autowired ObjectMapper objectMapper;
  
    @Bean
    DatacenterNameProvider dcNameProvider() {
      if (storageClustersProperties.getDatacenterNameSource() == DcNameSource.ENV) {
        return new EnvironmentVariableDatacenterNameProvider(
            storageClustersProperties.getDatacenterNameSourceEnv());
      } else {
        return new DefaultDatacenterNameProvider();
      }
    }
  
    @Bean(initMethod = "start", destroyMethod = "stop")
    ZookeeperClientManager clientManager() {
      return new ZookeeperClientManager(storageClustersProperties, dcNameProvider());
    }
  
    @Bean
    ZookeeperGroupRepositoryFactory zookeeperGroupRepositoryFactory() {
      return new DefaultZookeeperGroupRepositoryFactory();
    }
  
    @Bean(initMethod = "start")
    ZookeeperRepositoryManager repositoryManager(
        ZookeeperGroupRepositoryFactory zookeeperGroupRepositoryFactory) {
      return new ZookeeperRepositoryManager(
          clientManager(),
          dcNameProvider(),
          objectMapper,
          zookeeperPaths(),
          zookeeperGroupRepositoryFactory);
    }
  
    @Bean
    ZookeeperPaths zookeeperPaths() {
      return new ZookeeperPaths(storageClustersProperties.getPathPrefix());
    }
  
    @Bean
    MultiDatacenterRepositoryCommandExecutor multiDcRepositoryCommandExecutor(
        ZookeeperGroupRepositoryFactory zookeeperGroupRepositoryFactory, ModeService modeService) {
      return new MultiDatacenterRepositoryCommandExecutor(
          repositoryManager(zookeeperGroupRepositoryFactory),
          storageClustersProperties.isTransactional(),
          modeService);
    }
  
    @Bean
    SummedSharedCounter summedSharedCounter(ZookeeperClientManager manager) {
      return new SummedSharedCounter(
          manager.getClients(),
          storageClustersProperties.getSharedCountersExpiration(),
          storageClustersProperties.getRetrySleep(),
          storageClustersProperties.getRetryTimes());
    }
  
    @Bean
    GroupRepository groupRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperGroupRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    CredentialsRepository credentialsRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperCredentialsRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    TopicRepository topicRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperTopicRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths(), groupRepository());
    }
  
    @Bean
    SubscriptionRepository subscriptionRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperSubscriptionRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths(), topicRepository());
    }
  
    @Bean
    OAuthProviderRepository oAuthProviderRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperOAuthProviderRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    MessagePreviewRepository messagePreviewRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperMessagePreviewRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    WorkloadConstraintsRepository workloadConstraintsRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperWorkloadConstraintsRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    @Primary
    @Qualifier("zookeeperOfflineRetransmissionRepository")
    OfflineRetransmissionRepository zookeeperOfflineRetransmissionRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperOfflineRetransmissionRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    DatacenterReadinessRepository readinessRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperDatacenterReadinessRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  
    @Bean
    InactiveTopicsRepository inactiveTopicsRepository() {
      ZookeeperClient localClient = clientManager().getLocalClient();
      return new ZookeeperInactiveTopicsRepository(
          localClient.getCuratorFramework(), objectMapper, zookeeperPaths());
    }
  }
  