package be.iminds.ilabt.jfed.highlevel.jobs;

import be.iminds.ilabt.jfed.call_log_output.HtmlLogOutput;
import be.iminds.ilabt.jfed.experiment.Experiment;
import be.iminds.ilabt.jfed.experiment.ExperimentPart;
import be.iminds.ilabt.jfed.experiment.SfaExperimentPart;
import be.iminds.ilabt.jfed.experiment.tasks.ExperimentTaskStatus;
import be.iminds.ilabt.jfed.fedmon.webapi.service.json.Server;
import be.iminds.ilabt.jfed.highlevel.controller.TaskThread;
import be.iminds.ilabt.jfed.highlevel.jobs.parts.ExperimentPartController;
import be.iminds.ilabt.jfed.highlevel.jobs.parts.ExperimentPartControllerManager;
import be.iminds.ilabt.jfed.highlevel.jobs.parts.NonSfaExperimentPartController;
import be.iminds.ilabt.jfed.highlevel.jobs.report.JobReportWrappingLogger;
import be.iminds.ilabt.jfed.highlevel.jobs.report.StitchingJobReport;
import be.iminds.ilabt.jfed.highlevel.jobs.states.AllocateSliversOnAuthorityState;
import be.iminds.ilabt.jfed.highlevel.jobs.states.CreateUserSpecListState;
import be.iminds.ilabt.jfed.highlevel.jobs.states.JobStateFactory;
import be.iminds.ilabt.jfed.highlevel.jobs.states.StopExperimentAtAuthorityPartState;
import be.iminds.ilabt.jfed.highlevel.tasks.GetAdvertisementTask;
import be.iminds.ilabt.jfed.highlevel.tasks.HighLevelTaskFactory;
import be.iminds.ilabt.jfed.highlevel.tasks.SCSRequestTask;
import be.iminds.ilabt.jfed.highlevel.util.AggregateManagerWrapperFactory;
import be.iminds.ilabt.jfed.lowlevel.JFedException;
import be.iminds.ilabt.jfed.lowlevel.TestbedInfoSource;
import be.iminds.ilabt.jfed.lowlevel.api.AbstractGeniAggregateManager;
import be.iminds.ilabt.jfed.lowlevel.api.StitchingComputationService;
import be.iminds.ilabt.jfed.lowlevel.api_wrapper.AggregateManagerWrapper;
import be.iminds.ilabt.jfed.lowlevel.authority.AuthorityFinder;
import be.iminds.ilabt.jfed.lowlevel.stitching.StitchingDirector;
import be.iminds.ilabt.jfed.rspec.model.ModelRspec;
import be.iminds.ilabt.jfed.rspec.model.ModelRspecType;
import be.iminds.ilabt.jfed.rspec.model.StringRspec;
import be.iminds.ilabt.jfed.rspec.rspec_source.RequestRspecSource;
import be.iminds.ilabt.jfed.util.GeniUrn;
import be.iminds.ilabt.jfed.util.TargetAuthority;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

/**
 * User: twalcari
 * Date: 12/2/2015
 * Time: 16:43
 */
public class AllocateExperimentJob extends AbstractJob<Boolean> {
    private static final Logger LOG = LoggerFactory.getLogger(AllocateExperimentJob.class);

    private static final org.slf4j.Logger ACTUAL_LOG = LoggerFactory.getLogger(AllocateExperimentJob.class);
    private static final int TIME_BETWEEN_ALLOCATE_RETRIES_MS = 10_000;
    private static final int MAX_ALLOCATE_STITCH_ACTION_TRIES = 3;

    //this variable is typically set for debugging purposes
    private Integer waitForReadyDelay = null;

    private final Server scsAuthority;
    private final JobStateFactory jobStateFactory;

    private final TestbedInfoSource testbedInfoSource;
    private final AuthorityFinder authorityFinder;
    private final AggregateManagerWrapperFactory aggregateManagerWrapperFactory;
    private final ExperimentPartControllerManager epcManager;

    private final StitchingJobReport stitchingJobReport;
    private final JobReportWrappingLogger log;
    private final JobReportWrappingLogger stitchingDirectorLog;

    private final AllocationDirector allocationDirector;

