<?php

namespace FiloBlu\Flow\Model;

use Exception;
use FiloBlu\Flow\Api\DocumentInterfaceFactory;
use FiloBlu\Flow\Api\DocumentRepositoryInterface;
use FiloBlu\Flow\Api\SwatchInterfaceFactory;
use FiloBlu\Flow\Helper\Data;
use FiloBlu\Flow\Helper\LoggerProvider;
use FiloBlu\Flow\Model\Channel\In\AbstractModel;
use FiloBlu\Flow\Model\Channel\In\Document;
use FiloBlu\Flow\Model\Channel\In\Images;
use FiloBlu\Flow\Model\Channel\In\Swatch;
use FiloBlu\Flow\Model\Channel\In\Videos;
use FiloBlu\Flow\Model\System\ProductAttributesFactory as FlowAttributesFactory;
use Magento\Framework\App\Config\ScopeConfigInterface;
use Magento\Framework\App\ResourceConnection;
use Magento\Framework\Exception\LocalizedException;
use Magento\Framework\ObjectManagerInterface;
use Magento\Framework\Registry;
use Magento\Store\Model\ScopeInterface;
use Monolog\Logger;
use RuntimeException;
use stdClass;
use Throwable;
use Zend_Db;
use Zend_Db_Expr;

use function array_key_exists;
use function count;
use function in_array;
use function is_array;

/**
 * Class Processor
 * @package FiloBlu\Flow\Model
 */
class Processor
{
    /**
     * @var int
     */
    const META_PROCESSED_NONE = 0;
    /**
     * @var int
     */
    const META_PROCESSED_SUCCESS = 1;
    /**
     * @var int
     */
    const META_PROCESSED_ERROR = 2;
    /**
     * Max tries before mark the flow as faulty
     *
     * @var int
     */
    const MAX_IMAGE_PROCESS_RETRY_COUNT = 3;


    /**
     * Max tries before mark the flow as faulty
     *
     * @var int
     */
    const MAX_VIDEO_PROCESS_RETRY_COUNT = 3;

    /**
     * Max tries before mark the flow as faulty
     *
     * @var int
     */
    const MAX_DOCUMENT_PROCESS_RETRY_COUNT = 3;

    /**
     * @var string
     */
    const CHANNEL_TYPE_IMAGE_CODE = 'image';

    /**
     * @var string
     */
    const CHANNEL_TYPE_DOCUMENT_CODE = 'document';

    /**
     * @var string
     */
    const CHANNEL_TYPE_VIDEO_CODE = 'video';

    /**
     * @var string
     */
    const CHANNEL_TYPE_SWATCH_CODE = 'swatch';

    /**
     * @var string
     */
    const CHANNEL_TYPE_FILE_CODE = 'file';

    /**
     * @var int
     */
    const LOWER_PROCESS_ORDER = 1000;

    /**
     * @var int
     */
    const LOWEST_PRIORITY_WEIGHT = 1000;

    /**
     * Processing order of known flows
     *
     * @var array
     */
    protected static $_processOrder = [
        'From\Product'              => 0,
        'From\Eavs'                 => 1,
        'From\Grouped'              => 2,
        'From\EavsGrouped'          => 3,
        'From\Stock'                => 4,
        'From\Multistock'           => 5,
        'From\Price'                => 6,
        'From\SharedCatalog'        => 7,
        'From\SharedCatalogPrices'  => 8,
        'From\Brotherhood'          => 9,
        'Imagesflow'                => 10,
        'Videosflow'                => 11
    ];

    /**
     * Waiting condition
     *
     * An inbound flow will not be executed if a condition will met
     * @var array
     */
    protected static $_waitingQueue = [

        /* Eavs will not be processed if there is ite in processing */
        'From\Eavs' => [
            [
                'condition' => 'flow = :flow AND status IN (:status)',
                'arguments' => [
                    'flow'   => 'From\Product',
                    'status' => [
                        Inboundflow::STATUS_PROCESSING
                    ]
                ]
            ]
        ],
        'From\Grouped' => [
            [
                'condition' => 'flow IN (:flow) AND status IN (:status)',
                'arguments' => [
                    'flow'   => [
                        'From\Eavs',
                        'From\Product'
                    ],
                    'status' => [
                        Inboundflow::STATUS_PROCESSING
                    ]
                ]
            ]
        ],
        'From\EavsGrouped' => [
            [
                'condition' => 'flow IN (:flow) AND status IN (:status)',
                'arguments' => [
                    'flow'   => [
                        'From\Grouped',
                        'From\Eavs',
                        'From\Product'
                    ],
                    'status' => [
                        Inboundflow::STATUS_PROCESSING
                    ]
                ]
            ]
        ],
        'From\Price' => [
            [
                'condition' => 'flow IN (:flow) AND status IN (:status)',
                'arguments' => [
                    'flow'   => [
                        'From\EavsGrouped',
                        'From\Grouped',
                        'From\Eavs',
                        'From\Product'
                    ],
                    'status' => [
                        Inboundflow::STATUS_PROCESSING
                    ]
                ]
            ]
        ],
        'From\SharedCatalogPrices' => [
            [
                'condition' => 'flow IN (:flow) AND status IN (:status)',
                'arguments' => [
                    'flow'   => [
                        'From\EavsGrouped',
                        'From\Grouped',
                        'From\Eavs',
                        'From\Product',
                        'From\Price'
                    ],
                    'status' => [
                        Inboundflow::STATUS_PROCESSING
                    ]
                ]
            ]
        ]
    ];

