/**
 *
 */
package com.arms.config;

import com.arms.egovframework.javaservice.esframework.custom.client.CustomOpenSearchRestTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.conn.ConnectionKeepAliveStrategy;
import org.opensearch.client.HttpAsyncResponseConsumerFactory;
import org.opensearch.client.RequestOptions;
import org.opensearch.client.RestHighLevelClient;
import org.opensearch.data.client.orhlc.AbstractOpenSearchConfiguration;
import org.opensearch.data.client.orhlc.ClientConfiguration;
import org.opensearch.data.client.orhlc.RestClients;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.core.ElasticsearchOperations;
import org.springframework.data.elasticsearch.core.RefreshPolicy;
import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter;

import jakarta.annotation.PreDestroy;



@Slf4j
@Configuration
public class OpensearchClientConfig extends AbstractOpenSearchConfiguration {

	private final ElasticsearchProperties properties;

	private RestHighLevelClient client;

	@Value("${opensearch.buffer-limit-mb:300}")
	private int bufferLimitMb;

	public OpensearchClientConfig(ElasticsearchProperties properties) {
		this.properties = properties;
	}

	@Override
	@SuppressWarnings("java:S2095")
	public RestHighLevelClient opensearchClient() {

		final ClientConfiguration clientConfiguration =
				ClientConfiguration
					.builder()
					.connectedTo(properties.getUrl())
					.withClientConfigurer(RestClients.RestClientConfigurationCallback.from(
							httpAsyncClientBuilder -> {
								httpAsyncClientBuilder.setKeepAliveStrategy(this.connectionKeepAliveStrategy());
								return httpAsyncClientBuilder;
							}
					))
					.withConnectTimeout(60000)
					.withSocketTimeout(60000)
				.build();

		this.client = RestClients.create(clientConfiguration).rest();
		return this.client;
	}

	@PreDestroy
	public void closeClient() {
		try {
			if (client != null) {
				client.close();
			}
		} catch (Exception e) {
			log.error("Error closing Elasticsearch client: ", e);
		}
	}

	@Override
	protected RefreshPolicy refreshPolicy() {
		return RefreshPolicy.IMMEDIATE;
	}

	@Bean(name = "keepAliveStrategy")
	public ConnectionKeepAliveStrategy connectionKeepAliveStrategy() {
		return (response, context) -> 180*1000;
	}

	@Bean(name = "largeBufferRequestOptions")
	public RequestOptions largeBufferRequestOptions() {
		int bufferLimit = bufferLimitMb * 1024 * 1024;
		log.info("OpenSearch buffer limit: {}MB", bufferLimitMb);
		RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder();
		builder.setHttpAsyncResponseConsumerFactory(
				new HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory(bufferLimit));
		return builder.build();
	}

	@Override
	@Bean(name = {"elasticsearchOperations", "elasticsearchTemplate", "opensearchTemplate"})
	public ElasticsearchOperations elasticsearchOperations(
			ElasticsearchConverter elasticsearchConverter, RestHighLevelClient opensearchClient) {

		CustomOpenSearchRestTemplate template = new CustomOpenSearchRestTemplate(opensearchClient, elasticsearchConverter);
		template.setRefreshPolicy(refreshPolicy());

		return template;
	}
}