chill-bundles/src/Bundle/ChillMainBundle/Service/Import/PostalCodeBaseImporter.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;

/**
 * Optimized way to load postal code into database.
 */
class PostalCodeBaseImporter
{
    private const CREATE_TEMP_TABLE = <<<'SQL'
        CREATE TEMPORARY TABLE chill_main_postal_code_temp (
            countrycode VARCHAR(10),
            label VARCHAR(255),
            code VARCHAR(100),
            refpostalcodeid VARCHAR(255),
            postalcodeSource VARCHAR(255),
            lon FLOAT,
            lat FLOAT,
            srid INT
        )
        SQL;

    private const INSERT_TEMP = <<<'SQL'
        INSERT INTO chill_main_postal_code_temp
            (countrycode, label, code, refpostalcodeid, postalcodeSource, lon, lat, srid)
        VALUES
            {{ values }}
        SQL;

    private const UPSERT = <<<'SQL'
        WITH g AS (
            SELECT DISTINCT
                country.id AS country_id,
                temp.*
            FROM chill_main_postal_code_temp temp
            JOIN country ON country.countrycode = temp.countrycode
        )
        INSERT INTO chill_main_postal_code (id, country_id, label, code, origin, refpostalcodeid, postalcodeSource, center, createdAt, updatedAt, deletedAt)
               SELECT
                   nextval('chill_main_postal_code_id_seq'),
                   g.country_id,
                   g.label,
                   g.code,
                   0,
                   g.refpostalcodeid,
                   g.postalcodeSource,
                   CASE WHEN (g.lon != 0.0 AND g.lat != 0.0) THEN ST_Transform(ST_setSrid(ST_point(g.lon, g.lat), g.srid), 4326) ELSE NULL END,
                   NOW(),
                   NOW(),
                   NULL
               FROM g
        ON CONFLICT (code, refpostalcodeid, postalcodeSource) WHERE refpostalcodeid IS NOT NULL DO UPDATE
            SET label = excluded.label,
                center = excluded.center,
                deletedAt = NULL,
                updatedAt = CASE WHEN NOT st_equals(excluded.center, chill_main_postal_code.center) OR excluded.label != chill_main_postal_code.label OR chill_main_postal_code.deletedAt IS NOT NULL THEN NOW() ELSE chill_main_postal_code.updatedAt END
        SQL;

    private const DELETE_MISSING = <<<'SQL'
        UPDATE chill_main_postal_code
        SET deletedAt = NOW(), updatedAt = NOW()
        WHERE postalcodeSource = ?
          AND deletedAt IS NULL
          AND NOT EXISTS (
              SELECT 1 FROM chill_main_postal_code_temp temp
              WHERE temp.code = chill_main_postal_code.code
                AND temp.refpostalcodeid = chill_main_postal_code.refpostalcodeid
                AND temp.postalcodeSource = chill_main_postal_code.postalcodeSource
          )
        SQL;

    private const VALUE = '(?, ?, ?, ?, ?, ?, ?, ?)';

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

    private array $waitingForInsert = [];

    private bool $isInitialized = false;

    private ?string $currentSource = null;

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

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

        if ($this->isInitialized && null !== $this->currentSource) {
            $this->defaultConnection->transactional(function (Connection $connection): void {
                $connection->executeStatement(self::UPSERT);
                $connection->executeStatement(self::DELETE_MISSING, [$this->currentSource]);
            });
            $this->deleteTemporaryTable();
        }

        $this->isInitialized = false;
        $this->currentSource = null;
    }

    public function importCode(
        string $countryCode,
        string $label,
        string $code,
        string $refPostalCodeId,
        string $refPostalCodeSource,
        float $centerLat,
        float $centerLon,
        int $centerSRID,
    ): void {
        if (!$this->isInitialized) {
            $this->initialize($refPostalCodeSource);
        }

        if ($this->currentSource !== $refPostalCodeSource) {
            throw new \LogicException('Cannot store postal codes from different sources during same import. Execute finalize to commit inserts before changing the source');
        }

        $this->waitingForInsert[] = [
            $countryCode,
            $label,
            $code,
            $refPostalCodeId,
            $refPostalCodeSource,
            $centerLon,
            $centerLat,
            $centerSRID,
        ];

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

    private function initialize(string $source): void
    {
        $this->currentSource = $source;
        $this->deleteTemporaryTable();
        $this->createTemporaryTable();
        $this->isInitialized = true;
    }

    private function createTemporaryTable(): void
    {
        $this->defaultConnection->executeStatement(self::CREATE_TEMP_TABLE);
    }

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

    private function doInsertPending(): void
    {
        if ([] == $this->waitingForInsert) {
            return;
        }

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

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

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

        try {
            $statement->executeStatement(array_merge(...$this->waitingForInsert));
        } catch (\Exception $e) {
            // in some case, we can add debug code here
            // dump($this->waitingForInsert);
            throw $e;
        } finally {
            $this->waitingForInsert = [];
        }
    }
}