create command to create allocation instances from flux

src/Command/CreateAllocationsFromExistingFlux.php

+<?php
+
+declare(strict_types=1);
+
+namespace App\Command;
+
+use App\Entity\Adherent;
+use App\Entity\Flux;
+use App\Entity\Allocation;
+use App\Utils\CustomEntityManager;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
+use Symfony\Component\Console\Input\InputOption;
+
+/**
+ * This command is used to create Allocation instances for existing Flux,
+ * in order to be used in an external context
+ */
+class CreateAllocationsFromExistingFlux extends Command
+{
+    protected static $defaultName = 'kohinos:ssa:create-allocations-from-existing-flux';
+
+    protected $em;
+    protected $io;
+    protected $param;
+
+    public function __construct(CustomEntityManager $em) {
+        $this->em = $em;
+        parent::__construct();
+    }
+
+    protected function configure()
+    {
+        $this
+            ->setDescription('SSA : créer les instances d\'allocations à partir des flux de cotisation existantes')
+            ->addOption(
+                'batchamount',
+                null,
+                InputOption::VALUE_REQUIRED,
+                "Nombre de cotisations à traiter lors de cette opération"
+            );
+        ;
+    }
+
+    private function readCsv() {
+
+    }
+
+    /**
+     * @param InputInterface  $input
+     * @param OutputInterface $output
+     *
+     * @return int
+     */
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $this->io = new SymfonyStyle($input, $output);
+
+        $this->io->title('START - creating allocations');
+
+        // get batch amount param
+        $batchAmount = $input->hasOption('batchamount') ? $input->getOption('batchamount') : null;
+
+        // get cotisations that don't have an allocation set
+        $fluxRepository = $this->em->getRepository(Flux::class);
+        $cotisations = $fluxRepository->getTavCotisationsWithoutAllocation($batchAmount);
+        
+        $count = 0;
+        foreach ($cotisations as $key => $cotis) {
+            // $this->io->success(json_encode($cotis));
+
+            // create allocation
+            $allocation = new Allocation();
+            $allocation->setCotisationFlux($fluxRepository->find($cotis["id"]));
+
+            // set adherent
+            if ($cotis["adherent_id"] !== null) {
+                $adherent_id = $cotis["adherent_id"];
+            } else {
+                $adherent_id = $cotis["adherent_dest_id"];
+            }
+            $allocation->setAdherent($this->em->getRepository(Adherent::class)->find($adherent_id));            
+
+            // set montant & complementary flux if exists
+            $complementaryFlux = $fluxRepository->getRelatedTavAllocationFlux($cotis["id"]);
+            $amount = floatval($cotis["montant"]);
+            if ($complementaryFlux["id"] !== null) {
+                $allocation->setComplementaryFlux($fluxRepository->find($complementaryFlux["id"]));
+
+                if ($complementaryFlux["type"] == "reversement_cotisation_adherent") {
+                    $amount += floatval($complementaryFlux["montant"]);
+                } else if ($complementaryFlux["type"] == "prelevement_cotisation_adherent") {
+                    $amount -= floatval($complementaryFlux["montant"]);
+                } 
+            }
+
+            $allocation->setAmount($amount);
+            $this->em->persist($allocation);
+
+            $count += 1;
+        }
+
+        $this->em->flush();
+
+        $this->io->success("End, created {$count} allocations");
+
+        $memoryUsage = memory_get_usage(true) / 1024 / 1024;
+        $this->io->text("Batch finished with memory: ${memoryUsage}M");
+
+        return 0;
+    }
+}

src/Repository/FluxRepository.php

@@ -452,4 +452,58 @@ class FluxRepository extends ServiceEntityRepository
 
         return $results;
     }
+
+    /**
+     * @param int $limit, amount to fetch
+     * 
+     * @return Array list of cotisations
+     */
+    public function getTavCotisationsWithoutAllocation($limit)
+    {
+        $sqlQuery = "SELECT f.id, f.montant, f.adherent_id, f.adherent_dest_id
+            FROM {$this->tableName} f
+            LEFT JOIN allocation a ON a.cotisation_flux_id = f.id
+            WHERE f.type IN ('achat_monnaie_adherent', 'vente_emlc_adherent')
+            AND a.id IS NULL
+            ORDER BY f.created_at DESC";
+
+        if (null !== $limit && 0 !== intval($limit)) {
+            $sqlQuery .= " LIMIT " . intval($limit); 
+        }
+        
+        $statement = $this->connection->prepare($sqlQuery);
+        $statement->execute();
+        $results = $statement->fetchAll();
+
+        return $results;
+    }
+
+    /**
+     * Given a TAV cotisation flux, get the potential complementary flux that completes the full TAV allocation.
+     * @see App\Entity\Allocation 
+     * 
+     * @param int $fluxCotisId, cotisation flux id
+     */
+    public function getRelatedTavAllocationFlux($fluxCotisId) {
+        $sqlQuery = "SELECT fb.id, fb.type, fb.montant
+            FROM {$this->tableName} f
+            LEFT JOIN flux fb
+                ON ABS(TIMESTAMPDIFF(MICROSECOND, f.created_at, fb.created_at)) < 2000000
+                AND 
+                    (
+                        f.adherent_id = fb.adherent_id
+                        OR f.adherent_id = fb.adherent_dest_id
+                        OR f.adherent_dest_id = fb.adherent_id
+                        OR f.adherent_dest_id = fb.adherent_dest_id
+                    )
+                AND fb.type IN ('prelevement_cotisation_adherent', 'reversement_cotisation_adherent')
+            WHERE f.id = :fluxId";
+
+        $statement = $this->connection->prepare($sqlQuery);
+        $statement->bindValue(':fluxId', $fluxCotisId);
+        $statement->execute();
+        $results = $statement->fetchAll();
+
+        return $results[0];
+    }
 }