Merge branch '112-addresses-recollate' into 'master'

Feature: re-associate addresses with addresses references and postal code references in a cronjob and allow a cronjob to pass data from one execution to the next one

Closes #112

See merge request Chill-Projet/chill-bundles!580
This commit is contained in:
Julien Fastré 2023-07-14 08:56:56 +00:00
commit 718de2fad0
15 changed files with 520 additions and 12 deletions

View File

@ -0,0 +1,6 @@
kind: DX
body: '[cronjob] when a cronjob is executed, it may return an array of data that will
be passed as argument on the next execution'
time: 2023-07-12T11:36:03.813179067+02:00
custom:
Issue: ""

View File

@ -0,0 +1,6 @@
kind: Feature
body: |
[addresses] Add a cronjob to re-associate addresses with addresses reference every 6 hours
time: 2023-07-12T18:00:23.037677413+02:00
custom:
Issue: "112"

View File

@ -19,5 +19,13 @@ interface CronJobInterface
public function getKey(): string;
public function run(): void;
/**
* Execute the cronjob
*
* If data is returned, this data is passed as argument on the next execution
*
* @param array $lastExecutionData the data which was returned from the previous execution
* @return array|null optionally return an array with the same data than the previous execution
*/
public function run(array $lastExecutionData): null|array;
}

View File