    private Boolean success = null;
    private Collection<GeniUrn> failedResources = null; //TODO: register resources that could not be reserved here

    public AllocateExperimentJob(Experiment experiment, Server scsAuthority, HighLevelTaskFactory hltf,
                                 TaskThread tt, JobStateFactory jobStateFactory,
                                 TestbedInfoSource testbedInfoSource, AuthorityFinder authorityFinder,
                                 AggregateManagerWrapperFactory aggregateManagerWrapperFactory, ExperimentPartControllerManager epcManager) {
        super("Allocate Experiment " + experiment.getName(), experiment, hltf, tt);
        this.scsAuthority = scsAuthority;
        this.jobStateFactory = jobStateFactory;
        this.testbedInfoSource = testbedInfoSource;
        this.authorityFinder = authorityFinder;
        this.aggregateManagerWrapperFactory = aggregateManagerWrapperFactory;
        this.epcManager = epcManager;

        assert experiment.getRequestedEndTime() != null;


        //do some initial setup for advanced logging

        this.stitchingJobReport = new StitchingJobReport(this);
        this.log = new JobReportWrappingLogger(ACTUAL_LOG, stitchingJobReport);
        this.stitchingDirectorLog = new JobReportWrappingLogger(StitchingDirector.ACTUAL_LOG, stitchingJobReport);

        if (experiment.getNewRequestRspecSource().getStringRspec().isStitching(testbedInfoSource, authorityFinder)) {
            this.allocationDirector =
                    new StitchingAllocationDirector(experiment, testbedInfoSource, authorityFinder, stitchingDirectorLog);

        } else {
            this.allocationDirector = new BasicAllocationDirector(experiment);
        }
    }

    @Override
    public Boolean execute() throws Exception {
        assert experiment.getNewRequestRspecSource() != null;

        //-2. Do some preparations
        CreateUserSpecListState createUserSpecListState = jobStateFactory.createCreateUserSpecListState(experiment);
        setAndRunState(createUserSpecListState);

        experiment.updateUserSpecs(createUserSpecListState.getUserSpecs());

        log.debug("AllocateExperimentJob started");

        //make sure we have a valid experiment
        assert experiment.getSlice() != null;
        assert experiment.getRequestedEndTime() != null;

        if (allocationDirector instanceof StitchingAllocationDirector) {

            SCSRequestState scsRequestState = new SCSRequestState();
            setAndRunState(scsRequestState);

            if (scsRequestState.getStatus() != ExperimentTaskStatus.SUCCESS) {
                log.error("SCS call failed", scsRequestState.getTaskExecution().getException());
                success = false;
                return false; //we give up
            }

            StitchingComputationService.ComputePathResult result = scsRequestState.getTask().getResult();
            assert result != null;

            //Update the slice request rspec with the enriched request-rspec from the SCS
            experiment.getSlice().setRequestRspec(new RequestRspecSource(new StringRspec(result.getServiceRspec()), ModelRspecType.FX));

            ((StitchingAllocationDirector) allocationDirector).setComputePathResult(result);
        }

        setAndRunState(new PopulateExperimentPartsState());

        if (hasNonSfaParts()) {
            setAndRunState(new CreateNonSfaExperimentPartsContainersState());

            setAndRunState(new StartNonSfaPartsState());
        }

        if (!hasSfaParts())
            return true;


            final AllocateSfaPartsState allocateSfaPartsState = new AllocateSfaPartsState();
            setAndRunState(allocateSfaPartsState);

            if (allocateSfaPartsState.isPermanentFailure()) {
                log.info("AllocatePartsState reported permanent failure");

                success = false;
                return false;
            }  else {
                success = true;
                return true;
            }
    }

    public Boolean isSuccess() {
        return success;
    }

    public StitchingJobReport getStitchingJobReport() {
        return stitchingJobReport;
    }

    public Collection<GeniUrn> getFailedResources() {
        return failedResources;
    }

    public class SCSRequestState extends SingleCallState<SCSRequestTask> {
        //TODO: check mixed case of stitched and non-stitched authorities. Does SCS return non-stitched nodes/links in it's new RSpec?

        protected SCSRequestState() {
            super("Request stitched link setup-instructions",
                    hltf.createSCSRequestTask(experiment.getSlice(), experiment.getNewRequestRspecSource(),
                            experiment.getRequestedEndTime(), scsAuthority));
        }

