368 lines
14 KiB
PHP
368 lines
14 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace App\ContentFeed;
|
||
|
||
use App\Chatbot\TripCatalog;
|
||
use App\ContentFeed\Exception\ContentFeedStoppedException;
|
||
use App\ContentFeed\Exception\ContentFeedUnavailableException;
|
||
use App\ContentFeed\Exception\ContentRequestFailedException;
|
||
use App\ContentFeed\Exception\PagePhaseEndedException;
|
||
use App\Entity\ContentDocument;
|
||
use App\Enum\ContentFetchState;
|
||
use App\Model\ContentFeed\ContentManifest;
|
||
use App\Model\ContentFeed\ContentSyncReport;
|
||
use App\Model\ContentFeed\ManifestItem;
|
||
use App\Repository\ContentDocumentRepository;
|
||
use App\Repository\ContentSyncStateRepository;
|
||
use Doctrine\ORM\EntityManagerInterface;
|
||
use Psr\Log\LoggerInterface;
|
||
use Symfony\Component\Clock\ClockInterface;
|
||
|
||
/**
|
||
* Mirrors the TYPO3 content feed into content_document.
|
||
*
|
||
* A run reads the manifest, removes what it no longer lists, refetches the records and
|
||
* pages whose `modified` changed, and re-merges detail pages into their records. A failed
|
||
* manifest changes nothing. Bulk page work belongs in the night window: outside it, a run
|
||
* fetches at most PAGES_OUTSIDE_WINDOW pages and leaves the rest for the next one.
|
||
*
|
||
* With `refresh`, items whose stored copy is older than REFRESH_AFTER_DAYS are refetched
|
||
* regardless of `modified`, oldest first: the backstop for the changes TYPO3 cannot see,
|
||
* such as a referenced content element changing on another page.
|
||
*/
|
||
class ContentSyncer
|
||
{
|
||
public const PAGES_OUTSIDE_WINDOW = 50;
|
||
public const REFRESH_AFTER_DAYS = 7;
|
||
|
||
private const NIGHT_WINDOW_TIMEZONE = 'Europe/Berlin';
|
||
private const NIGHT_WINDOW_FROM_HOUR = 1;
|
||
private const NIGHT_WINDOW_TO_HOUR = 5;
|
||
private const FLUSH_EVERY_PAGES = 20;
|
||
|
||
public function __construct(
|
||
private readonly ContentFeedClient $client,
|
||
private readonly ContentDocumentRepository $documentRepository,
|
||
private readonly ContentSyncStateRepository $stateRepository,
|
||
private readonly DocumentBodyBuilder $bodyBuilder,
|
||
private readonly HtmlToMarkdownConverter $converter,
|
||
private readonly TripCatalog $tripCatalog,
|
||
private readonly EntityManagerInterface $entityManager,
|
||
private readonly ClockInterface $clock,
|
||
private readonly LoggerInterface $logger,
|
||
) {
|
||
}
|
||
|
||
public function run(bool $refresh = false, bool $ignoreWindow = false): ContentSyncReport
|
||
{
|
||
$report = new ContentSyncReport();
|
||
$state = $this->stateRepository->get();
|
||
$this->entityManager->persist($state);
|
||
$now = $this->clock->now();
|
||
|
||
if (null !== $state->getPausedUntil() && $now < $state->getPausedUntil()) {
|
||
$report->stoppedBy = 'Feed switched off by TYPO3 until '.$state->getPausedUntil()->format('c');
|
||
|
||
return $report;
|
||
}
|
||
|
||
$state->setPausedUntil(null)->setLastRunAt($now);
|
||
$this->client->startRun();
|
||
|
||
try {
|
||
$manifest = $this->client->manifest();
|
||
$documents = $this->applyManifest($manifest, $report);
|
||
$state->setManifestGeneratedAt($manifest->generatedAt);
|
||
$this->entityManager->flush();
|
||
|
||
$staleBefore = $refresh ? $now->modify(sprintf('-%d days', self::REFRESH_AFTER_DAYS)) : null;
|
||
$this->syncRecords($documents, $staleBefore, $report);
|
||
$this->syncPages($documents, $staleBefore, $ignoreWindow || $this->inNightWindow($now), $report);
|
||
} catch (ContentFeedUnavailableException $e) {
|
||
$state->setPausedUntil($e->retryAfter);
|
||
$report->stoppedBy = $e->getMessage();
|
||
} catch (ContentFeedStoppedException $e) {
|
||
$report->stoppedBy = $e->getMessage();
|
||
}
|
||
|
||
if (isset($documents)) {
|
||
$this->merge();
|
||
}
|
||
|
||
$state->setLastError($report->stoppedBy ?? $report->pagePhaseEndedBy);
|
||
$this->entityManager->flush();
|
||
|
||
if (null === $report->stoppedBy) {
|
||
$this->logger->info('Content feed sync finished', $report->toLogContext());
|
||
} else {
|
||
$this->logger->warning('Content feed sync stopped', $report->toLogContext());
|
||
}
|
||
|
||
if ($report->isComplete() && $report->changedAnything()) {
|
||
$this->tripCatalog->logUnmatchedCodes();
|
||
}
|
||
|
||
return $report;
|
||
}
|
||
|
||
/**
|
||
* Deletes what the manifest no longer lists and brings every listed item's metadata up
|
||
* to date. Content is not touched here.
|
||
*
|
||
* @return array<string, array{document: ContentDocument, item: ManifestItem}>
|
||
*/
|
||
private function applyManifest(ContentManifest $manifest, ContentSyncReport $report): array
|
||
{
|
||
$stored = $this->documentRepository->findAllByKey();
|
||
|
||
foreach (array_diff_key($stored, $manifest->items) as $document) {
|
||
$this->entityManager->remove($document);
|
||
++$report->deleted;
|
||
}
|
||
|
||
$documents = [];
|
||
|
||
foreach ($manifest->items as $key => $item) {
|
||
$document = $stored[$key] ?? null;
|
||
|
||
if (null === $document) {
|
||
$document = new ContentDocument($item->type, $item->uid, $item->language, $item->modified);
|
||
$this->entityManager->persist($document);
|
||
++$report->added;
|
||
}
|
||
|
||
$document->setModified($item->modified)->setSite($item->site)->setUrl($item->url);
|
||
|
||
if ($document->isPage()) {
|
||
$document
|
||
->setTitle($item->title ?? '')
|
||
->setBreadcrumb([] === $item->breadcrumb ? null : implode(' › ', $item->breadcrumb))
|
||
->setData(['contentUrl' => $item->contentUrl, 'description' => $item->description, 'detailOf' => $item->detailOf]);
|
||
}
|
||
|
||
$documents[$key] = ['document' => $document, 'item' => $item];
|
||
}
|
||
|
||
return $documents;
|
||
}
|
||
|
||
/**
|
||
* @param array<string, array{document: ContentDocument, item: ManifestItem}> $documents
|
||
*/
|
||
private function syncRecords(array $documents, ?\DateTimeImmutable $staleBefore, ContentSyncReport $report): void
|
||
{
|
||
/** @var array<string, array<int, list<ContentDocument>>> $stale type => uid => documents; the records endpoint takes no language */
|
||
$stale = [];
|
||
|
||
foreach ($this->dueForFetch($documents, $staleBefore, pages: false) as $document) {
|
||
$stale[$document->getType()][$document->getUid()][] = $document;
|
||
}
|
||
|
||
foreach ($stale as $type => $byUid) {
|
||
foreach (array_chunk(array_keys($byUid), ContentFeedClient::RECORDS_BATCH_SIZE) as $uids) {
|
||
try {
|
||
$records = $this->client->records($type, $uids);
|
||
} catch (ContentRequestFailedException $e) {
|
||
$this->logger->warning('Content feed: records batch failed, retried next run', ['type' => $type, 'error' => $e->getMessage()]);
|
||
$report->failed += count($uids);
|
||
continue;
|
||
}
|
||
|
||
foreach ($uids as $uid) {
|
||
foreach ($byUid[$uid] as $document) {
|
||
$record = $records[$uid] ?? null;
|
||
|
||
if (null === $record) {
|
||
// Requested but absent: no longer eligible.
|
||
$this->entityManager->remove($document);
|
||
++$report->deleted;
|
||
continue;
|
||
}
|
||
|
||
$this->applyRecord($document, $record);
|
||
++$report->records;
|
||
}
|
||
}
|
||
|
||
$this->entityManager->flush();
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @param array<string, mixed> $record
|
||
*/
|
||
private function applyRecord(ContentDocument $document, array $record): void
|
||
{
|
||
$title = $this->bodyBuilder->title($document->getType(), $record);
|
||
|
||
$document
|
||
->setData($record)
|
||
->setTitle('' !== $title ? $title : $document->getTitle())
|
||
// The merged pages are appended in merge().
|
||
->setBodyMd($this->bodyBuilder->build($document->getType(), $record))
|
||
->markFetched(ContentFetchState::Ok, $this->clock->now());
|
||
}
|
||
|
||
/**
|
||
* @param array<string, array{document: ContentDocument, item: ManifestItem}> $documents
|
||
*
|
||
* @throws ContentFeedStoppedException
|
||
*/
|
||
private function syncPages(array $documents, ?\DateTimeImmutable $staleBefore, bool $unlimited, ContentSyncReport $report): void
|
||
{
|
||
$due = $this->dueForFetch($documents, $staleBefore, pages: true);
|
||
|
||
if (!$unlimited && count($due) > self::PAGES_OUTSIDE_WINDOW) {
|
||
$report->deferred = count($due) - self::PAGES_OUTSIDE_WINDOW;
|
||
$due = array_slice($due, 0, self::PAGES_OUTSIDE_WINDOW);
|
||
}
|
||
|
||
$fetched = 0;
|
||
|
||
try {
|
||
foreach ($due as $document) {
|
||
$this->fetchPage($document, $report);
|
||
|
||
if (0 === ++$fetched % self::FLUSH_EVERY_PAGES) {
|
||
$this->entityManager->flush();
|
||
}
|
||
}
|
||
} catch (PagePhaseEndedException $e) {
|
||
$report->pagePhaseEndedBy = $e->getMessage();
|
||
$this->logger->warning('Content feed: page phase ended early', ['reason' => $e->getMessage()]);
|
||
} finally {
|
||
$this->entityManager->flush();
|
||
}
|
||
}
|
||
|
||
private function fetchPage(ContentDocument $document, ContentSyncReport $report): void
|
||
{
|
||
$contentUrl = $document->getData()['contentUrl'] ?? null;
|
||
|
||
if (!is_string($contentUrl)) {
|
||
$document->markFetched(ContentFetchState::Failed, $this->clock->now());
|
||
++$report->failed;
|
||
|
||
return;
|
||
}
|
||
|
||
$result = $this->client->page($contentUrl);
|
||
|
||
if ($result->transient) {
|
||
++$report->failed;
|
||
|
||
return;
|
||
}
|
||
|
||
if (!$result->isOk()) {
|
||
$this->logger->notice('Content feed: page failed, not refetched until it changes', [
|
||
'uid' => $document->getUid(),
|
||
'contentUrl' => $contentUrl,
|
||
'status' => $result->failedStatus,
|
||
]);
|
||
$document->markFetched(ContentFetchState::Failed, $this->clock->now());
|
||
++$report->failed;
|
||
|
||
return;
|
||
}
|
||
|
||
$markdown = $this->converter->convert($result->html, $document->getUrl() ?? $contentUrl);
|
||
$document->setBodyMd($markdown)->markFetched('' === $markdown ? ContentFetchState::Empty : ContentFetchState::Ok, $this->clock->now());
|
||
++$report->pages;
|
||
}
|
||
|
||
/**
|
||
* Items whose `modified` changed since their last fetch, plus, on a refresh, those
|
||
* fetched before `$staleBefore`, oldest first.
|
||
*
|
||
* @param array<string, array{document: ContentDocument, item: ManifestItem}> $documents
|
||
*
|
||
* @return list<ContentDocument>
|
||
*/
|
||
private function dueForFetch(array $documents, ?\DateTimeImmutable $staleBefore, bool $pages): array
|
||
{
|
||
$due = [];
|
||
|
||
foreach ($documents as ['document' => $document]) {
|
||
if ($document->isPage() !== $pages || !$this->entityManager->contains($document)) {
|
||
continue;
|
||
}
|
||
|
||
$outdated = null !== $staleBefore && (null === $document->getSyncedAt() || $document->getSyncedAt() < $staleBefore);
|
||
|
||
if ($document->isStale() || $outdated) {
|
||
$due[] = $document;
|
||
}
|
||
}
|
||
|
||
usort($due, static fn (ContentDocument $a, ContentDocument $b): int => ($a->getSyncedAt()?->getTimestamp() ?? 0) <=> ($b->getSyncedAt()?->getTimestamp() ?? 0));
|
||
|
||
return $due;
|
||
}
|
||
|
||
/**
|
||
* Appends every detail page to the record it is the detail page of, and keeps the page
|
||
* itself out of the index. Recomputed from scratch on every run, so a change or deletion
|
||
* on either side re-merges.
|
||
*/
|
||
private function merge(): void
|
||
{
|
||
$documents = $this->documentRepository->findAllByKey();
|
||
/** @var array<string, list<ContentDocument>> $pagesByRecord */
|
||
$pagesByRecord = [];
|
||
|
||
foreach ($documents as $page) {
|
||
if (!$page->isPage()) {
|
||
continue;
|
||
}
|
||
|
||
$target = null;
|
||
foreach ($page->getData()['detailOf'] ?? [] as $ref) {
|
||
$candidate = $documents[ContentDocument::key($ref['type'], $ref['uid'], $page->getLanguage())] ?? null;
|
||
|
||
if (null !== $candidate && ContentFetchState::Ok === $candidate->getFetchState()) {
|
||
$target = $candidate;
|
||
break;
|
||
}
|
||
}
|
||
|
||
$page->setMergedInto($target?->getRecordRef());
|
||
|
||
if (null !== $target && ContentFetchState::Ok === $page->getFetchState() && '' !== $page->getBodyMd()) {
|
||
$pagesByRecord[$target->getKey()][] = $page;
|
||
}
|
||
}
|
||
|
||
foreach ($documents as $record) {
|
||
if ($record->isPage() || ContentFetchState::Ok !== $record->getFetchState() || null === $record->getData()) {
|
||
continue;
|
||
}
|
||
|
||
$pages = $pagesByRecord[$record->getKey()] ?? [];
|
||
usort($pages, static fn (ContentDocument $a, ContentDocument $b): int => $a->getUid() <=> $b->getUid());
|
||
|
||
$body = $this->bodyBuilder->build($record->getType(), $record->getData());
|
||
foreach ($pages as $page) {
|
||
$body .= "\n\n".$page->getBodyMd();
|
||
}
|
||
|
||
$record->setBodyMd(trim($body));
|
||
|
||
// A record without a reachable page of its own is linkable through its detail page.
|
||
if (null === $record->getUrl() && [] !== $pages) {
|
||
$record->setUrl($pages[0]->getUrl())->setSite($record->getSite() ?? $pages[0]->getSite());
|
||
}
|
||
}
|
||
}
|
||
|
||
private function inNightWindow(\DateTimeImmutable $now): bool
|
||
{
|
||
$hour = (int) $now->setTimezone(new \DateTimeZone(self::NIGHT_WINDOW_TIMEZONE))->format('G');
|
||
|
||
return $hour >= self::NIGHT_WINDOW_FROM_HOUR && $hour < self::NIGHT_WINDOW_TO_HOUR;
|
||
}
|
||
}
|