/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */
package groovy.concurrent;

import org.apache.groovy.runtime.async.AsyncSupport;
import org.apache.groovy.runtime.async.FlowPublisherAdapter;
import org.apache.groovy.runtime.async.GroovyPromise;

import java.lang.ref.SoftReference;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.ServiceLoader;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Central registry for {@link AwaitableAdapter} instances.
 * <p>
 * On class-load, adapters are discovered via {@link ServiceLoader} from
 * {@code META-INF/services/groovy.concurrent.AwaitableAdapter}. Two built-in
 * adapters are always present after SPI-loaded ones:
 * <ul>
 *   <li>{@link FlowPublisherAdapter} for {@link java.util.concurrent.Flow.Publisher}
 *       (single-value {@code await} and multi-value {@code for await}).</li>
 *   <li>A {@code Future} fallback handling {@link CompletableFuture},
 *       {@link CompletionStage}, and {@link Future} (via a blocking wrapper).</li>
 * </ul>
 *
 * @since 6.0.0
 */
public final class AwaitableAdapterRegistry {

    private static final List<AwaitableAdapter> adapters = new CopyOnWriteArrayList<>();

    /**
     * Marks a type no adapter supports, so that a cache entry whose {@link SoftReference}
     * was cleared can be told apart from a type that resolved to no adapter.
     */
    private static final AwaitableAdapter NO_ADAPTER = new AwaitableAdapter() {
        @Override
        public boolean supportsAwaitable(Class<?> type) {
            return false;
        }

        @Override
        public <T> Awaitable<T> toAwaitable(Object source) {
            throw new IllegalStateException("NO_ADAPTER cannot adapt");
        }
    };

    /**
     * Adapter lookups cached per source class. A {@code ClassValue} association lives as
     * long as its key class, and the common keys here are platform classes such as
     * {@link CompletableFuture}, so the value is held through a {@link SoftReference}:
     * the association then strongly reaches only {@code java.base} objects and never pins
     * the adapter's class loader (GROOVY-12280). A cleared reference is re-resolved from
     * {@link #adapters} on the next lookup. The {@link AtomicReference} swap is how
     * {@link #register} / {@link #unregister} publish a fresh {@code ClassValue};
     * {@code ClassValue} itself is already thread-safe for concurrent {@code get}.
     */
    private static final AtomicReference<ClassValue<SoftReference<AwaitableAdapter>>> awaitableCache =
            new AtomicReference<>(buildAwaitableCache());

    static {
        // Load SPI adapters
        try {
            ServiceLoader<AwaitableAdapter> loader = ServiceLoader.load(
                    AwaitableAdapter.class, AwaitableAdapterRegistry.class.getClassLoader());
            for (Iterator<AwaitableAdapter> it = loader.iterator(); it.hasNext(); ) {
                try {
                    adapters.add(it.next());
                } catch (Throwable ignored) {
                    // Skip adapters that fail to load (missing dependencies)
                }
            }
        } catch (Throwable ignored) {
            // ServiceLoader failure — continue with built-in adapter only
        }
        // Built-in JDK Flow.Publisher adapter (after SPI, before Future fallback).
        // SPI adapters take precedence so framework-specific types resolve first.
        adapters.add(new FlowPublisherAdapter());
        // Built-in fallback adapter (lowest priority)
        adapters.add(new BuiltInAdapter());
    }

    private AwaitableAdapterRegistry() { }

    /**
     * Registers an adapter at the highest priority (before SPI-loaded adapters).
     */
    public static void register(AwaitableAdapter adapter) {
        Objects.requireNonNull(adapter);
        adapters.add(0, adapter);
        awaitableCache.set(buildAwaitableCache());
    }

    /**
     * Removes a previously registered adapter.
     */
    public static void unregister(AwaitableAdapter adapter) {
        adapters.remove(adapter);
        awaitableCache.set(buildAwaitableCache());
    }