        @Nonnull
        @Override
        public ExperimentTaskStatus executeState(Job<?> job) throws InterruptedException {
            super.executeState(job);

            return getTask().getResult() != null ? ExperimentTaskStatus.SUCCESS : ExperimentTaskStatus.FAILED;
        }
    }

    private class PopulateExperimentPartsState extends State {
        protected PopulateExperimentPartsState() {
            super("Processing experiment information");
        }

        @Nonnull
        @Override
        public ExperimentTaskStatus executeState(Job<?> job) throws InterruptedException, JFedException {
            Set<TargetAuthority> authorities = new HashSet<>();

            //create SfaExperimentParts
            if (allocationDirector instanceof StitchingAllocationDirector) {
                Collection<Server> auths = ((StitchingAllocationDirector) allocationDirector).getAllAuthorities();
                List<TargetAuthority> targetAuthorities = new ArrayList<>();
                //this is suboptimal, BUT it will work, as the central broker architecture (exogeni) requires
                // the central broker auth to be used for stitching
                for (Server auth : auths) {
                    if (auth.hasFlag(Server.Flag.centralBrokerArchitectureBrokerauth)) {
                        Server connectAuth = testbedInfoSource.getByUrn(
                                auth.getDefaultComponentManagerUrn(),
                                TestbedInfoSource.SubAuthMatchAllowed.ALLOW_TOPLEVEL,
                                TestbedInfoSource.SubAuthMatchPreference.PREFER_TOPLEVEL,
                                true/*ignoreUrnTypeAndName*/);
                        if (connectAuth == null) {
                            log.error("ERROR: Component Manager Urn for central broker architecture toplevel authority in Request RSpec unknown (connectAuth=null): " +
                                    "\"" + auth.getDefaultComponentManagerUrn() + "\".");
                            throw new UnknownAuthorityException(auth.getDefaultComponentManagerAsGeniUrn());
                        }
                        authorities.add(new TargetAuthority(auth.getDefaultComponentManagerAsGeniUrn(), auth, connectAuth));
                    } else {
                        authorities.add(new TargetAuthority(auth.getDefaultComponentManagerAsGeniUrn(), auth, auth));
                    }
                }
            } else {
                assert experiment.getSlice() != null;
                assert experiment.getSlice().getRequestRspec() != null;
                for (final GeniUrn componentManagerUrn : experiment.getSlice().getRequestRspec().getAllComponentManagerUrns()) {
                    final Server connectAuth =
                            authorityFinder.findByUrn(componentManagerUrn, AuthorityFinder.Purpose.CREATE_SLIVER);
                    final Server rspecAuth =
                            authorityFinder.findByUrn(componentManagerUrn, AuthorityFinder.Purpose.REQUEST_RSPEC);

                    if (connectAuth == null) {
                        log.error("ERROR: Component Manager Urn in Request RSpec unknown (connectAuth=null): \"" + componentManagerUrn + "\".");
                        throw new UnknownAuthorityException(componentManagerUrn);
                    }
                    if (rspecAuth == null) {
                        log.error("ERROR: Component Manager Urn in Request RSpec unknown (rspecAuth=null): \"" + componentManagerUrn + "\".");
                        throw new UnknownAuthorityException(componentManagerUrn);
                    }

                    if (connectAuth.isEdgeVlan())
                        continue;

                    authorities.add(new TargetAuthority(componentManagerUrn, rspecAuth, connectAuth));
                }
            }

            for (TargetAuthority auth : authorities) {
                experiment.addExperimentPart(new SfaExperimentPart(experiment, auth.getServerToConnect(), auth.getServerForRspecComponentManager()));
            }

            epcManager.populateNonSfaExperimentParts(experiment);


            return ExperimentTaskStatus.SUCCESS;
        }

    }

    public class AllocateSfaPartsState extends SlicedState<AllocatePartStateSlice> {
        private final List<AllocatePartStateSlice> slices = new ArrayList<>();
        private final Map<ExperimentPart, StateSliceThread<AllocatePartStateSlice>> threadMap = new HashMap<>();