    /**
     * Priority weights
     *
     * @var array
     */
    protected static $_priorityWeights = [
        Inboundflow::PRIORITY_CRITICAL => 0,
        Inboundflow::PRIORITY_HIGH     => 1,
        Inboundflow::PRIORITY_NORMAL   => 2,
        Inboundflow::PRIORITY_LOW      => 4
    ];

    /**
     * @var Logger
     */
    protected $_logger;

    /**
     * @var array
     */
    protected $_config = [];

    /**
     * @var Data
     */
    protected $_helperData;

    /**
     * @var Channel
     */
    protected $_channel;

    /**
     * @var Inboundflow
     */
    protected $_inboundflow;

    /**
     * @var Imagesflow
     */
    protected $_imageflow;

    /**
     * @var array
     */
    protected $_channels = [];

    /**
     * @var array
     */
    protected $_inboundItems = [];

    /**
     * @var ResourceConnection
     */
    protected $_resourceConnection;

    /**
     * @var ObjectManagerInterface
     */
    protected $objectManager;

    /**
     * @var ScopeConfigInterface
     */
    protected $_scopeConfig;

    /**
     * @var FlowAttributesFactory
     */
    protected $productAttributesFactory;

    /**
     * @var InboundflowFactory
     */
    protected $inboundFlowFactory;

    /**
     * @var ImagesflowFactory
     */
    protected $imageFlowFactory;

    /**
     * @var VideosflowFactory
     */
    protected $videoFlowFactory;

    /**
     * @var Registry
     */
    protected $registry;
    /**
     * @var ResourceModel\Inboundflow\CollectionFactory
     */
    protected $inboundFlowCollectionFactory;

    /**
     * @var DocumentInterfaceFactory
     */
    protected $document;

    /**
     * @var SwatchInterfaceFactory
     */
    protected $swatchFactory;

    /**
     * @var Swatch
     */
    protected $swatch;

    /**
     * Default priorities to process
     *
     * @var array
     */
    protected $_priority = [
        Inboundflow::PRIORITY_HIGH,
        Inboundflow::PRIORITY_NORMAL,
        Inboundflow::PRIORITY_LOW
    ];

    /**
     * @var bool
     */
    protected $initialized = false;

    /**
     * @var DocumentRepositoryInterface
     */
    protected $documentRepository;

    /**
     * @var Document
     */
    protected $doc;

    /**
     * Processor constructor.
     * @param Registry $registry
     * @param Data $helperData
     * @param LoggerProvider $loggerProvider
     * @param Channel $channel
     * @param ImagesflowFactory $imageFlowFactory
     * @param VideosflowFactory $videosflowFactory
     * @param DocumentRepositoryInterface $documentRepository
     * @param DocumentInterfaceFactory $document
     * @param Document $doc
     * @param SwatchInterfaceFactory $swatchFactory
     * @param Swatch $swatch
     * @param ScopeConfigInterface $scopeConfig
     * @param ResourceConnection $resourceConnection
     * @param FlowAttributesFactory $productAttributesFactory
     * @param InboundflowFactory $inboundFlowFactory
     * @param ResourceModel\Inboundflow\CollectionFactory $inboundFlowCollectionFactory
     * @param ObjectManagerInterface $objectManager
     */
    public function __construct(
        Registry $registry,
        Data $helperData,
        LoggerProvider $loggerProvider,
        Channel $channel,
        ImagesflowFactory $imageFlowFactory,
        VideosflowFactory $videosFlowFactory,
        DocumentRepositoryInterface $documentRepository,
        DocumentInterfaceFactory $document,
        Document $doc,
        SwatchInterfaceFactory $swatchFactory,
        Swatch $swatch,
        ScopeConfigInterface $scopeConfig,
        ResourceConnection $resourceConnection,
        FlowAttributesFactory $productAttributesFactory,
        InboundflowFactory $inboundFlowFactory,
        ResourceModel\Inboundflow\CollectionFactory $inboundFlowCollectionFactory,
        ObjectManagerInterface $objectManager
    ) {
        $this->_logger = $loggerProvider->getLogger();
        $this->_channel = $channel;
        $this->_scopeConfig = $scopeConfig;
        $this->_helperData = $helperData;
        $this->productAttributesFactory = $productAttributesFactory;
        $this->inboundFlowFactory = $inboundFlowFactory;
        $this->imageFlowFactory = $imageFlowFactory;
        $this->videoFlowFactory = $videosFlowFactory;
        $this->documentRepository = $documentRepository;
        $this->document = $document;
        $this->objectManager = $objectManager;
        $this->registry = $registry;
        $this->_resourceConnection = $resourceConnection;
        $this->inboundFlowCollectionFactory = $inboundFlowCollectionFactory;
        $this->doc = $doc;
        $this->swatch = $swatch;
        $this->swatchFactory = $swatchFactory;
    }

    /**
     * @param Inboundflow $flow
     * @return int
     * @deprecated
     * Get the priority weight for the given flow
     *
     */
    protected static function getPriorityWeight(Inboundflow $flow)
    {
        $priority = $flow->getPriority();

        return self::$_priorityWeights[$priority] ?? self::LOWEST_PRIORITY_WEIGHT;
    }