@ -14,6 +14,7 @@ namespace Chill\MainBundle\Cron;
use Chill\MainBundle\Entity\CronJobExecution;
use Chill\MainBundle\Repository\CronJobExecutionRepositoryInterface;
use DateTimeImmutable;
use Doctrine\DBAL\Types\Types;
use Doctrine\ORM\EntityManagerInterface;
use Exception;
use Psr\Log\LoggerInterface;
@ -46,6 +47,8 @@ class CronManager implements CronManagerInterface
private const UPDATE_BEFORE_EXEC = 'UPDATE ' . CronJobExecution::class . ' cr SET cr.lastStart = :now WHERE cr.key = :key';
private const UPDATE_LAST_EXECUTION_DATA = 'UPDATE ' . CronJobExecution::class . ' cr SET cr.lastExecutionData = :data WHERE cr.key = :key';
private CronJobExecutionRepositoryInterface $cronJobExecutionRepository;
private EntityManagerInterface $entityManager;
@ -85,6 +88,9 @@ class CronManager implements CronManagerInterface
foreach ($orderedJobs as $job) {
if ($job->canRun($lasts[$job->getKey()] ?? null)) {
if (array_key_exists($job->getKey(), $lasts)) {
$executionData = $lasts[$job->getKey()]->getLastExecutionData();
$this->entityManager
->createQuery(self::UPDATE_BEFORE_EXEC)
->setParameters([
@ -96,12 +102,17 @@ class CronManager implements CronManagerInterface
$execution = new CronJobExecution($job->getKey());
$this->entityManager->persist($execution);
$this->entityManager->flush();
$executionData = $execution->getLastExecutionData();
}
$this->entityManager->clear();
// note: at this step, the entity manager does not have any entity CronJobExecution
// into his internal memory
try {
$this->logger->info(sprintf('%sWill run job', self::LOG_PREFIX), ['job' => $job->getKey()]);
$job->run();
$result = $job->run($executionData);
$this->entityManager
->createQuery(self::UPDATE_AFTER_EXEC)
@ -112,6 +123,14 @@ class CronManager implements CronManagerInterface
])
->execute();
if (null !== $result) {
$this->entityManager
->createQuery(self::UPDATE_LAST_EXECUTION_DATA)
->setParameter('data', $result, Types::JSON)
->setParameter('key', $job->getKey(), Types::STRING)
->execute();
}
$this->logger->info(sprintf('%sSuccessfully run job', self::LOG_PREFIX), ['job' => $job->getKey()]);
return;
@ -133,7 +152,7 @@ class CronManager implements CronManagerInterface
}
/**
* @return array<0: CronJobInterface[], 1: array<string, CronJobExecution>>
* @return array{0: array<CronJobInterface>, 1: array<string, CronJobExecution>}
*/
private function getOrderedJobs(): array
{
@ -174,7 +193,7 @@ class CronManager implements CronManagerInterface
{
foreach ($this->jobs as $job) {
if ($job->getKey() === $forceJob) {
$job->run();
$job->run([]);
}
}
}

View File

@ -31,7 +31,6 @@ class CronJobExecution
private string $key;
/**
* @var DateTimeImmutable
* @ORM\Column(type="datetime_immutable", nullable=true, options={"default": null})
*/
private ?DateTimeImmutable $lastEnd = null;
@ -46,6 +45,11 @@ class CronJobExecution
*/
private ?int $lastStatus = null;
/**
* @ORM\Column(type="json", options={"default": "'{}'::jsonb", "jsonb": true})
*/
private array $lastExecutionData = [];
public function __construct(string $key)
{
$this->key = $key;
@ -92,4 +96,16 @@ class CronJobExecution
return $this;
}
public function getLastExecutionData(): array
{
return $this->lastExecutionData;
}
public function setLastExecutionData(array $lastExecutionData): CronJobExecution
{
$this->lastExecutionData = $lastExecutionData;
return $this;
}
}

View File

@ -0,0 +1,145 @@
<?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\AddressGeographicalUnit;
use Doctrine\DBAL\Connection;
use Psr\Log\LoggerInterface;
final readonly class CollateAddressWithReferenceOrPostalCode implements CollateAddressWithReferenceOrPostalCodeInterface
{
private const LOG_PREFIX = '[collate addresses] ';
/**
* For the address having an "invented" postal code, find the postal code "reference" with the same code,
* and the most similar name. When two reference code match, we add
*
* This query intentionally includes also address with reference, as the reference may be wrong.
*/
private const FORCE_ORIGINAL_POSTAL_CODE = <<<'SQL'
WITH recollate AS (
SELECT * FROM (
SELECT cma.id AS address_id, cmpc.id, cmpc.label, cmpc.code, cmpc_reference.id AS cmpc_reference_id, cmpc_reference.label, cmpc_reference.code,
RANK() OVER (PARTITION BY cma.id ORDER BY SIMILARITY(cmpc.label, cmpc_reference.label) DESC, cmpc_reference.id ASC) AS ranked
FROM
chill_main_address cma JOIN chill_main_postal_code cmpc on cma.postcode_id = cmpc.id,
chill_main_postal_code cmpc_reference
WHERE
-- use only postal code which are reference
cmpc_reference.id != cmpc.id AND cmpc_reference.origin = 0
-- only where cmpc is created manually
AND cmpc.origin != 0
-- only when postal code match
AND TRIM(REPLACE(LOWER(cmpc.code), ' ', '')) = LOWER(cmpc_reference.code)
AND cmpc.country_id = cmpc_reference.country_id
AND cma.id > :since_id -- to set the first id
) sq
WHERE ranked = 1)
UPDATE chill_main_address SET postcode_id = cmpc_reference_id FROM recollate WHERE recollate.address_id = chill_main_address.id;
SQL;
/**
* associate the address with the most similar address reference.
*
* This query intentionally ignores the existing addressreference_id, to let fixing the address match the
* most similar address reference.
*/
private const FORCE_MOST_SIMILAR_ADDRESS_REFERENCE = <<<'SQL'
WITH recollate AS (
SELECT * FROM (
SELECT cma.id AS address_id, cma.streetnumber, cma.street, cmpc.code, cmpc.label, cmar.id AS address_reference_id, cmar.streetnumber, cmar.street, cmpc_reference.code, cmpc_reference.label,
similarity(cma.street, cmar.street),
RANK() OVER (PARTITION BY cma.id ORDER BY SIMILARITY (cma.street, cmar.street) DESC, SIMILARITY (cma.streetnumber, cmar.streetnumber), cmar.id ASC) AS ranked
FROM
chill_main_address cma
JOIN chill_main_postal_code cmpc on cma.postcode_id = cmpc.id,
chill_main_address_reference cmar JOIN chill_main_postal_code cmpc_reference ON cmar.postcode_id = cmpc_reference.id
WHERE
cma.addressreference_id != cmar.id
-- only if cmpc is a reference (must be matched before executing this query)
AND cma.postcode_id = cmar.postcode_id
-- join cmpc to cma
AND SIMILARITY(LOWER(cma.street), LOWER(cmar.street)) > 0.6 AND LOWER(cma.streetnumber) = LOWER(cmar.streetnumber)
-- only addresses which match the address reference - let the user decide if the reference has changed
AND cma.refstatus = 'match'
-- only the most recent
AND cma.id > :since_id
) AS sq
WHERE ranked = 1
)
UPDATE chill_main_address SET addressreference_id = recollate.address_reference_id FROM recollate WHERE chill_main_address.id = recollate.address_id;
SQL;
private const UPDATE_POINT = <<<'SQL'
WITH address_geom AS (
SELECT cma.id AS address_id, COALESCE(cmar.point, cmpc.center) AS point
FROM chill_main_address cma
LEFT JOIN chill_main_address_reference cmar ON cma.addressreference_id = cmar.id
LEFT JOIN chill_main_postal_code cmpc ON cma.postcode_id = cmpc.id
WHERE cma.id > :since_id
)
UPDATE chill_main_address SET point = address_geom.point FROM address_geom WHERE address_geom.address_id = chill_main_address.id
SQL;
private const MAX_ADDRESS_ID = <<<'SQL'
SELECT MAX(id) AS max_id FROM chill_main_address;
SQL;
public function __construct(
private Connection $connection,
private LoggerInterface $logger,
) {
}
/**
* @throws \Throwable
*/
public function __invoke(int $sinceId = 0): int
{
try {
[
$postCodeSetReferenceFromMostSimilar,
$addressReferenceMatch,
$pointUpdates,
$lastId,
] = $this->connection->transactional(function () use ($sinceId) {
$postCodeSetReferenceFromMostSimilar = $this->connection->executeStatement(self::FORCE_ORIGINAL_POSTAL_CODE, ['since_id' => $sinceId]);
$addressReferenceMatch = $this->connection->executeStatement(self::FORCE_MOST_SIMILAR_ADDRESS_REFERENCE, ['since_id' => $sinceId]);
$pointUpdates = $this->connection->executeStatement(self::UPDATE_POINT, ['since_id' => $sinceId]);
$lastId = $this->connection->fetchOne(self::MAX_ADDRESS_ID);
return [
$postCodeSetReferenceFromMostSimilar,
$addressReferenceMatch,
$pointUpdates,
$lastId,
];
});
} catch (\Throwable $e) {
$this->logger->error(self::LOG_PREFIX . "error while re-collating addresses", [
'message' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
throw $e;
}
$this->logger->info(self::LOG_PREFIX . "Collate the addresses with reference", [
'set_postcode_from_most_similar' => $postCodeSetReferenceFromMostSimilar,
'address_reference_match' => $addressReferenceMatch,
'point_update' => $pointUpdates,
'since_id' => $sinceId,
'last_id' => $lastId,
]);
return $lastId;
}
}

View File

@ -0,0 +1,50 @@
<?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\AddressGeographicalUnit;
use Chill\MainBundle\Cron\CronJobInterface;
use Chill\MainBundle\Entity\CronJobExecution;
use Symfony\Component\Clock\ClockInterface;
final readonly class CollateAddressWithReferenceOrPostalCodeCronJob implements CronJobInterface
{
private const LAST_MAX_ID = 'last-max-id';
public function __construct(
private ClockInterface $clock,
private CollateAddressWithReferenceOrPostalCodeInterface $collateAddressWithReferenceOrPostalCode,
) {
}
public function canRun(?CronJobExecution $cronJobExecution): bool
{
if (null === $cronJobExecution) {
return true;
}
$now = $this->clock->now();
return $now->sub(new \DateInterval('PT6H')) > $cronJobExecution->getLastStart();
}
public function getKey(): string
{
return 'collate-address';
}
public function run(array $lastExecutionData): null|array
{
$maxId = ($this->collateAddressWithReferenceOrPostalCode)($lastExecutionData[self::LAST_MAX_ID] ?? 0);
return [self::LAST_MAX_ID => $maxId];
}
}

View File

@ -0,0 +1,20 @@
<?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\AddressGeographicalUnit;
interface CollateAddressWithReferenceOrPostalCodeInterface
{
/**
* @throws \Throwable
*/
public function __invoke(int $sinceId = 0): int;
}

View File

@ -49,8 +49,10 @@ class RefreshAddressToGeographicalUnitMaterializedViewCronJob implements CronJob
return 'refresh-materialized-view-address-to-geog-units';
}
public function run(): void
public function run(array $lastExecutionData): null|array
{
$this->connection->executeQuery('REFRESH MATERIALIZED VIEW view_chill_main_address_geographical_unit');
return null;
}
}

View File

@ -0,0 +1,87 @@
<?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 Cron;
use Chill\MainBundle\Cron\CronJobInterface;
use Chill\MainBundle\Cron\CronManager;
use Chill\MainBundle\Entity\CronJobExecution;
use Chill\MainBundle\Repository\CronJobExecutionRepository;
use Doctrine\ORM\EntityManagerInterface;
use Prophecy\Argument;
use Prophecy\PhpUnit\ProphecyTrait;
use Psr\Log\NullLogger;
use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
/**
* @internal
* @coversNothing
*/
class CronJobDatabaseInteractionTest extends KernelTestCase
{
use ProphecyTrait;
private EntityManagerInterface $entityManager;
private CronJobExecutionRepository $cronJobExecutionRepository;
protected function setUp(): void
{
self::bootKernel();
$this->entityManager = self::$container->get(EntityManagerInterface::class);
$this->cronJobExecutionRepository = self::$container->get(CronJobExecutionRepository::class);
}
public function testCompleteLifeCycle(): void
{
$cronjob = $this->prophesize(CronJobInterface::class);
$cronjob->canRun(null)->willReturn(true);
$cronjob->canRun(Argument::type(CronJobExecution::class))->willReturn(true);
$cronjob->getKey()->willReturn('test-with-data');
$cronjob->run([])->willReturn(['test' => 'execution-0']);
$cronjob->run(['test' => 'execution-0'])->willReturn(['test' => 'execution-1']);
$cronjob->run([])->shouldBeCalledOnce();
$cronjob->run(['test' => 'execution-0'])->shouldBeCalledOnce();
$manager = new CronManager(
$this->cronJobExecutionRepository,
$this->entityManager,
[$cronjob->reveal()],
new NullLogger()
);
// run a first time
$manager->run();
// run a second time
$manager->run();
}
}
class JobWithReturn implements CronJobInterface
{
public function canRun(?CronJobExecution $cronJobExecution): bool
{
return true;
}
public function getKey(): string
{
return 'with-data';
}
public function run(array $lastExecutionData): null|array
{
return ['data' => 'test'];
}
}

View File

@ -40,7 +40,7 @@ final class CronManagerTest extends TestCase
$jobToExecute = $this->prophesize(CronJobInterface::class);
$jobToExecute->getKey()->willReturn('to-exec');
$jobToExecute->canRun(Argument::type(CronJobExecution::class))->willReturn(true);
$jobToExecute->run()->shouldBeCalled();
$jobToExecute->run([])->shouldBeCalled();
$executions = [
['key' => $jobOld1->getKey(), 'lastStart' => new DateTimeImmutable('yesterday'), 'lastEnd' => new DateTimeImmutable('1 hours ago'), 'lastStatus' => CronJobExecution::SUCCESS],
@ -64,7 +64,7 @@ final class CronManagerTest extends TestCase
$jobAlreadyExecuted = new JobCanRun('k');
$jobNeverExecuted = $this->prophesize(CronJobInterface::class);
$jobNeverExecuted->getKey()->willReturn('never-executed');
$jobNeverExecuted->run()->shouldBeCalled();
$jobNeverExecuted->run([])->shouldBeCalled();
$jobNeverExecuted->canRun(null)->willReturn(true);
$executions = [
@ -86,7 +86,7 @@ final class CronManagerTest extends TestCase
$jobAlreadyExecuted = new JobCanRun('k');
$jobNeverExecuted = $this->prophesize(CronJobInterface::class);
$jobNeverExecuted->getKey()->willReturn('never-executed');
$jobNeverExecuted->run()->shouldBeCalled();
$jobNeverExecuted->run([])->shouldBeCalled();
$jobNeverExecuted->canRun(null)->willReturn(true);
$executions = [
@ -178,8 +178,9 @@ class JobCanRun implements CronJobInterface
return $this->key;
}
public function run(): void
public function run(array $lastExecutionData): null|array
{
return null;
}
}
@ -195,7 +196,8 @@ class JobCannotRun implements CronJobInterface
return 'job-b';
}
public function run(): void
public function run(array $lastExecutionData): null|array
{
return null;
}
}

View File

@ -0,0 +1,68 @@
<?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 Services\AddressGeographicalUnit;
use Chill\MainBundle\Entity\CronJobExecution;
use Chill\MainBundle\Service\AddressGeographicalUnit\CollateAddressWithReferenceOrPostalCodeCronJob;
use Chill\MainBundle\Service\AddressGeographicalUnit\CollateAddressWithReferenceOrPostalCodeInterface;
use PHPUnit\Framework\TestCase;
use Prophecy\PhpUnit\ProphecyTrait;
use Symfony\Component\Clock\MockClock;
/**
* @internal
* @coversNothing
*/
class CollateAddressWithReferenceOrPostalCodeCronJobTest extends TestCase
{
use ProphecyTrait;
/**
* @dataProvider provideDataCanRun
*/
public function testCanRun(\DateTimeImmutable $now, ?\DateTimeImmutable $lastExecution, bool $expected): void
{
$execution = match ($lastExecution) {
null => null,
default => (new CronJobExecution('collate-address'))->setLastStart($lastExecution),
};
$clock = new MockClock($now);
$collator = $this->prophesize(CollateAddressWithReferenceOrPostalCodeInterface::class);
$job = new CollateAddressWithReferenceOrPostalCodeCronJob($clock, $collator->reveal());
self::assertEquals($expected, $job->canRun($execution));
}
public function testRun(): void
{
$clock = new MockClock();
$collator = $this->prophesize(CollateAddressWithReferenceOrPostalCodeInterface::class);
$collator->__invoke(0)->shouldBeCalledOnce();
$collator->__invoke(0)->willReturn(1);
$job = new CollateAddressWithReferenceOrPostalCodeCronJob($clock, $collator->reveal());
$actual = $job->run(['last-max-id' => 0]);
self::assertEquals(['last-max-id' => 1], $actual);
}
public static function provideDataCanRun(): iterable
{
yield [new \DateTimeImmutable('2023-07-10T12:00:00'), new \DateTimeImmutable('2023-07-10T11:00:00'), false];
yield [new \DateTimeImmutable('2023-07-10T12:00:00'), new \DateTimeImmutable('2023-07-10T05:00:00'), true];
yield [new \DateTimeImmutable('2023-07-10T12:00:00'), new \DateTimeImmutable('2023-07-01T12:00:00'), true];
yield [new \DateTimeImmutable('2023-07-10T12:00:00'), null, true];
}
}

View File

@ -0,0 +1,44 @@
<?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 Services\AddressGeographicalUnit;
use Chill\MainBundle\Service\AddressGeographicalUnit\CollateAddressWithReferenceOrPostalCode;
use Doctrine\DBAL\Connection;
use Psr\Log\NullLogger;
use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
/**
* @internal
* @coversNothing
*/
class CollateAddressWithReferenceOrPostalCodeTest extends KernelTestCase
{
private Connection $connection;
protected function setUp(): void
{
self::bootKernel();
$this->connection = self::$container->get(Connection::class);
}
public function testRun(): void
{
$collator = new CollateAddressWithReferenceOrPostalCode(
$this->connection,
new NullLogger()
);
$result = $collator(0);
self::assertGreaterThan(0, $result);
}
}

View File

@ -0,0 +1,33 @@
<?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\Migrations\Main;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\Migrations\AbstractMigration;
final class Version20230711152947 extends AbstractMigration
{
public function getDescription(): string
{
return 'Add data to ';
}
public function up(Schema $schema): void
{
$this->addSql('ALTER TABLE chill_main_cronjob_execution ADD lastExecutionData JSONB DEFAULT \'{}\'::jsonb NOT NULL');
}
public function down(Schema $schema): void
{
$this->addSql('ALTER TABLE chill_main_cronjob_execution DROP COLUMN lastExecutionData');
}
}

View File

@ -39,8 +39,10 @@ readonly class AccompanyingPeriodStepChangeCronjob implements CronJobInterface
return 'accompanying-period-step-change';
}
public function run(): void
public function run(array $lastExecutionData): null|array
{
($this->requestor)();
return null;
}
}