        protected AllocateSfaPartsState() {
            super("Allocate resources with StitchingDirector");

            for (SfaExperimentPart experimentPart : getSfaParts()) {
                AllocatePartStateSlice allocatePartStateSlice = new AllocatePartStateSlice(experimentPart);
                slices.add(allocatePartStateSlice);
                StateSliceThread<AllocatePartStateSlice> thread =
                        new StateSliceThread<>(allocatePartStateSlice, getName());
                threadMap.put(experimentPart, thread);
            }

        }

        @Nonnull
        @Override
        protected ExperimentTaskStatus executeState(Job<?> job) throws InterruptedException, JFedException {
            threadMap.values().forEach(Thread::start);

            while (allocationDirector.areAnyActionsLeft()) {
                List<StitchingDirector.StitchAction> stitchActions = allocationDirector.getStitchActions();

                if (!stitchActions.isEmpty()) {
                    for (StitchingDirector.StitchAction stitchAction : stitchActions) {
                        ExperimentPart part = experiment.getPart(stitchAction.getServer());
                        threadMap.get(part).getStateSlice().submitStitchAction(new AllocateAction(stitchAction));
                    }
                } else {
                    log.trace("No stitchActions available for the moment, but director says it's not finished yet.");
                }

                Thread.sleep(250);
            }
            threadMap.values().forEach(apt -> apt.getStateSlice().submitStitchAction(AllocateAction.createFinishedAction()));

            return allocationDirector.isGaveUp() ? ExperimentTaskStatus.FAILED : ExperimentTaskStatus.SUCCESS;
        }

        public boolean isPermanentFailure() {
            return allocationDirector.isGaveUp();
        }

        public Map<ExperimentPart, StateSliceThread<AllocatePartStateSlice>> getThreadMap() {
            return threadMap;
        }

        public Collection<AllocatePartStateSlice> getSlices() {
            return slices;
        }
    }

    private static class AllocateAction {
        private final boolean finished;

        private final StitchingDirector.StitchAction stitchAction;

        public AllocateAction(StitchingDirector.StitchAction stitchAction) {
            this.stitchAction = stitchAction;
            this.finished = false;
        }

        private AllocateAction() {
            this.stitchAction = null;
            this.finished = true;
        }

        public static AllocateAction createFinishedAction() {
            return new AllocateAction();
        }

        public boolean isFinished() {
            return finished;
        }

        public StitchingDirector.StitchAction getStitchAction() {
            return stitchAction;
        }

        @Override
        public String toString() {
            return "AllocateAction{" +
                    "finished=" + finished +
                    ", stitchAction=" + stitchAction +
                    '}';
        }
    }

    public class AllocatePartStateSlice extends ExperimentPartStateSlice<SfaExperimentPart> {
        private final Logger log = AllocateExperimentJob.this.log;  //TODO: split into separate logger
        private final BlockingQueue<AllocateAction> queue = new LinkedBlockingQueue<>();

        private int allocateStitchActionRetryCount = 0;

        private AllocatePartStateSlice(SfaExperimentPart experimentPart) {
            super(AllocateExperimentJob.this, experimentPart);
        }

        @Override
        public ExperimentTaskStatus statefulRun() throws JFedException, InterruptedException {
            while (true) {
                WaitForNextAllocateActionState waitForNextAllocateActionState = new WaitForNextAllocateActionState();
                setAndRunState(waitForNextAllocateActionState);

                AllocateAction allocateAction = waitForNextAllocateActionState.getAllocateAction();

                if (allocateAction.isFinished())
                    return ExperimentTaskStatus.SUCCESS;

                StitchingDirector.StitchAction stitchAction = allocateAction.getStitchAction();

                switch (stitchAction.getType()) {
                    case ALLOCATE:
                        processAllocateStitchAction(stitchAction);
                        break;
                    case DELETE:
                        processDeleteStitchAction();
                        break;
                    case LISTRESOURCES:
                        processListResourcesStitchAction();

                        break;
                    default:
                        throw new IllegalArgumentException("Unexpected StitchAction: " + stitchAction);
                }
            }

        }

