/* Hibernate, Relational Persistence for Idiomatic Java
 *
 * SPDX-License-Identifier: Apache-2.0
 * Copyright: Red Hat Inc. and Hibernate Authors
 */
package org.hibernate.reactive.provider.service;

import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CountDownLatch;

import org.hibernate.reactive.logging.impl.Log;
import org.hibernate.reactive.logging.impl.LoggerFactory;
import org.hibernate.reactive.pool.ReactiveConnection;
import org.hibernate.reactive.pool.ReactiveConnectionPool;
import org.hibernate.reactive.vertx.VertxInstance;
import org.hibernate.service.ServiceRegistry;
import org.hibernate.tool.schema.internal.exec.GenerationTarget;

import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;

/**
 * Adaptor that redirects DDL generated by the schema export
 * tool to the reactive connection.
 */
public class ReactiveGenerationTarget implements GenerationTarget {
	private static final Log log = LoggerFactory.make( Log.class, MethodHandles.lookup() );

	private final ServiceRegistry registry;
	private VertxInstance vertxSupplier;
	private ReactiveConnectionPool service;
	private Set<String> statements;
	private final List<String> commands = new ArrayList<>();

	private volatile CountDownLatch done;

	public ReactiveGenerationTarget(ServiceRegistry registry) {
		this.registry = registry;
	}

	@Override
	public void prepare() {
		service = registry.getService( ReactiveConnectionPool.class );
		vertxSupplier = registry.getService( VertxInstance.class );
		statements = new HashSet<>();
		done = new CountDownLatch( 1 );
	}

	@Override
	public void accept(String command) {
		// avoid executing duplicate DDL statements
		// (hack specifically to avoid multiple
		// inserts into a sequence emulation table)
		if ( statements.add( command ) ) {
			commands.add( command );
		}
	}

	@Override
	public void release() {
		statements = null;
		if ( !commands.isEmpty() ) {
			vertxSupplier.getVertx().getOrCreateContext().runOnContext( v1 ->
					service.getConnection()
						.thenCompose( this::executeCommands )
						// An error could have happened getting the connection (executeCommands will hide all other exceptions)
						.handle( ReactiveGenerationTarget::logCommandFailure )
						.thenAccept( v -> done.countDown() )
			);

			if ( done != null ) {
				try {
					done.await();
				}
				catch (InterruptedException e) {
					log.warnf( "Interrupted while performing schema export operations", e.getMessage() );
					Thread.currentThread().interrupt();
				}
			}
		}
	}

	/**
	 * Execute all commands and log exceptions without propagating them.
	 * This method never fails.
	 */
	private CompletionStage<Void> executeCommands(ReactiveConnection reactiveConnection) {
		CompletionStage<Void> result = voidFuture();
		for ( String command : commands ) {
			result = result.thenApply( v -> command )
					.thenCompose( reactiveConnection::execute )
					.handle( ReactiveGenerationTarget::logCommandFailure );
		}
		return result
				.thenApply( v -> reactiveConnection )
				.thenCompose( ReactiveConnection::close )
				// In case there is a failure closing the connection
				.handle( ReactiveGenerationTarget::logCommandFailure );
	}

	private static <U> U logCommandFailure(Void ignore, Throwable throwable) {
		if ( throwable != null ) {
			log.ddlCommandFailed( throwable.getMessage() );
		}
		return null;
	}
}