    /**
     * @param Inboundflow $flow
     * @return int
     * @deprecated
     * Get the process order for the given flow
     *
     */
    protected static function getProcessOrder(Inboundflow $flow)
    {
        $type = $flow->getFlow();

        return self::$_processOrder[$type] ?? self::LOWER_PROCESS_ORDER;
    }

    /**
     * Process flows with priority set to critical
     *
     * @throws Exception
     */
    public function processCritical()
    {
        $this->addPriority(Inboundflow::PRIORITY_CRITICAL);
        $this->fetchChannels();
        $this->receiveData();
        $this->parse();
        $this->process();
    }

    /**
     * Add priority to process.
     *
     * @param $priority
     * @throws Exception
     * @see \FiloBlu\Flow\Model\Inboundflow
     */
    protected function addPriority($priority)
    {
        if (!Inboundflow::isValidPriority($priority)) {
            throw new Exception("Invalid priority {$priority}");
        }

        $isCritical = in_array(Inboundflow::PRIORITY_CRITICAL, $this->_priority, false);

        if ($isCritical || $priority === Inboundflow::PRIORITY_CRITICAL) {
            $this->_priority = [Inboundflow::PRIORITY_CRITICAL];
            return;
        }

        if (in_array($priority, $this->_priority, false)) {
            return;
        }

        $this->_priority[] = $priority;
    }

    /**
     * Fetch form channels on the specified bus
     *
     * @param int $bus
     * @throws Exception
     */
    public function fetchChannels($bus = 0)
    {
        $this->init();
        $channels = $this->getChannels();

        if (!count($channels)) {
            throw new Exception('No channel configured');
        }

        /** @var AbstractModel $channel */
        foreach ($channels as $name => $channel) {
            if ($bus && (int)$channel->getConfig('bus') !== $bus) {
                continue;
            }

            $channelPriority = $channel->getConfig('priority');

            if (!in_array($channelPriority, $this->_priority, false)) {
                continue;
            }

            /* One image channel at time can fetch */
            if ($channel->getConfig('type') === self::CHANNEL_TYPE_IMAGE_CODE) {
                $pending = $this->inboundFlowCollectionFactory->create()
                    ->addFieldToFilter('status', ['in' => [Inboundflow::STATUS_RECEIVING, Inboundflow::STATUS_FOUND]])
                    ->addFieldToFilter('type', self::CHANNEL_TYPE_IMAGE_CODE)
                    ->setPageSize(1)
                    ->getItems();

                if (count($pending) !== 0) {
                    continue;
                }
            }

            try {
                $list = $channel->rlist();

                if ($list !== false && count($list)) {
                    $this->populateModel($channel, $name, $list);
                }
            } catch (Exception $exception) {
                $this->_logger->error($exception->getMessage(), ['exception' => $exception]);
            } catch (Throwable $throwable) {
                $this->_logger->error($throwable->getMessage(), ['exception' => $throwable]);
            }
        }
    }

    /**
     *
     */
    public function init()
    {
        if ($this->initialized) {
            return;
        }

        // register_shutdown_function(callable $callback, mixed ...$args): void

        $attributesToCopy = $this->productAttributesFactory->create()
            ->getFinalAttributesToCopy();

        // TODO: Remove registry
        $this->registry->register('attribute_configurable_copy', $attributesToCopy, true);
        $this->initConfig();
        // TODO: too slow, make it faster setChannels
        $this->setChannels();
        $this->cleanupDeadFlows();
        $this->setInboundFlowItems();

        $this->initialized = true;
    }

    /**
     *
     */
    public function initConfig()
    {
        $collection = $this->_channel->getCollection();

        if (empty($this->_config['channels'])) {
            $this->_config['channels'] = new stdClass();
            $this->_config['channels_config'] = new stdClass();
        }

        foreach ($collection as $channel) {
            $config_data = json_decode($channel->getData('data'), true);
            $channel_name = $channel->getData('name');

            if (!isset($config_data['config']['flow'])) {
                continue;
            }

            $channel_model_class = $this->_helperData->getChannelModels($config_data['config']['flow']);
            $channel_type = $channel->getData('type');
            $this->_config['channels']->{$channel_type}[$channel_name] = $channel_model_class;
            $this->_config['channels_config']->{$channel_type}[$channel_name] = $config_data;
        }
    }

    /**
     * @return void
     */
    public function cleanupDeadFlows()
    {
        $channels = $this->getChannels();

        if (!count($channels)) {
            throw new RuntimeException('No channel configured');
        }

        $connection = $this->_resourceConnection->getConnection();
        $flowTable = $connection->getTableName('flow_inboundflow');

        /** @var AbstractModel $channel */
        foreach ($channels as $name => $channel) {
            $where = [];
            $where['channel = ?'] = $name;
            $where['TIMESTAMPDIFF(MINUTE,TIMESTAMP(last_activity), CURRENT_TIMESTAMP()) > ?'] = $channel
                ->getChannelConfig()
                ->getKillAfter();
            $where['status IN (?)'] = [
                Inboundflow::STATUS_RECEIVING,
                Inboundflow::STATUS_PARSING,
                Inboundflow::STATUS_PROCESSING
            ];
            $connection->update($flowTable, ['status' => Inboundflow::STATUS_DEAD], $where);
        }
    }

    /**
     * Get all channels from config
     *
     * @return array
     */
    public function getChannels()
    {
        return $this->_channels;
    }