        private void processListResourcesStitchAction() throws InterruptedException {
            GetAdvertisementState getAdvertisementState = new GetAdvertisementState();
            try {
                setAndRunState(getAdvertisementState);
            } catch (JFedException e) {
                log.error("Error while performing ListResources on {}", experimentPart.getConnectSfaAuthority(), e);
            }
            allocationDirector.processListResourcesResult(experimentPart.getConnectSfaAuthority(),
                    getAdvertisementState.getTask().getAdvertisementRspecString());
        }

        private void processDeleteStitchAction() throws InterruptedException, JFedException {
            StopExperimentAtAuthorityPartState stopExperimentAtAuthorityPartState =
                    jobStateFactory.createStopExperimentAtAuthorityPartState(AllocateExperimentJob.this, experimentPart);
            try {
                setAndRunState(stopExperimentAtAuthorityPartState);
            } catch (JFedException e) {
                log.error("Error while deleting resources from experiment", e);
            } finally {
                allocationDirector.processDeleteResult(experimentPart.getConnectSfaAuthority());
            }
        }

        private void processAllocateStitchAction(StitchingDirector.StitchAction stitchAction) throws InterruptedException, JFedException {
            RequestRspecSource overridingRequestRspecSource = stitchAction.getRequestRspec() != null ?
                    new RequestRspecSource(new StringRspec(stitchAction.getRequestRspec())) : null;

            final AllocateSliversOnAuthorityState allocateSliversOnAuthorityState =
                    jobStateFactory.createAllocateSliversOnAuthorityState(AllocateExperimentJob.this, experimentPart, experiment.getUserSpecs(), overridingRequestRspecSource);


            AbstractGeniAggregateManager.AggregateManagerReply reply = null;

            try {
                setAndRunState(allocateSliversOnAuthorityState);

                if (allocateSliversOnAuthorityState.isSuccess()) {
                    reply = allocateSliversOnAuthorityState.getRawReply();

                } else if (allocateSliversOnAuthorityState.getException() instanceof AggregateManagerWrapper.VlanUnavailableException ||
                        allocateSliversOnAuthorityState.getException() instanceof AggregateManagerWrapper.AnyVlanUnsupportedException) {
                    JFedException e = (JFedException) allocateSliversOnAuthorityState.getException();
                    assert e.getSfaReply() == null || e.getSfaReply() instanceof AbstractGeniAggregateManager.AggregateManagerReply :
                            "Unexpected reply type: " + e.getSfaReply().getClass().getName();
                    //catch error, and report it to the director instead
                    reply = (AbstractGeniAggregateManager.AggregateManagerReply) e.getSfaReply();
                } else if (allocateSliversOnAuthorityState.getException() instanceof JFedException) {

                    JFedException e = (JFedException) allocateSliversOnAuthorityState.getException();

                    if (e.getSfaReply() != null && e.getSfaReply() instanceof AbstractGeniAggregateManager.AggregateManagerReply) {
                        reply = (AbstractGeniAggregateManager.AggregateManagerReply) e.getSfaReply();
                    } else {
                        if (StitchingDirector.isUnrecoverableError(e) == Boolean.TRUE) {
                            log.error("Giving up as {} reported an unrecoverable error {}", experimentPart.getName(), e.getMessage());
                            allocationDirector.reportGiveUp();
                        } else {
                            retryAllocate(stitchAction);
                            return; //we don't want to process this allocate-try any further
                        }
                    }
                } else {
                    log.error("Unexpected state: unknown error while allocating", allocateSliversOnAuthorityState.getException());
                }
            } catch (JFedException ex) {
                log.error("UNEXPECTED error while allocating", ex);
            }

            log.trace("   Processing result of CreateSliver call for " + experimentPart.getName());
            StitchingDirector.AllocateVerdict verdict = allocationDirector.processAllocateResult(stitchAction, reply);

            if (verdict == StitchingDirector.AllocateVerdict.CALLER_DELAYED_RETRY) {
                retryAllocate(stitchAction);
            }
        }

