feat: add MCP server for the website chatbot over a mirrored TYPO3 content feed

src/ContentFeed/ContentSyncer.php

@@ -0,0 +1,367 @@
+<?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;
+    }
+}