    /**
     * Converts the given source to an {@link Awaitable}.
     */
    @SuppressWarnings("unchecked")
    static <T> Awaitable<T> toAwaitable(Object source) {
        if (source == null) {
            return Awaitable.of(null);
        }
        if (source instanceof Awaitable) return (Awaitable<T>) source;
        Class<?> type = source.getClass();
        AwaitableAdapter adapter = adapterFor(type);
        if (adapter != null) {
            return adapter.toAwaitable(source);
        }
        throw new IllegalArgumentException(
                "No Awaitable adapter found for type: " + type.getName()
                        + ". Register an AwaitableAdapter via ServiceLoader or AwaitableAdapterRegistry.register().");
    }

    /**
     * The adapter for the given type, or {@code null} when none supports it.
     * <p>
     * A cleared cache reference is removed and recomputed once; should the fresh
     * reference already be cleared as well, the answer comes from an uncached scan,
     * so the lookup terminates under any memory pressure.
     */
    private static AwaitableAdapter adapterFor(Class<?> type) {
        ClassValue<SoftReference<AwaitableAdapter>> cache = awaitableCache.get();
        AwaitableAdapter adapter = cache.get(type).get();
        if (adapter == null) {
            cache.remove(type);
            adapter = cache.get(type).get();
            if (adapter == null) {
                adapter = resolveAdapter(type);
            }
        }
        return adapter == NO_ADAPTER ? null : adapter;
    }

    /**
     * Resolves the first adapter supporting the supplied type, or {@link #NO_ADAPTER}.
     */
    private static AwaitableAdapter resolveAdapter(Class<?> type) {
        for (AwaitableAdapter adapter : adapters) {
            if (adapter.supportsAwaitable(type)) {
                return adapter;
            }
        }
        return NO_ADAPTER;
    }

    /**
     * Converts the given source to an {@link Iterable} for {@code for await}.
     */
    @SuppressWarnings("unchecked")
    public static <T> Iterable<T> toIterable(Object source) {
        if (source == null) {
            throw new IllegalArgumentException("Cannot convert null to Iterable");
        }
        Class<?> type = source.getClass();
        for (AwaitableAdapter adapter : adapters) {
            if (adapter.supportsIterable(type)) {
                return adapter.toIterable(source);
            }
        }
        throw new IllegalArgumentException(
                "No Iterable adapter found for type: " + type.getName()
                        + ". Register an AwaitableAdapter via ServiceLoader or AwaitableAdapterRegistry.register().");
    }

    // Keep this method static and the ClassValue below free of any enclosing-instance reference:
    // capturing one would let the association hold this registry, its class and its loader, undoing
    // the loader-unloading this class exists to allow (GROOVY-12280).
    private static ClassValue<SoftReference<AwaitableAdapter>> buildAwaitableCache() {
        return new ClassValue<>() {
            @Override
            protected SoftReference<AwaitableAdapter> computeValue(Class<?> type) {
                return new SoftReference<>(resolveAdapter(type));
            }
        };
    }

    /**
     * Built-in adapter for JDK Future types.
     */
    private static final class BuiltInAdapter implements AwaitableAdapter {

        /** {@inheritDoc} */
        @Override
        public boolean supportsAwaitable(Class<?> type) {
            return CompletionStage.class.isAssignableFrom(type)
                    || Future.class.isAssignableFrom(type);
        }

        /** {@inheritDoc} */
        @Override
        @SuppressWarnings("unchecked")
        public <T> Awaitable<T> toAwaitable(Object source) {
            if (source instanceof CompletableFuture) {
                return GroovyPromise.of((CompletableFuture<T>) source);
            }
            if (source instanceof CompletionStage) {
                return GroovyPromise.of(((CompletionStage<T>) source).toCompletableFuture());
            }
            if (source instanceof Future) {
                CompletableFuture<T> cf = new CompletableFuture<>();
                Future<T> future = (Future<T>) source;
                // Wrap blocking Future in a CF (submitted to default executor)
                CompletableFuture.runAsync(() -> {
                    try {
                        cf.complete(future.get());
                    } catch (ExecutionException e) {
                        cf.completeExceptionally(e.getCause());
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        cf.completeExceptionally(e);
                    } catch (Throwable e) {
                        cf.completeExceptionally(e);
                    }
                }, AsyncSupport.getExecutor());
                return GroovyPromise.of(cf);
            }
            throw new IllegalArgumentException("Cannot convert to Awaitable: " + source.getClass());
        }
    }
}