    /**
     * @param string $type
     * @return $this
     */
    public function setChannels($type = 'in')
    {
        if (!isset($this->_config['channels'])) {
            return $this;
        }

        $channels = $this->_config['channels'];

        foreach ($channels->{$type} as $name => $channelModelClass) {
            $flowModuleName = $this->_helperData->getFlowModuleName($this->_config, $name, $type);
            $channelClass = '\FiloBlu\\' . $flowModuleName . '\Model\Channel\\' . ucfirst($type) . '\\' . ucfirst(
                    $channelModelClass
                );
            $model = false;

            try {
                $model = $this->objectManager->create($channelClass);
                $model->load($name);
            } catch (Exception $e) {
                $this->_logger->info($e->getMessage());
            }

            if ($model) {
                $this->_channels[$name] = $model;
            } else {
                $this->_logger->error("Error in channel $type $channelModelClass config. Channel skipped");
            }
        }

        return $this;
    }

    /**
     * TODO: Find better place
     * ci sono due problemi in questo metodo:
     * 1] i dati vengono caricati in memoria anche se non sono utilizzati
     * 2] i dati che vengono caricati sono senza alcun filtro quindi in memoria ci troviamo anceh oggetti gia processati
     *    a tal proposito e' stato aggiunto un filtro sulla collection con degli $allowedStatuses
     * @return array
     */
    public function setInboundFlowItems()
    {
        $this->_logger->info(__METHOD__);

        $allowedStatuses = [Inboundflow::STATUS_FOUND];

        if (!$this->_inboundItems) {
            $collection = $this->inboundFlowCollectionFactory->create();
            $collection->addFieldToFilter('status', ['in' => $allowedStatuses]);
            foreach ($collection as $inbound) {
                $this->_inboundItems[] = $inbound->getData('name');
            }
        }

        return $this->_inboundItems;
    }

    /**
     * TODO : fix better logic!
     *
     * @param $channel
     * @param $channelName
     * @param $list
     * @return mixed
     * @throws Exception
     */
    public function populateModel($channel, $channelName, $list)
    {
        $channelType = $channel->getConfig('type');
        $channelFlow = $channel->getConfig('flow');
        $channelPriority = $channel->getConfig('priority');

        switch ($channelType) {
            case self::CHANNEL_TYPE_VIDEO_CODE:
                /** @var Inboundflow $model */
                $model = $this->inboundFlowFactory->create();
                $model->setName('videos_' . date('Ymd_Hmi') . '.csv');
                $model->setFlow($channelFlow);
                $model->setType($channelType);
                $model->setChannel($channelName);
                $model->setRetry(0);
                $model->setStatus(Inboundflow::STATUS_FOUND);
                $model->setPriority($channelPriority);

                $lid = null;

                try {
                    $model->save();
                    $lid = $model->getId();

                    if (!$lid) {
                        return false;
                    }
                } catch (Exception $e) {
                    $this->_logger->info($e->getMessage());
                }

                foreach ($list as $item) {
                    $model = $this->videoFlowFactory->create();
                    $model->setMetaFile($lid);
                    $model->setVideoName($item);
                    $model->setVideoStatus(Inboundflow::STATUS_FOUND);
                    $model->setData('video_queue_time', date('Y-m-d H:i:s'));
                    try {
                        $model->save();
                    } catch (Exception $e) {
                        $this->_logger->info($e->getMessage());
                    }
                }

                break;

            case self::CHANNEL_TYPE_IMAGE_CODE:
                /** @var Inboundflow $model */
                $model = $this->inboundFlowFactory->create();
                $model->setName('images_' . date('Ymd_Hmi') . '.csv');
                $model->setFlow($channelFlow);
                $model->setType($channelType);
                $model->setChannel($channelName);
                $model->setRetry(0);
                $model->setStatus(Inboundflow::STATUS_FOUND);
                $model->setPriority($channelPriority);
                $model->setLastActivity(date('Y-m-d H:i:s'));

                $lid = null;

                try {
                    $model->save();
                    $lid = $model->getId();

                    if (!$lid) {
                        return false;
                    }
                } catch (Exception $e) {
                    $this->_logger->info($e->getMessage());
                }

                foreach ($list as $item) {
                    $model = $this->imageFlowFactory->create();
                    $model->setMetaFile($lid);
                    $model->setImageName($item);
                    $model->setImageStatus(Inboundflow::STATUS_FOUND);
                    $model->setData('image_queue_time', date('Y-m-d H:i:s'));
                    try {
                        $model->save();
                    } catch (Exception $e) {
                        $this->_logger->info($e->getMessage());
                    } finally {
                        $this->_resourceConnection->getConnection()
                            ->update(
                                $this->_resourceConnection->getTableName('flow_inboundflow'),
                                ['last_activity' => date('Y-m-d H:i:s')],
                                ['id = ?'=> $lid]
                            );
                    }
                }

                break;

            case self::CHANNEL_TYPE_DOCUMENT_CODE:
                /** @var Inboundflow $model */
                $model = $this->inboundFlowFactory->create();
                $model->setName($channelName . '_' . date('Ymd_Hmi') . '.csv');
                $model->setFlow($channelFlow);
                $model->setType($channelType);
                $model->setChannel($channelName);
                $model->setRetry(0);
                $model->setStatus(Inboundflow::STATUS_FOUND);
                $model->setPriority($channelPriority);

                $lid = null;

                try {
                    $model->save();
                    $lid = $model->getId();

                    if (!$lid) {
                        return false;
                    }
                } catch (Exception $e) {
                    $this->_logger->info($e->getMessage());
                }

                $connector = $this->getChannel($channelName)->getConnector();
                foreach ($list as $item) {
                    $document = $this->document->create();
                    $document->setName($item);
                    $document->setSource($connector->ftpDir);
                    $document->setDestination($connector->destinationDir);
                    $document->setStatus(Inboundflow::STATUS_FOUND);
                    $document->setMetaFile($lid);
                    $document->setMetaInsertTime(date('Y-m-d H:i:s'));
                    $document->setMetaProcessed(0);
                    try {
                        $this->documentRepository->save($document);
                    } catch (Exception $e) {
                        $this->_logger->info($e->getMessage());
                    }
                }
                break;

            case self::CHANNEL_TYPE_SWATCH_CODE:
                /** @var Inboundflow $model */
                $model = $this->inboundFlowFactory->create();
                $model->setName($channelName . '_' . date('Ymd_Hmi') . '.csv');
                $model->setFlow($channelFlow);
                $model->setType($channelType);
                $model->setChannel($channelName);
                $model->setRetry(0);
                $model->setStatus(Inboundflow::STATUS_FOUND);
                $model->setPriority($channelPriority);

                $lid = null;

                try {
                    $model->save();
                    $lid = $model->getId();

                    if (!$lid) {
                        return false;
                    }
                } catch (Exception $e) {
                    $this->_logger->info($e->getMessage());
                }

                foreach ($list as $item) {
                    $model = $this->swatchFactory->create();
                    $model->setMetaFile($lid);
                    $model->setName($item);
                    $model->setStatus(Inboundflow::STATUS_FOUND);
                    $model->setMetaInsertTime(date('Y-m-d H:i:s'));
                    try {
                        $model->save();
                    } catch (Exception $e) {
                        $this->_logger->info($e->getMessage());
                    }
                }

                break;

            case self::CHANNEL_TYPE_FILE_CODE:
                foreach ($list as $item) {
                    // Se il file è già presente nella tabella inbound, non deve essere reimportato

                    if (in_array($item, $this->_inboundItems, false)) {
                        continue;
                    }

                    $this->copyToIte($channel, $item);

                    $model = $this->inboundFlowFactory->create();
                    $model->setName($item);
                    $model->setFlow($channelFlow);
                    $model->setType($channelType);
                    $model->setChannel($channelName);
                    $model->setRetry(0);
                    $model->setStatus(Inboundflow::STATUS_FOUND);
                    $model->setPriority($channelPriority);
                    try {
                        $model->save();
                    } catch (Exception $e) {
                        $this->_logger->info($e->getMessage());
                    }
                }
                break;

            default:
                throw new Exception("Unknown channel type $channelType");
        }

        if (isset($model)) {
            return $model;
        }

        return true;
    }