        private void retryAllocate(StitchingDirector.StitchAction stitchAction) throws InterruptedException {
            if (++allocateStitchActionRetryCount > MAX_ALLOCATE_STITCH_ACTION_TRIES) {
                log.error("Giving up stitching as number of allocation tries on {} exceeded maximum of {}", experimentPart.getName(), MAX_ALLOCATE_STITCH_ACTION_TRIES);
                allocationDirector.reportGiveUp();
                return;
            }

            log.info("retryAllocate: CreateSliver failed (allocateStitchActionRetryCount=" + allocateStitchActionRetryCount + "), " +
                    "will delete and try again in " + TIME_BETWEEN_ALLOCATE_RETRIES_MS + "ms " +
                    "now=" + HtmlLogOutput.dateToString(new Date()));

            Thread.sleep(TIME_BETWEEN_ALLOCATE_RETRIES_MS);
            allocationDirector.reportReadyForRetry(experimentPart.getConnectSfaAuthority());
        }


        public void submitStitchAction(AllocateAction stitchAction) {
            boolean success = queue.offer(stitchAction);
            assert success : "Could not add StitchAction to ExperimentPart " + experimentPart.getName();
        }


        public class GetAdvertisementState extends SingleCallState<GetAdvertisementTask> {
            protected GetAdvertisementState() {
                super("Fetch advertisement from " + experimentPart.getName() + " for list of available VLAN's",
                        hltf.getAvailableAdvertisement(experimentPart.getConnectSfaAuthority()));
            }
        }

        public class WaitForNextAllocateActionState extends be.iminds.ilabt.jfed.highlevel.jobs.State {

            private AllocateAction allocateAction;

            protected WaitForNextAllocateActionState() {
                super("Waiting for next StitchAction instruction");
            }

            @Nonnull
            @Override
            protected ExperimentTaskStatus executeState(Job<?> job) throws InterruptedException, JFedException {
                allocateAction = queue.take();

                return ExperimentTaskStatus.SUCCESS;
            }

            public AllocateAction getAllocateAction() {
                return allocateAction;
            }
        }
    }

    private Collection<SfaExperimentPart> getSfaParts() {
        return experiment.getParts().stream()
                .filter(part -> part instanceof SfaExperimentPart)
                .map(SfaExperimentPart.class::cast)
                .collect(Collectors.toList());

    }


    private Collection<NonSfaExperimentPartController> getNonSfaExperimentPartControllers() {

        Set<NonSfaExperimentPartController> controllers = new HashSet<>();

        for (ExperimentPart experimentPart : experiment.getParts()) {
            if (experimentPart instanceof SfaExperimentPart)
                continue;


            ExperimentPartController epc = epcManager.getController(experimentPart);
            if (!(epc instanceof NonSfaExperimentPartController))
                throw new RuntimeException("Expected a non-sfa ExperimentPartController");

            controllers.add((NonSfaExperimentPartController) epc);
        }

        return controllers;
    }

    private boolean hasNonSfaParts() {
        return experiment.getParts().stream().anyMatch(part -> !(part instanceof SfaExperimentPart));
    }

    private boolean hasSfaParts() {
        return experiment.getParts().stream().anyMatch(part -> (part instanceof SfaExperimentPart));
    }

    public final class CreateNonSfaExperimentPartsContainersState extends SlicedState<StateSlice> {

        private final List<StateSlice> slices;

        protected CreateNonSfaExperimentPartsContainersState() {
            super("Creating non-SFA experimentparts containers");
            slices = getNonSfaExperimentPartControllers().stream()
                    .map(epc -> epc.initializeContainer(AllocateExperimentJob.this))
                    .collect(Collectors.toList());
        }

        @Override
        public Collection<StateSlice> getSlices() {
            return slices;
        }
    }


    public final class StartNonSfaPartsState extends SlicedState<StateSlice> {

        private final List<StateSlice> slices;

        protected StartNonSfaPartsState() {
            super("Starting non-SFA resources");

            slices = new ArrayList<>();

            for (ExperimentPart experimentPart : experiment.getParts()) {
                if (experimentPart instanceof SfaExperimentPart)
                    continue;

                ExperimentPartController epc = epcManager.getController(experimentPart);
                if (!(epc instanceof NonSfaExperimentPartController))
                    throw new RuntimeException("Expected a non-sfa ExperimentPartController");

                //noinspection unchecked
                slices.add(
                        ((NonSfaExperimentPartController) epc).start(AllocateExperimentJob.this, experimentPart));

            }
        }

        @Override
        public Collection<StateSlice> getSlices() {
            return slices;
        }
    }

}
