/*
 * Copyright 2008-2026 Async-IO.org
 *
 * Licensed 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 org.atmosphere.cpr;

import org.atmosphere.cache.BroadcasterCacheInspector;
import org.atmosphere.cpr.BroadcastFilter.BroadcastAction;
import org.atmosphere.util.ExecutorsFactory;
import org.atmosphere.util.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Handle {@link Broadcaster} configuration like {@link ExecutorService} and {@link BroadcastFilter}.
 *
 * @author Jeanfrancois Arcand
 */
public class BroadcasterConfig {

    private static final Logger logger = LoggerFactory.getLogger(BroadcasterConfig.class);

    protected final ConcurrentLinkedQueue<BroadcastFilter> filters = new ConcurrentLinkedQueue<>();
    protected final ConcurrentLinkedQueue<PerRequestBroadcastFilter> perRequestFilters = new ConcurrentLinkedQueue<>();

    private final ReentrantLock executorLock = new ReentrantLock();
    private final ReentrantLock filterLock = new ReentrantLock();

    private ExecutorService executorService;
    private ExecutorService asyncWriteService;
    private ScheduledExecutorService scheduler;
    private BroadcasterCache broadcasterCache = BroadcasterCache.DEFAULT;
    private final AtmosphereConfig config;
    private boolean isExecutorShared;
    private boolean isAsyncExecutorShared;
    private final boolean shared;
    private String broadcasterId;
    private boolean handleExecutors;
    private List<String> filterList;

    /**
     * Create a new BroadcasterConfig. Remember to call init() after the object has been created.
     *
     */
    public BroadcasterConfig(List<String> broadcastFilters, AtmosphereConfig config, String broadcasterId) {
        this(broadcastFilters, config, true, broadcasterId);
    }

    /**
     * Create a new BroadcasterConfig. Remember to call init() after the object has been created.
     *
     */
    public BroadcasterConfig(List<String> broadcastFilters, AtmosphereConfig config, boolean handleExecutors, String broadcasterId) {
        this.config = config;
        this.broadcasterId = broadcasterId;
        this.shared = config.framework().isShareExecutorServices();
        this.handleExecutors = handleExecutors;
        this.filterList = broadcastFilters;
    }

    /**
     * Create a new BroadcasterConfig. Remember to call init() after the object has been created.
     *
     */
    public BroadcasterConfig(ExecutorService executorService, ExecutorService asyncWriteService,
                             ScheduledExecutorService scheduler, AtmosphereConfig config, String broadcasterId) {
        this.executorService = executorService;
        this.scheduler = scheduler;
        this.asyncWriteService = asyncWriteService;
        this.config = config;
        this.broadcasterId = broadcasterId;
        this.handleExecutors = true;
        this.shared = config.framework().isShareExecutorServices();
    }

    /**
     * Initialize BroadcastFilters and BroadcasterCache. Must always be called after creating a new BroadcasterConfig!
     */
    public BroadcasterConfig init() {
        if (handleExecutors) {
            configExecutors();
        }

        if (filterList != null) {
            configureBroadcasterFilter(filterList);
        }
        configureBroadcasterCache();
        return this;
    }