    /**
     * Get channel from name
     *
     * @param string $name
     * @return Channel|null
     */
    public function getChannel($name)
    {
        $this->init();
        return $this->_channels[$name] ?? null;
    }

    /**
     * @param \FiloBlu\Flow\Model\Channel\In\AbstractModel $channel
     * @param string|null $filename
     * @return void
     * @throws \Exception
     */
    public function copyToIte(AbstractModel $channel, $filename)
    {
        $config = $channel->getChannelConfig();

        if (!$config->getCopyToIte() || !$filename) {
            return;
        }

        $iteChannel = $this->getChannel($config->getIteChannelName());

        $connector = $iteChannel->getConnector();

        switch ($config->getCopyToIteMode()) {
            case \FiloBlu\Flow\Model\Channel\Config::COPY_TO_ITE_MODE_FROM_EAVS_FILE:
                $iteFilename = str_replace(
                    $config->getCopyToIteReplaceFrom(),
                    $config->getCopyToIteReplaceTo(),
                    $filename
                );
                break;

            default:
            case \FiloBlu\Flow\Model\Channel\Config::COPY_TO_ITE_MODE_AUTO:
                $iteFilename = 'ite_' . date('Ymd_Hmi') . '.csv';
                break;
        }

        $localIteFile = $connector->getLocalDir() . '/' . $iteFilename;

        if (file_exists($localIteFile)) {
            return;
        }

        $connector->openConnection();
        $connector->downloadFile($localIteFile, $filename);
        $connector->close();

        try {
            $this->inboundFlowFactory
                ->create([
                    'data' => [
                        'name'     => $iteFilename,
                        'flow'     => $iteChannel->getConfig('flow'),
                        'type'     => $iteChannel->getConfig('type'),
                        'priority' => $iteChannel->getConfig('priority'),
                        'channel'  => $iteChannel->getName(),
                        'retry'    => 0
                    ]
                ])
                ->setStatus(Inboundflow::STATUS_RECEIVED)
                ->save();
        } catch (Exception $e) {
            $this->_logger->info($e->getMessage());
        }
    }

