chill-bundles/src/Bundle/ChillMainBundle/Service/Import/GeographicalUnitBaseImporter.php

<?php

declare(strict_types=1);

/*
 * Chill is a software for social workers
 *
 * For the full copyright and license information, please view
 * the LICENSE file that was distributed with this source code.
 */

namespace Chill\MainBundle\Service\Import;

use Doctrine\DBAL\Connection;
use Doctrine\DBAL\Statement;
use Doctrine\DBAL\Types\Types;
use Exception;
use Psr\Log\LoggerInterface;
use RuntimeException;
use function array_key_exists;
use function count;

final class GeographicalUnitBaseImporter
{
    private const INSERT = <<<'SQL'
        INSERT INTO geographical_unit_temp
            (layerKey, layerName, unitName, unitKey, geom)
        SELECT
            i.layerKey, i.layerName, i.unitName, i.unitKey,
            ST_Transform(ST_setSrid(ST_GeomFromText(i.wkt), i.srid), 4326)
        FROM
            (VALUES
                {{ values }}
             ) AS i (layerKey, layerName, unitName, unitKey, wkt, srid)
        SQL;

    private const LOG_PREFIX = '[GeographicalUnitBAseImporter] ';

    private const VALUE = '(?, ?::jsonb, ?, ?, ?, ?::int)';

    /**
     * @var array<int, Statement>
     */
    private array $cachingStatements = [];

    private bool $isInitialized = false;

    private array $waitingForInsert = [];

    public function __construct(private readonly Connection $defaultConnection, private readonly LoggerInterface $logger) {}

    public function finalize(): void
    {
        $this->doInsertPending();

        $this->prepareForFinalize();

        $this->updateGeographicalUnitTable();

        $this->deleteTemporaryTable();

        $this->isInitialized = false;
    }

    public function importUnit(
        string $layerKey,
        array $layerName,
        string $unitName,
        string $unitKey,
        string $geomAsWKT,
        ?int $srid = null
    ): void {
        $this->initialize();

        $this->waitingForInsert[] = [
            'layerKey' => $layerKey,
            'layerName' => $layerName,
            'unitName' => $unitName,
            'unitKey' => $unitKey,
            'geomAsWKT' => $geomAsWKT,
            'srid' => $srid,
        ];

        if (100 <= count($this->waitingForInsert)) {
            $this->doInsertPending();
        }
    }

    private function createTemporaryTable(): void
    {
        $this->defaultConnection->executeStatement("CREATE TEMPORARY TABLE geographical_unit_temp (
            layerKey TEXT DEFAULT '' NOT NULL,
            layerName JSONB DEFAULT '[]'::jsonb NOT NULL,
            unitName TEXT default '' NOT NULL,
            unitKey TEXT default '' NOT NULL,
            geom GEOMETRY(MULTIPOLYGON, 4326)
        )");

        $this->defaultConnection->executeStatement('SET work_mem TO \'50MB\'');
    }

    private function deleteTemporaryTable(): void
    {
        $this->defaultConnection->executeStatement('DROP TABLE IF EXISTS geographical_unit_temp');
    }

    private function doInsertPending(): void
    {
        $forNumber = count($this->waitingForInsert);

        if (0 === $forNumber) {
            return;
        }

        if (!array_key_exists($forNumber, $this->cachingStatements)) {
            $sql = strtr(self::INSERT, [
                '{{ values }}' => implode(
                    ', ',
                    array_fill(0, $forNumber, self::VALUE)
                ),
            ]);

            $this->logger->debug(self::LOG_PREFIX . ' generated sql for insert', [
                'sql' => $sql,
                'forNumber' => $forNumber,
            ]);

            $this->cachingStatements[$forNumber] = $this->defaultConnection->prepare($sql);
        }

        $statement = $this->cachingStatements[$forNumber];

        try {
            $i = 0;

            foreach ($this->waitingForInsert as $insert) {
                $statement->bindValue(++$i, $insert['layerKey'], Types::STRING);
                $statement->bindValue(++$i, $insert['layerName'], Types::JSON);
                $statement->bindValue(++$i, $insert['unitName'], Types::STRING);
                $statement->bindValue(++$i, $insert['unitKey'], Types::STRING);
                $statement->bindValue(++$i, $insert['geomAsWKT'], Types::STRING);
                $statement->bindValue(++$i, $insert['srid'], Types::INTEGER);
            }

            $affected = $statement->executeStatement();

            if (0 === $affected) {
                throw new RuntimeException('no row affected');
            }
        } catch (Exception $e) {
            throw $e;
        } finally {
            $this->waitingForInsert = [];
        }
    }

    private function initialize(): void
    {
        if ($this->isInitialized) {
            return;
        }

        $this->deleteTemporaryTable();
        $this->createTemporaryTable();
        $this->isInitialized = true;
    }

    private function prepareForFinalize(): void
    {
        $this->defaultConnection->executeStatement(
            'CREATE INDEX idx_ref_add_temp ON geographical_unit_temp (unitKey)'
        );
    }

    private function updateGeographicalUnitTable(): void
    {
        $this->defaultConnection->transactional(
            function () {
                // 0) create new layers
                $this->defaultConnection->executeStatement(
                    "
                    WITH unique_layers AS (
                        SELECT DISTINCT layerKey, layerName FROM geographical_unit_temp
                    )
                    INSERT INTO chill_main_geographical_unit_layer (id, name, refid)
                        SELECT
                            nextval('chill_main_geographical_unit_layer_id_seq'),
                            layerName,
                            layerKey
                        FROM unique_layers
                    ON CONFLICT (refid)
                    DO UPDATE SET name=EXCLUDED.name
                "
                );

                //1) Add new units
                $this->logger->info(self::LOG_PREFIX . 'upsert new units');
                $affected = $this->defaultConnection->executeStatement("INSERT INTO chill_main_geographical_unit
            (id, geom, unitname, layer_id, unitrefid)
            SELECT
                nextval('chill_main_geographical_unit_id_seq'),
                st_makevalid(geom),
                unitName,
                layer.id,
                unitKey
            FROM geographical_unit_temp JOIN chill_main_geographical_unit_layer AS layer ON layer.refid = layerKey
            ON CONFLICT (layer_id, unitrefid)
            DO UPDATE
            SET geom = EXCLUDED.geom, unitname = EXCLUDED.unitname
        ");
                $this->logger->info(self::LOG_PREFIX . 'units upserted', ['upserted' => $affected]);

                //3) Delete units
                $this->logger->info(self::LOG_PREFIX . 'soft delete adresses');
                $affected = $this->defaultConnection->executeStatement('WITH to_delete AS (
                    SELECT cmgu.id
                    FROM chill_main_geographical_unit AS cmgu
                    JOIN chill_main_geographical_unit_layer AS cmgul ON cmgul.id = cmgu.layer_id
                    JOIN geographical_unit_temp AS gut ON cmgul.refid = gut.layerKey AND cmgu.unitrefid = gut.unitKey
                )
                DELETE FROM chill_main_geographical_unit
                WHERE id NOT IN (SELECT id FROM to_delete)
                ');
                $this->logger->info(self::LOG_PREFIX . 'addresses deleted', ['deleted' => $affected]);
            }
        );
    }
}