    @SuppressWarnings("unchecked")
    private void configureBroadcasterCache() {
        try {
            String className = config.framework().getBroadcasterCacheClassName();
            if (className != null) {
                broadcasterCache = config.framework().newClassInstance(BroadcasterCache.class,
                        (Class<BroadcasterCache>) IOUtils.loadClass(getClass(), className));
                configureSharedCacheExecutor();
                broadcasterCache.configure(config);
            }

            for (BroadcasterCacheInspector b : config.framework().inspectors()) {
                broadcasterCache.inspector(b);
            }

            for (BroadcasterCacheListener l : config.framework().broadcasterCacheListeners()) {
                broadcasterCache.addBroadcasterCacheListener(l);
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected void configureSharedCacheExecutor() {
        if (!shared) return;

        config.properties().put("shared", "true");
    }

    protected BroadcasterConfig broadcasterID(String broadcasterId) {
        this.broadcasterId = broadcasterId;
        initClusterExtension();
        return this;
    }

    protected void initClusterExtension() {
        for (BroadcastFilter mf : filters) {
            if (mf instanceof ClusterBroadcastFilter cbf) {
                try {
                    Broadcaster b = config.getBroadcasterFactory().lookup(broadcasterId, false);
                    if (b != null) {
                        filterLock.lock();
                        try {
                            cbf.setBroadcaster(b);
                        } finally {
                            filterLock.unlock();
                        }
                    }
                } catch (Throwable t) {
                    logger.error("", t);
                }
            }
        }
    }

    public boolean handleExecutors() {
        return handleExecutors;
    }

    protected void configExecutors() {
        executorLock.lock();
        try {
            if (shared) {
                handleExecutors = false;
                isExecutorShared = true;
                isAsyncExecutorShared = true;
            }

            executorService = ExecutorsFactory.getMessageDispatcher(config, broadcasterId);
            asyncWriteService = ExecutorsFactory.getAsyncOperationExecutor(config, broadcasterId);
            scheduler = ExecutorsFactory.getScheduler(config);
        } finally {
            executorLock.unlock();
        }
    }

    /**
     * Set an {@link ExecutorService} which can be used to dispatch
     * {@link AtmosphereResourceEvent}. By default, the result of {@link ExecutorsFactory#getMessageDispatcher(AtmosphereConfig, String)}
     * is used if this method is not invoked.
     *
     * @param executorService to be used when broadcasting.
     */
    public BroadcasterConfig setExecutorService(ExecutorService executorService) {
        return setExecutorService(executorService, false);
    }

    /**
     * Set an {@link ExecutorService} which can be used to dispatch
     * {@link AtmosphereResourceEvent}. By default, the result of {@link ExecutorsFactory#getMessageDispatcher(AtmosphereConfig, String)}
     * is used if this method is not invoked.
     *
     * @param executorService  to be used when broadcasting.
     * @param isExecutorShared true if the life cycle of the {@link ExecutorService} will be executed by the application.
     *                         It means Atmosphere will NOT invoke the shutdown method when {@link org.atmosphere.cpr.BroadcasterConfig#destroy()}
     *                         is invoked.
     */
    public BroadcasterConfig setExecutorService(ExecutorService executorService, boolean isExecutorShared) {
        if (!this.isExecutorShared && this.executorService != null) {
            this.executorService.shutdown();
        }
        this.executorService = executorService;
        this.isExecutorShared = isExecutorShared;
        return this;
    }

    /**
     * Return the {@link ExecutorService} this {@link Broadcaster} supports.
     * By default it returns the result of {@link ExecutorsFactory#getMessageDispatcher(AtmosphereConfig, String)}}.
     *
     * @return An ExecutorService.
     */
    public ExecutorService getExecutorService() {
        return executorService;
    }

    /**
     * Set an {@link ExecutorService} which can be used to write
     * {@link org.atmosphere.cpr.AtmosphereResourceEvent#getMessage()}. By default, the result of {@link ExecutorsFactory#getAsyncOperationExecutor(AtmosphereConfig, String)}
     * is used if this method is not invoked.
     *
     * @param asyncWriteService to be used when writing events.
     */
    public BroadcasterConfig setAsyncWriteService(ExecutorService asyncWriteService) {
        return setAsyncWriteService(asyncWriteService, false);
    }

    /**
     * Set an {@link ExecutorService} which can be used to write
     * {@link org.atmosphere.cpr.AtmosphereResourceEvent#getMessage()}. By default, the result of {@link ExecutorsFactory#getAsyncOperationExecutor(AtmosphereConfig, String)}
     * is used if this method is not invoked.
     *
     * @param asyncWriteService     to be used when writing events.
     * @param isAsyncExecutorShared true if the life cycle of the {@link ExecutorService} will be executed by the application.
     *                              It means Atmosphere will NOT invoke the shutdown method when this {@link org.atmosphere.cpr.BroadcasterConfig#destroy()}
     *                              is invoked.
     */
    public BroadcasterConfig setAsyncWriteService(ExecutorService asyncWriteService, boolean isAsyncExecutorShared) {
        if (!this.isAsyncExecutorShared && this.asyncWriteService != null) {
            this.asyncWriteService.shutdown();
        }
        this.asyncWriteService = asyncWriteService;
        this.isAsyncExecutorShared = isAsyncExecutorShared;
        return this;
    }

    /**
     * Return the {@link ExecutorService} this {@link Broadcaster} use for executing asynchronous write of events.
     * By default it returns the result of {@link ExecutorsFactory#getAsyncOperationExecutor(AtmosphereConfig, String)}.
     *
     * @return An ExecutorService.
     */
    public ExecutorService getAsyncWriteService() {
        return asyncWriteService;
    }

    /**
     * Add a {@link BroadcastFilter}.
     *
     * @param e {@link BroadcastFilter}
     * @return true if successfully added
     */
    public boolean addFilter(BroadcastFilter e) {
        return addFilter(e, true);
    }

    /**
     * Add a {@link BroadcastFilter}.
     *
     * @param e {@link BroadcastFilter}
     * @return true if successfully added
     */
    protected boolean addFilter(BroadcastFilter e, boolean init) {
        logDuplicateFilter(e);
        if (filters.contains(e)) return false;

        if (e instanceof BroadcastFilterLifecycle lifecycle) {
            lifecycle.init(config);
        }

        if (init && e instanceof ClusterBroadcastFilter cbf) {
            Broadcaster b = config.getBroadcasterFactory().lookup(broadcasterId, false);
            if (b != null) {
                filterLock.lock();
                try {
                    cbf.setBroadcaster(b);
                } finally {
                    filterLock.unlock();
                }
            }
        }

        if (e instanceof PerRequestBroadcastFilter perRequest) {
            perRequestFilters.add(perRequest);
        }

        return filters.offer(e);
    }

    private void logDuplicateFilter(BroadcastFilter e) {
        for (BroadcastFilter f : filters) {
            if (f.getClass().isInstance(e)) {
                logger.trace("Duplicate Filter instance {}", f.getClass());
            }
        }
    }

    /**
     * Return the current list of installed {@link BroadcastFilter}s.
     *
     * @return the current list of installed {@link BroadcastFilter}s
     */
    public Collection<BroadcastFilter> filters() {
        return filters;
    }

    public void destroy() {
        destroy(false);
    }

    protected void destroy(boolean force) {

        broadcasterCache.cleanup();
        if ((force || !shared) && broadcasterCache != null) {
            broadcasterCache.stop();
        }

        removeAllFilters();

        if (!force && !handleExecutors) return;

        if ((force || !isExecutorShared) && executorService != null) {
            executorService.shutdownNow();
            awaitQuietly(executorService, "executorService");
        }
        if ((force || !isAsyncExecutorShared) && asyncWriteService != null) {
            asyncWriteService.shutdownNow();
            awaitQuietly(asyncWriteService, "asyncWriteService");
        }

        if ((force || !shared) && scheduler != null) {
            scheduler.shutdownNow();
            awaitQuietly(scheduler, "scheduler");
        }
    }

    private void awaitQuietly(ExecutorService service, String name) {
        try {
            if (!service.awaitTermination(5, TimeUnit.SECONDS)) {
                logger.warn("{} did not terminate within 5 seconds", name);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * Force shutdown of all {@link ExecutorService}s.
     */
    public void forceDestroy() {
        destroy(true);
    }

    /**
     * Remove a {@link BroadcastFilter}.
     *
     * @param filter {@link BroadcastFilter}
     * @return true if successfully removed
     */
    public boolean removeFilter(BroadcastFilter filter) {
        if (filter instanceof BroadcastFilterLifecycle lifecycle) {
            lifecycle.destroy();
        }

        if (filter instanceof PerRequestBroadcastFilter prf) {
            perRequestFilters.remove(prf);
        }

        return filters.remove(filter);
    }

    /**
     * Remove all {@link BroadcastFilter}s.
     */
    public void removeAllFilters() {
        for (BroadcastFilter filter : filters) {
            removeFilter(filter);
        }
    }

    /**
     * Check if this object contains {@link BroadcastFilter}s.
     *
     * @return true if this object contains {@link BroadcastFilter}s
     */
    public boolean hasFilters() {
        return !filters.isEmpty();
    }

    /**
     * Check if this object contains {@link BroadcastFilter}s.
     *
     * @return true if this object contains {@link BroadcastFilter}s
     */
    public boolean hasPerRequestFilters() {
        if (filters.isEmpty()) {
            return false;
        } else {
            for (BroadcastFilter b : filters) {
                if (b instanceof PerRequestBroadcastFilter) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Invoke {@link BroadcastFilter}s in the order they were added.
     *
     * @param object the broadcasted object.
     * @return BroadcastAction that tell Atmosphere to invoke the next filter or not.
     */
    protected BroadcastAction filter(Object object) {

        Object newO = unwrap(object);
        boolean isManipulated = false;
        if (newO != null && !newO.equals(object)) {
            isManipulated = true;
            object = newO;
        }

        var transformed = new BroadcastAction(object);
        for (BroadcastFilter mf : filters) {
            filterLock.lock();
            try {
                transformed = mf.filter(broadcasterId, object, transformed.message());
                if (transformed == null
                        || transformed.action() == BroadcastAction.ACTION.ABORT
                        || transformed.action() == BroadcastAction.ACTION.SKIP) {
                    return transformed;
                }
            } finally {
                filterLock.unlock();
            }
        }
        return wrap(transformed, isManipulated);
    }

    /**
     * Invoke {@link BroadcastFilter}s in the order they were added, with a unique {@link AtmosphereRequest}.
     *
     * @param r       {@link AtmosphereResource}
     * @param message the broadcasted object.
     * @param message the broadcasted object.
     * @return BroadcastAction that tell Atmosphere to invoke the next filter or not.
     */
    protected BroadcastAction filter(AtmosphereResource r, Object message, Object originalMessage) {

        Object newO = unwrap(message);
        boolean isManipulated = false;
        if (!newO.equals(message)) {
            isManipulated = true;
            message = newO;
        }

        var transformed = new BroadcastAction(message);
        for (PerRequestBroadcastFilter mf : perRequestFilters) {
            filterLock.lock();
            try {
                transformed = mf.filter(broadcasterId, r, originalMessage, transformed.message());
                if (transformed == null
                        || transformed.action() == BroadcastAction.ACTION.ABORT
                        || transformed.action() == BroadcastAction.ACTION.SKIP) {
                    return transformed;
                }
            } finally {
                filterLock.unlock();
            }
        }
        return wrap(transformed, isManipulated);
    }

    /**
     * Apply all filters to the {@link AtmosphereResource} and the provided {@link List} of messages.
     *
     * @param r             {@link AtmosphereResource}
     * @param cacheMessages list of messages
     * @return the new list of objects.
     */
    public List<Object> applyFilters(AtmosphereResource r, List<Object> cacheMessages) {
        var filteredMessage = new LinkedList<Object>();
        BroadcastFilter.BroadcastAction a;
        for (Object o : cacheMessages) {
            a = filter(o);
            if (a.action() == BroadcastFilter.BroadcastAction.ACTION.ABORT) return List.of();

            if (a.action() == BroadcastAction.ACTION.SKIP) {
                filteredMessage.add(a.message());
                return filteredMessage;
            }

            a = filter(r, a.message(), a.originalMessage());
            if (a.action() == BroadcastFilter.BroadcastAction.ACTION.ABORT) return List.of();

            if (a.action() == BroadcastAction.ACTION.SKIP) {
                filteredMessage.add(a.message());
                return filteredMessage;
            }

            filteredMessage.add(a.message());
        }
        return filteredMessage;
    }

    /**
     * Set an {@link ExecutorService} which can be used to dispatch
     * {@link AtmosphereResourceEvent}s. By default a {@link java.util.concurrent.ScheduledExecutorService}
     * is used if this method is not invoked.
     *
     * @param scheduler to be used when broadcasting.
     * @return this.
     */
    public BroadcasterConfig setScheduledExecutorService(ScheduledExecutorService scheduler) {
        if (this.scheduler != null) {
            this.scheduler.shutdown();
        }
        this.scheduler = scheduler;
        return this;
    }

    /**
     * Return the {@link ScheduledExecutorService} this {@link Broadcaster} supports.
     * By default it returns an {@link Executors#newScheduledThreadPool} and will use
     * the underlying number of core/protocol as an indication of the thread number.
     *
     * @return An ExecutorService.
     */
    public ScheduledExecutorService getScheduledExecutorService() {
        return scheduler;
    }

    /**
     * Set a {@link BroadcasterCache}.
     *
     * @param broadcasterCache a {@link BroadcasterCache}
     * @return this
     */
    public BroadcasterConfig setBroadcasterCache(BroadcasterCache broadcasterCache) {
        this.broadcasterCache = broadcasterCache;
        return this;
    }

    /**
     * Get the {@link BroadcasterCache} used for this {@link Broadcaster}.
     *
     * @return a {@link BroadcasterCache}
     */
    public BroadcasterCache getBroadcasterCache() {
        return broadcasterCache;
    }

    @SuppressWarnings("unchecked")
    void configureBroadcasterFilter(List<String> list) {
        for (String broadcastFilter : list) {
            BroadcastFilter bf = null;
            try {
                bf = config.framework().newClassInstance(BroadcastFilter.class, (Class<BroadcastFilter>) IOUtils.loadClass(getClass(), broadcastFilter));
            } catch (Exception e) {
                logger.warn("Error trying to instantiate BroadcastFilter: {}", broadcastFilter, e);
            }
            if (bf != null) {
                addFilter(bf);
            }
        }
    }

    protected Object unwrap(Object o) {
        Object manipulated = o;
        for (FilterManipulator f: config.framework().filterManipulators()) {
            manipulated = f.unwrap(manipulated);
        }
        return manipulated;
    }

    protected BroadcastAction wrap(BroadcastAction a, boolean wasUnwraped) {
        for (FilterManipulator f: config.framework().filterManipulators()) {
            a = f.wrap(a, wasUnwraped);
        }
        return a;
    }

    /**
     * Return the {@link AtmosphereConfig} value. This value might be null
     * if the associated {@link Broadcaster} has been created manually.
     *
     * @return {@link AtmosphereConfig}
     */
    public AtmosphereConfig getAtmosphereConfig() {
        return config;
    }

    /**
     * Manipulate the message before and after they are getting filtered by {@link org.atmosphere.cpr.BroadcastFilter}
     */
    public interface FilterManipulator {

        Object unwrap(Object o);

        BroadcastAction wrap(BroadcastAction a, boolean wasWrapped);
    }
}