    /**
     * Retrieve remote files and save them locally
     *
     * @throws Exception
     */
    public function receiveData()
    {
        $this->init();

        $flows = $this->getInboundFlowsToProcess(Inboundflow::STATUS_FOUND);

        foreach ($flows as $flow) {
            try {
                if ($flow->getStatus() === Inboundflow::STATUS_RECEIVING) {
                    continue;
                }

                $flow->setStatus(Inboundflow::STATUS_RECEIVING)->save();

                $channelName = $flow->getChannel();
                $channel = $this->getChannel($channelName);

                if ($channel === null) {
                    $flow->setStatus(Inboundflow::STATUS_ERROR);
                    $flow->setLog("No channel found with name {$channelName}");
                    $flow->save();
                    continue;
                }

                if (!($channel instanceof AbstractModel)) {
                    $message = "Error in channel {$channel->model} config during receiving";
                    $flow->setLog($message);
                    $flow->setStatus(Inboundflow::STATUS_ERROR)->save();
                    continue;
                }

                $received = $channel->receiveFile($flow);

                if ($received) {
                    $flow->setStatus(Inboundflow::STATUS_RECEIVED)->save();
                } else {
                    $flow->setLog('Error while receiving file');
                    $flow->setStatus(Inboundflow::STATUS_ERROR)->save();
                }
            } catch (Exception $e) {
                $flow->setLog($e->getMessage());
                $flow->setStatus(Inboundflow::STATUS_ERROR)->save();
            }
        }
    }

    /**
     * Retrieve inbound flows to be processed sorted by process order and priority
     * filtered by status
     *
     * @param $status
     * @return Inboundflow [] |DataObject[]
     * @throws Exception
     */
    public function getInboundFlowsToProcess($status)
    {
        if (!Inboundflow::isValidStatus($status)) {
            throw new Exception("Invalid status $status");
        }

        $connection = $this->_resourceConnection->getConnection();
        $processOrder = implode(
            ',',
            array_map(static function ($a) use ($connection) {
                return $connection->quote((string)$a);
            }, array_keys(self::$_processOrder))
        );
        $priorities = implode(
            ',',
            array_map(static function ($a) use ($connection) {
                return $connection->quote((string)$a);
            }, array_keys(self::$_priorityWeights))
        );

        /** @var Inboundflow [] $flows */
        $inboundFlowCollection = $this->inboundFlowCollectionFactory->create();
        $inboundFlowCollection->addFieldToFilter('status', $status);
        $inboundFlowCollection->addFieldToFilter('priority', ['in' => $this->_priority]);
        $inboundFlowCollection->getSelect()
            ->order('retry ASC')
            ->order(new Zend_Db_Expr("FIELD(flow, $processOrder) ASC"))
            ->order(new Zend_Db_Expr("FIELD(priority, $priorities) ASC"))
            ->order('id ASC');

        return $inboundFlowCollection->getItems();
    }

    /**
     * Parse
     * @throws Exception
     */
    public function parse()
    {
        $this->init();

        $flows = $this->getInboundFlowsToProcess(Inboundflow::STATUS_RECEIVED);

        foreach ($flows as $flow) {
            if ($flow->getStatus() === Inboundflow::STATUS_PARSING) {
                continue;
            }

            $flow->setStatus(Inboundflow::STATUS_PARSING);
            $flow->save();

            $channelName = $flow->getChannel();
            $channel = $this->getChannel($channelName);

            if ($channel === null) {
                $flow->setStatus(Inboundflow::STATUS_ERROR);
                $flow->setLog("No channel found with name {$channelName}");
                $flow->save();
                continue;
            }

            try {
                if ($channel->parse($flow)) {
                    $flow->setStatus(Inboundflow::STATUS_PARSED);
                    $flow->save();
                    continue;
                }

                $flow->setLog('Error while parsing');
                $flow->setStatus(Inboundflow::STATUS_ERROR);
                $flow->save();
            } catch (Exception $e) {
                $flow->setLog($e->getMessage());
                $flow->setStatus(Inboundflow::STATUS_ERROR);
                $flow->save();
            }
        }
    }

