/*
 * 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 org.apache.sling.event.it;


import static org.ops4j.pax.exam.CoreOptions.frameworkProperty;
import static org.ops4j.pax.exam.CoreOptions.junitBundles;
import static org.ops4j.pax.exam.CoreOptions.mavenBundle;
import static org.ops4j.pax.exam.CoreOptions.options;
import static org.ops4j.pax.exam.CoreOptions.systemProperty;
import static org.ops4j.pax.exam.CoreOptions.when;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Dictionary;
import java.util.Hashtable;
import java.util.List;

import javax.inject.Inject;

import org.apache.sling.api.resource.LoginException;
import org.apache.sling.api.resource.PersistenceException;
import org.apache.sling.api.resource.Resource;
import org.apache.sling.api.resource.ResourceResolver;
import org.apache.sling.api.resource.ResourceResolverFactory;
import org.apache.sling.discovery.PropertyProvider;
import org.apache.sling.event.impl.jobs.config.JobManagerConfiguration;
import org.apache.sling.event.jobs.JobManager;
import org.apache.sling.event.jobs.consumer.JobConsumer;
import org.apache.sling.event.jobs.consumer.JobExecutor;
import org.ops4j.pax.exam.Configuration;
import org.ops4j.pax.exam.CoreOptions;
import org.ops4j.pax.exam.Option;
import org.ops4j.pax.exam.cm.ConfigurationAdminOptions;
import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;
import org.osgi.framework.BundleException;
import org.osgi.framework.InvalidSyntaxException;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.cm.ConfigurationAdmin;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.event.EventConstants;
import org.osgi.service.event.EventHandler;
import org.slf4j.LoggerFactory;

public abstract class AbstractJobHandlingTest {

    private static final String BUNDLE_JAR_SYS_PROP = "project.bundle.file";

    /** The property containing the build directory. */
    private static final String SYS_PROP_BUILD_DIR = "bundle.build.dir";

    private static final String DEFAULT_BUILD_DIR = "target";

    private static final String PORT_CONFIG = "org.osgi.service.http.port";

    protected static final int DEFAULT_TEST_TIMEOUT = 1000*60*5;

    @Inject
    protected EventAdmin eventAdmin;

    @Inject
    protected ConfigurationAdmin configAdmin;

    @Inject
    protected BundleContext bc;

    protected List<ServiceRegistration<?>> registrations = new ArrayList<>();

    @Configuration
    public Option[] config() {
        final String buildDir = System.getProperty(SYS_PROP_BUILD_DIR, DEFAULT_BUILD_DIR);
        final String bundleFileName = System.getProperty( BUNDLE_JAR_SYS_PROP );
        final File bundleFile = new File( bundleFileName );
        if ( !bundleFile.canRead() ) {
            throw new IllegalArgumentException( "Cannot read from bundle file " + bundleFileName + " specified in the "
                + BUNDLE_JAR_SYS_PROP + " system property" );
        }

        String localRepo = System.getProperty("maven.repo.local", "");

        final String jackrabbitVersion = "2.13.1";
        final String oakVersion = "1.5.7";

        final String slingHome = new File(buildDir + File.separatorChar + "sling_" + System.currentTimeMillis()).getAbsolutePath();

        return options(
                frameworkProperty("sling.home").value(slingHome),
                frameworkProperty("repository.home").value(slingHome + File.separatorChar + "repository"),
                when( localRepo.length() > 0 ).useOptions(
                        systemProperty("org.ops4j.pax.url.mvn.localRepository").value(localRepo)
                ),
                when( System.getProperty(PORT_CONFIG) != null ).useOptions(
                        systemProperty(PORT_CONFIG).value(System.getProperty(PORT_CONFIG))),
                systemProperty("pax.exam.osgi.unresolved.fail").value("true"),

                ConfigurationAdminOptions.newConfiguration("org.apache.felix.jaas.ConfigurationSpi")
                    .create(true)
                    .put("jaas.defaultRealmName", "jackrabbit.oak")
                    .put("jaas.configProviderName", "FelixJaasProvider")
                    .asOption(),
                ConfigurationAdminOptions.factoryConfiguration("org.apache.felix.jaas.Configuration.factory")
                    .create(true)
                    .put("jaas.controlFlag", "optional")
                    .put("jaas.classname", "org.apache.jackrabbit.oak.spi.security.authentication.GuestLoginModule")
                    .put("jaas.ranking", 300)
                    .asOption(),
                ConfigurationAdminOptions.factoryConfiguration("org.apache.felix.jaas.Configuration.factory")
                    .create(true)
                    .put("jaas.controlFlag", "required")
                    .put("jaas.classname", "org.apache.jackrabbit.oak.security.authentication.user.LoginModuleImpl")
                    .asOption(),
                ConfigurationAdminOptions.factoryConfiguration("org.apache.felix.jaas.Configuration.factory")
                    .create(true)
                    .put("jaas.controlFlag", "sufficient")
                    .put("jaas.classname", "org.apache.jackrabbit.oak.security.authentication.token.TokenLoginModule")
                    .put("jaas.ranking", 200)
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.jackrabbit.oak.security.authentication.AuthenticationConfigurationImpl")
                    .create(true)
                    .put("org.apache.jackrabbit.oak.authentication.configSpiName", "FelixJaasProvider")
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.jackrabbit.oak.security.user.UserConfigurationImpl")
                    .create(true)
                    .put("groupsPath", "/home/groups")
                    .put("usersPath", "/home/users")
                    .put("defaultPath", "1")
                    .put("importBehavior", "besteffort")
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.jackrabbit.oak.security.user.RandomAuthorizableNodeName")
                    .create(true)
                    .put("enabledActions", new String[] {"org.apache.jackrabbit.oak.spi.security.user.action.AccessControlAction"})
                    .put("userPrivilegeNames", new String[] {"jcr:all"})
                    .put("groupPrivilegeNames", new String[] {"jcr:read"})
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.jackrabbit.oak.spi.security.user.action.DefaultAuthorizableActionProvider")
                    .create(true)
                    .put("length", 21)
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.jackrabbit.oak.plugins.segment.SegmentNodeStoreService")
                    .create(true)
                    .put("name", "Default NodeStore")
                    .asOption(),

                ConfigurationAdminOptions.factoryConfiguration("org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended")
                    .create(true)
                    .put("user.mapping", "org.apache.sling.event=admin")
                    .asOption(),
                ConfigurationAdminOptions.newConfiguration("org.apache.sling.jcr.resource.internal.JcrSystemUserValidator")
                    .create(true)
                    .put("allow.only.system.user", "false")
                    .asOption(),

                    // logging
                systemProperty("pax.exam.logging").value("none"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.log", "4.0.6"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.logservice", "1.0.6"),
                mavenBundle("org.slf4j", "slf4j-api", "1.7.13"),
                mavenBundle("org.slf4j", "jcl-over-slf4j", "1.7.13"),
                mavenBundle("org.slf4j", "log4j-over-slf4j", "1.7.13"),

                mavenBundle("commons-io", "commons-io", "2.4"),
                mavenBundle("commons-fileupload", "commons-fileupload", "1.3.1"),
                mavenBundle("commons-collections", "commons-collections", "3.2.2"),
                mavenBundle("commons-codec", "commons-codec", "1.10"),
                mavenBundle("commons-lang", "commons-lang", "2.6"),
                mavenBundle("org.apache.commons", "commons-lang3", "3.5"),
                mavenBundle("commons-pool", "commons-pool", "1.6"),

                mavenBundle("io.dropwizard.metrics", "metrics-core", "3.2.4"),

                mavenBundle("org.apache.servicemix.bundles", "org.apache.servicemix.bundles.concurrent", "1.3.4_1"),

                mavenBundle("org.apache.geronimo.bundles", "commons-httpclient", "3.1_1"),
                mavenBundle("org.apache.tika", "tika-core", "1.9"),
                mavenBundle("org.apache.tika", "tika-bundle", "1.9"),

                // infrastructure
                mavenBundle("org.apache.felix", "org.apache.felix.http.servlet-api", "1.1.2"),
                mavenBundle("org.apache.felix", "org.apache.felix.http.jetty", "3.1.6"),
                mavenBundle("org.apache.felix", "org.apache.felix.eventadmin", "1.4.8"),
                mavenBundle("org.apache.felix", "org.apache.felix.scr", "2.0.6"),
                mavenBundle("org.apache.felix", "org.apache.felix.configadmin", "1.8.10"),
                mavenBundle("org.apache.felix", "org.apache.felix.inventory", "1.0.4"),
                mavenBundle("org.apache.felix", "org.apache.felix.metatype", "1.1.2"),

                // sling
                mavenBundle("org.apache.sling", "org.apache.sling.settings", "1.3.8"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.osgi", "2.3.0"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.mime", "2.1.8"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.classloader", "1.3.2"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.johnzon", "1.0.0"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.scheduler", "2.4.14"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.threads", "3.2.4"),
                mavenBundle("org.apache.sling", "org.apache.sling.commons.metrics", "1.2.6"),

                mavenBundle("org.apache.sling", "org.apache.sling.auth.core", "1.3.12"),
                mavenBundle("org.apache.sling", "org.apache.sling.discovery.api", "1.0.2"),
                mavenBundle("org.apache.sling", "org.apache.sling.discovery.commons", "1.0.20"),
                mavenBundle("org.apache.sling", "org.apache.sling.discovery.standalone", "1.0.2"),

                mavenBundle("org.apache.sling", "org.apache.sling.api", "2.14.2"),
                mavenBundle("org.apache.sling", "org.apache.sling.resourceresolver", "1.4.18"),
                mavenBundle("org.apache.sling", "org.apache.sling.adapter", "2.1.10"),
                mavenBundle("org.apache.sling", "org.apache.sling.jcr.resource", "2.8.0"),
                mavenBundle("org.apache.sling", "org.apache.sling.jcr.classloader", "3.2.2"),
                mavenBundle("org.apache.sling", "org.apache.sling.jcr.contentloader", "2.2.4"),
                mavenBundle("org.apache.sling", "org.apache.sling.engine", "2.6.2"),
                mavenBundle("org.apache.sling", "org.apache.sling.serviceusermapper", "1.3.2"),

                mavenBundle("org.apache.sling", "org.apache.sling.jcr.jcr-wrapper", "2.0.0"),
                mavenBundle("org.apache.sling", "org.apache.sling.jcr.api", "2.4.0"),
                mavenBundle("org.apache.sling", "org.apache.sling.jcr.base", "2.4.0"),

                mavenBundle("com.google.guava", "guava", "15.0"),
                mavenBundle("org.apache.jackrabbit", "jackrabbit-api", jackrabbitVersion),
                mavenBundle("org.apache.jackrabbit", "jackrabbit-jcr-commons", jackrabbitVersion),
                mavenBundle("org.apache.jackrabbit", "jackrabbit-spi", jackrabbitVersion),
                mavenBundle("org.apache.jackrabbit", "jackrabbit-spi-commons", jackrabbitVersion),
                mavenBundle("org.apache.jackrabbit", "jackrabbit-jcr-rmi", jackrabbitVersion),

                mavenBundle("org.apache.felix", "org.apache.felix.jaas", "0.0.4"),

                mavenBundle("org.apache.jackrabbit", "oak-core", oakVersion),
                mavenBundle("org.apache.jackrabbit", "oak-commons", oakVersion),
                mavenBundle("org.apache.jackrabbit", "oak-lucene", oakVersion),
                mavenBundle("org.apache.jackrabbit", "oak-blob", oakVersion),
                mavenBundle("org.apache.jackrabbit", "oak-jcr", oakVersion),

                mavenBundle("org.apache.jackrabbit", "oak-segment", oakVersion),

                mavenBundle("org.apache.sling", "org.apache.sling.jcr.oak.server", "1.1.0"),

                mavenBundle("org.apache.sling", "org.apache.sling.testing.tools", "1.0.16"),
                mavenBundle("org.apache.httpcomponents", "httpcore-osgi", "4.1.2"),
                mavenBundle("org.apache.httpcomponents", "httpclient-osgi", "4.1.2"),


                // SLING-5560: delaying start of the sling.event bundle to
                // ensure the parameter 'startup.delay' is properly set to 1sec
                // for these ITs - as otherwise, the default of 30sec applies -
                // which will cause the tests to fail
                // @see setup() where the bundle is finally started - after reconfig
                CoreOptions.bundle( bundleFile.toURI().toString() ).start(false),

                junitBundles()
           );
    }

    protected JobManager getJobManager() {
        JobManager result = null;
        int count = 0;
        do {
            final ServiceReference<JobManager> sr = this.bc.getServiceReference(JobManager.class);
            if ( sr != null ) {
                result = this.bc.getService(sr);
            } else {
                count++;
                if ( count == 10 ) {
                    break;
                }
                sleep(500);
            }

        } while ( result == null );
        return result;
    }

    protected void sleep(final long time) {
        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // ignore
        }
    }

    public void setup() throws IOException {
        // set load delay to 3 sec
        final org.osgi.service.cm.Configuration c2 = this.configAdmin.getConfiguration("org.apache.sling.event.impl.jobs.jcr.PersistenceHandler", null);
        Dictionary<String, Object> p2 = new Hashtable<>();
        p2.put(JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY, 3L);
        // and startup.delay to 1sec - otherwise default of 30sec breaks tests!
        p2.put("startup.delay", 1L);
        c2.update(p2);

        // SLING-5560 : since the above (re)config is now applied, we're safe
        // to go ahead and start the sling.event bundle.
        // this time, the JobManagerConfiguration will be activated
        // with the 'startup.delay' set to 1sec - so that ITs actually succeed
        try {
            Bundle[] bundles = bc.getBundles();
            for (Bundle bundle : bundles) {
                if (bundle.getSymbolicName().contains("sling.event")) {
                    // assuming we only have 1 bundle that contains 'sling.event'
                    LoggerFactory.getLogger(getClass()).info("starting bundle... "+bundle);
                    bundle.start();
                    break;
                }
            }
        } catch (BundleException e) {
            LoggerFactory.getLogger(getClass()).error("could not start sling.event bundle: "+e, e);
            throw new RuntimeException(e);
        }
    }

    private int deleteCount;

    private void delete(final Resource rsrc )
    throws PersistenceException {
        final ResourceResolver resolver = rsrc.getResourceResolver();
        for(final Resource child : rsrc.getChildren()) {
            delete(child);
        }
        resolver.delete(rsrc);
        deleteCount++;
        if ( deleteCount >= 20 ) {
            resolver.commit();
            deleteCount = 0;
        }
    }

    public void cleanup() {
        // clean job area
        final ServiceReference<ResourceResolverFactory> ref = this.bc.getServiceReference(ResourceResolverFactory.class);
        final ResourceResolverFactory factory = this.bc.getService(ref);
        ResourceResolver resolver = null;
        try {
            resolver = factory.getAdministrativeResourceResolver(null);
            final Resource rsrc = resolver.getResource("/var/eventing");
            if ( rsrc != null ) {
                delete(rsrc);
                resolver.commit();
            }
        } catch ( final LoginException le ) {
            // ignore
        } catch (final PersistenceException e) {
            // ignore
        } catch ( final Exception e ) {
            // sometimes an NPE is thrown from the repository, as we
            // are in the cleanup, we can ignore this
        } finally {
            if ( resolver != null ) {
                resolver.close();
            }
        }
        // unregister all services
        for(final ServiceRegistration<?> reg : this.registrations) {
            reg.unregister();
        }
        this.registrations.clear();

        // remove all configurations
        try {
            final org.osgi.service.cm.Configuration[] cfgs = this.configAdmin.listConfigurations(null);
            if ( cfgs != null ) {
                for(final org.osgi.service.cm.Configuration c : cfgs) {
                    try {
                        c.delete();
                    } catch (final IOException io) {
                        // ignore
                    }
                }
            }
        } catch (final IOException io) {
            // ignore
        } catch (final InvalidSyntaxException e) {
            // ignore
        }
        this.sleep(1000);
    }

    /**
     * Helper method to register an event handler
     */
    protected ServiceRegistration<EventHandler> registerEventHandler(final String topic,
            final EventHandler handler) {
        final Dictionary<String, Object> props = new Hashtable<>();
        props.put(EventConstants.EVENT_TOPIC, topic);
        final ServiceRegistration<EventHandler> reg = this.bc.registerService(EventHandler.class,
                handler, props);
        this.registrations.add(reg);
        return reg;
    }

    protected long getConsumerChangeCount() {
        long result = -1;
        try {
            final Collection<ServiceReference<PropertyProvider>> refs = this.bc.getServiceReferences(PropertyProvider.class, "(changeCount=*)");
            if ( !refs.isEmpty() ) {
                result = (Long)refs.iterator().next().getProperty("changeCount");
            }
        } catch ( final InvalidSyntaxException ignore ) {
            // ignore
        }
        return result;
    }

    protected void waitConsumerChangeCount(final long minimum) {
        do {
            final long cc = getConsumerChangeCount();
            if ( cc >= minimum ) {
                // we need to wait for the topology events (TODO)
                sleep(200);
                return;
            }
            sleep(50);
        } while ( true );
    }

    /**
     * Helper method to register a job consumer
     */
    protected ServiceRegistration<JobConsumer> registerJobConsumer(final String topic,
            final JobConsumer handler) {
        long cc = this.getConsumerChangeCount();
        final Dictionary<String, Object> props = new Hashtable<>();
        props.put(JobConsumer.PROPERTY_TOPICS, topic);
        final ServiceRegistration<JobConsumer> reg = this.bc.registerService(JobConsumer.class,
                handler, props);
        this.registrations.add(reg);
        this.waitConsumerChangeCount(cc + 1);
        return reg;
    }

    /**
     * Helper method to register a job executor
     */
    protected ServiceRegistration<JobExecutor> registerJobExecutor(final String topic,
            final JobExecutor handler) {
        long cc = this.getConsumerChangeCount();
        final Dictionary<String, Object> props = new Hashtable<>();
        props.put(JobConsumer.PROPERTY_TOPICS, topic);
        final ServiceRegistration<JobExecutor> reg = this.bc.registerService(JobExecutor.class,
                handler, props);
        this.registrations.add(reg);
        this.waitConsumerChangeCount(cc + 1);
        return reg;
    }

    protected void unregister(final ServiceRegistration<?> reg) {
        if ( reg != null ) {
            this.registrations.remove(reg);
            reg.unregister();
        }
    }
}