    /**
     * @throws Exception
     */
    public function process()
    {
        $this->init();

        $flows = $this->getInboundFlowsToProcess(Inboundflow::STATUS_PARSED);

        foreach ($flows as $flow) {
            $originalStatus = $flow->getStatus();

            if ($flow->getStatus() === Inboundflow::STATUS_PROCESSING) {
                continue;
            }

            if (!$this->canRun($flow)) {
                continue;
            }

            $flow->setStatus(Inboundflow::STATUS_PROCESSING);
            $flow->save();

            $this->_logger->info($flow->getFlow());
            $type = $flow->getType();

            $channelConfig = $this->getChannel($flow->getChannel())->getChannelConfig();

            $maxRetryCount = $channelConfig->getMaxRetryCount();
            $retry = (int)$flow->getRetry();

            if ($originalStatus === Inboundflow::STATUS_ERROR && $channelConfig->isRetryAllowed(
                ) && $retry <= $maxRetryCount) {
                $this->_helperData->resetItemToState($flow, Inboundflow::STATUS_PARSED, false, true);
            }

            $log = $flow->getLog();

            try {
                switch ($type) {
                    case self::CHANNEL_TYPE_VIDEO_CODE:
                        $this->processVideos($flow);
                        break;
                    case self::CHANNEL_TYPE_IMAGE_CODE:
                        $this->processImages($flow);
                        break;
                    case self::CHANNEL_TYPE_FILE_CODE:
                        $this->processFile($flow);
                        break;
                    case self::CHANNEL_TYPE_DOCUMENT_CODE:
                        $this->processDocuments($flow);
                        break;
                    case self::CHANNEL_TYPE_SWATCH_CODE:
                        $this->processSwatches($flow);
                        break;
                    default:
                        throw new Exception("Unknown channel type $type");
                }

                if (!$flow->getRetry()) {
                    $flow->setStatus(Inboundflow::STATUS_PROCESSED);
                    $flow->save();
                }
            } catch (Exception $e) {
                $status = Inboundflow::STATUS_ERROR;
                $maxRetryCount = $channelConfig->getMaxRetryCount();
                $retry = (int)$flow->getRetry();
                if ($retry < $maxRetryCount && $channelConfig->isRetryAllowed()) {
                    $status = Inboundflow::STATUS_PARSED;
                    $flow->setRetry($retry + 1);
                }
                $flow->setStatus($status);
                $flow->setLastActivity(date('Y-m-d H:i:s'));
                $flow->setLog(__METHOD__ . ': ' . $e->getMessage() . "\n" . $log);
                $flow->save();
                $this->_logger->error($e->getMessage(), ['exception' => $e]);
            } catch (Throwable $t) {
                $status = Inboundflow::STATUS_ERROR;
                $maxRetryCount = $channelConfig->getMaxRetryCount();
                $retry = (int)$flow->getRetry();
                if ($retry < $maxRetryCount && $channelConfig->isRetryAllowed()) {
                    $status = Inboundflow::STATUS_PARSED;
                    $flow->setRetry($retry + 1);
                }
                $flow->setStatus($status);
                $flow->setLastActivity(date('Y-m-d H:i:s'));
                $flow->setLog(__METHOD__ . ': ' . $t->getMessage() . "\n" . $log);
                $flow->save();
                $this->_logger->error($t->getMessage(), ['exception' => $t]);
            }
        }
    }

    /**
     * @param Inboundflow $inboundFlow
     * @return bool
     * @throws LocalizedException
     */
    public function canRun(Inboundflow $inboundFlow)
    {
        $flowType = $inboundFlow->getFlow();
        $connection = $this->_resourceConnection->getConnection();
        $table = $this->inboundFlowFactory->create()->getResource()->getMainTable();
        $channelConfig = $this->getChannel($inboundFlow->getChannel())->getChannelConfig();

        if (!$channelConfig->allowConcurrency()) {
            $active = (int)$connection->fetchOne(
                $connection->select()
                    ->from($table, [new Zend_Db_Expr('COUNT(*)')])
                    ->where('flow = ?', $flowType)
                    ->where(
                        'status IN (?)',
                        [Inboundflow::STATUS_PROCESSING, Inboundflow::STATUS_PARSING, Inboundflow::STATUS_RECEIVING]
                    )
            );

            if ($active > 0) {
                return false;
            }
        }

        if (!array_key_exists($flowType, self::$_waitingQueue)) {
            return true;
        }

        $conditions = self::$_waitingQueue[$flowType];

        $sql = $connection->select()->from($table);
        $params = [];

        foreach ($conditions as $condition) {
            $sql->where($condition['condition']);
            foreach ($condition['arguments'] as $key => $param) {
                if (is_array($param)) {
                    $p = [];

                    foreach ($param as $v) {
                        $p[] = $connection->quote($v);
                    }
                    $params[$key] = implode(',', $p);
                } else {
                    $params[$key] = $param;
                }
            }
        }

        $results = $connection->fetchAll($sql, $params);

        return !(count($results) !== 0);
    }

    /**
     * @param Inboundflow $flow
     * @return Inboundflow|void
     * @throws Exception
     */
    public function processVideos(Inboundflow $flow)
    {
        if ($flow->getStatus() === Inboundflow::STATUS_ERROR) {
            return;
        }

        /** @var Images $model */
        $model = $this->objectManager->create(Videos::class);

        $model->loadConfig($flow->getChannel());
        $result = $model->insertData($flow);

        if ($result) {
            $flow->setStatus(Inboundflow::STATUS_PROCESSED);
            $flow->save();
            return;
        }

        $total = $model->getVideoToProcessCount();
        $errors = $model->getProcessedVideoErrorCount();

        $retry = $flow->getRetry();
        $log = $flow->getLog() . "Process #{$retry}: Video to process {$total} ({$errors} fault)\n";

        $flow->setRetry($retry + 1);

        if ($retry >= self::MAX_VIDEO_PROCESS_RETRY_COUNT) {
            $flow->setLog(
                "Unable to complete the flow after {$retry} tries. Look at flow videos table for more details.\n\n" . $log
            );
            $flow->setStatus(Inboundflow::STATUS_ERROR);
        } else {
            $flow->setLog($log);
            $flow->setStatus(Inboundflow::STATUS_PARSED);
        }

        return $flow->save();
    }

    /**
     * @param Inboundflow $flow
     * @return Inboundflow|void
     * @throws Exception
     */
    public function processImages(Inboundflow $flow)
    {
        if ($flow->getStatus() === Inboundflow::STATUS_ERROR) {
            return;
        }

        $channelName = $flow->getChannel();

        /** @var Images $channel */
        $channel = $this->getChannel($channelName);
        $result = $channel->insertData($flow);

        if ($result) {
            $flow->setStatus(Inboundflow::STATUS_PROCESSED);
            $flow->save();
            return;
        }

        $total = $channel->getImageToProcessCount();
        $errors = $channel->getProcessedImageErrorCount();

        $retry = $flow->getRetry();
        $log = $flow->getLog() . "Process #{$retry}: Image to process $total ($errors fault)\n";

        $flow->setRetry($retry + 1);

        if ($retry >= self::MAX_IMAGE_PROCESS_RETRY_COUNT) {
            $flow->setLog(
                "Unable to complete the flow after $retry tries. Look at flow images table for more details.\n\n" . $log
            );
            $flow->setStatus(Inboundflow::STATUS_ERROR);
        } else {
            $flow->setLog($log);
            $flow->setStatus(Inboundflow::STATUS_PARSED);
        }

        return $flow->save();
    }

    /**
     * @param Inboundflow $flow
     * @return bool
     * @throws \Exception
     */
    public function processFile(Inboundflow $flow)
    {
        $flowModuleName = $this->_helperData->getFlowModuleName($this->_config, $flow->getChannel());
        $model = $this->objectManager->create('FiloBlu\\' . $flowModuleName . '\Model\\' . $flow->getFlow());

        if (!$model) {
            return false;
        }
        $model->setChannel($this->getChannel($flow->getChannel()));
        $model->processFileRows($flow);
        $flow->setComment($model->getErrorMessage());

        return true;
    }

    /**
     * @param Inboundflow $flow
     * @return Inboundflow|void
     * @throws Exception
     */
    public function processDocuments(Inboundflow $flow)
    {
        if ($flow->getStatus() === Inboundflow::STATUS_ERROR) {
            return;
        }

        $channel = $this->getChannel($flow->getChannel());

        /** @var Document $model */
        $model = $this->doc;
        $model->setChannelData($channel);

        $result = $model->insertData($flow);

        if ($result) {
            $flow->setStatus(Inboundflow::STATUS_PROCESSED);
            $flow->save();
            return;
        }

        $total = $model->getDocumentToProcessCount();
        $errors = $model->getProcessedDocumentErrorCount();

        $retry = $flow->getRetry();
        $log = $flow->getLog() . "Process #$retry: Documents to process $total ($errors fault)\n";

        $flow->setRetry($retry + 1);

        if ($retry >= self::MAX_DOCUMENT_PROCESS_RETRY_COUNT) {
            $flow->setLog(
                "Unable to complete the flow after {$retry} tries. Look at flow documents table for more details.\n\n" . $log
            );
            $flow->setStatus(Inboundflow::STATUS_ERROR);
        } else {
            $flow->setLog($log);
            $flow->setStatus(Inboundflow::STATUS_PARSED);
        }

        return $flow->save();
    }

    /**
     * @param Inboundflow $flow
     * @return void
     * @throws Exception
     */
    public function processSwatches(Inboundflow $flow)
    {
        if ($flow->getStatus() === Inboundflow::STATUS_ERROR) {
            return;
        }

        /** @var Swatch $model */
        $model = $this->objectManager->create(Swatch::class);

        $result = $model->insertData($flow);

        if ($result) {
            $flow->setStatus(Inboundflow::STATUS_PROCESSED);
            $flow->save();
        }
    }

    /**
     * @throws Exception
     */
    public function clean()
    {
        if (!($days = (int)$this->_scopeConfig->getValue(
            'filoblu_flow/flow_clean/flow_clean_days',
            ScopeInterface::SCOPE_STORE
        ))) {
            return;
        }

        $this->init();

        $collection = $this->inboundFlowCollectionFactory->create();
        $connection = $collection->getConnection();
        $table = $collection->getMainTable();

        /** @var Inboundflow [] $flows */
        $flows = $collection->addFieldToFilter(
            'status',
            [
                Inboundflow::STATUS_PROCESSED,
                Inboundflow::STATUS_ERROR
            ]
        )
            ->addFieldToFilter(
                'receiving_time',
                ['lt' => new Zend_Db_Expr("DATE_SUB(CURDATE(), INTERVAL $days DAY)")]
            )->getItems();

        foreach ($flows as $flow) {
            $type = $flow->getType();
            $channel = $flow->getChannel();

            try {
                switch ($type) {
                    case self::CHANNEL_TYPE_IMAGE_CODE:
                        break;

                    case self::CHANNEL_TYPE_FILE_CODE:
                        $result = $connection->fetchRow(
                            "SELECT MAX(`id`) AS `id` FROM `{$table}` WHERE `channel` = ? ",
                            [$channel],
                            Zend_Db::FETCH_OBJ
                        );

                        if ($result->id !== $flow->getId()) {
                            $this->cleanFile($flow);
                        }
                        break;

                    default:
                        throw new Exception("Unknown channel type {$type}");
                }
            } catch (Exception $e) {
                $flow->setLog($e->getMessage());
                $flow->save();
            }
        }
    }

    /**
     * @param Inboundflow $flow
     * @return bool
     * @throws Exception
     */
    public function cleanFile(Inboundflow $flow)
    {
        $model = $this->objectManager->create('FiloBlu\Flow\Model\\' . $flow->getFlow());

        if (!$model) {
            return false;
        }

        try {
            $model->clean($flow);
            $flow->setStatus(Inboundflow::STATUS_CLEANED)->save();
            return true;
        } catch (Exception $e) {
            $this->_logger->info($e->getMessage());
            $flow->setLog($e->getMessage());
            $flow->save();
            return false;
        }
    }
}
