Skip to content

Commit 4790b8f

Browse files
bnowakbnowak
committed
Created AsyncPublicationService as internal service to manage async publications state
1 parent 3efa697 commit 4790b8f

7 files changed

Lines changed: 194 additions & 21 deletions

File tree

‎src/bundle/Core/Message/PublishContentAsync.php‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,15 +8,13 @@
88

99
namespace Ibexa\Bundle\Core\Message;
1010

11-
use Ibexa\Contracts\Core\Repository\Values\Content\Language;
12-
1311
final readonly class PublishContentAsync
1412
{
1513
public function __construct(
1614
public int $contentId,
1715
public int $versionNo,
1816
/** @var list<string> */
19-
public array $translations = Language::ALL,
17+
public array $translations,
2018
) {
2119
}
2220
}

‎src/bundle/Core/Message/PublishContentAsyncHandler.php‎

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -8,24 +8,16 @@
88

99
namespace Ibexa\Bundle\Core\Message;
1010

11-
use Ibexa\Contracts\Core\Repository\ContentService;
11+
use Ibexa\Core\Repository\ContentService\AsyncPublicationService;
1212

1313
final class PublishContentAsyncHandler
1414
{
15-
public function __construct(
16-
private ContentService $contentService,
17-
) {
15+
public function __construct(private AsyncPublicationService $asyncPublicationService)
16+
{
1817
}
1918

2019
public function __invoke(PublishContentAsync $message): void
2120
{
22-
$versionInfo = $this->contentService->loadVersionInfoById($message->contentId, $message->versionNo);
23-
24-
$this->contentService->publishVersion(
25-
$versionInfo,
26-
[$versionInfo->getInitialLanguage()->getLanguageCode()],
27-
);
28-
29-
// todo handle "publishing in progress" status
21+
$this->asyncPublicationService->processPublication($message->contentId, $message->versionNo, $message->translations);
3022
}
3123
}

‎src/bundle/Core/Resources/config/storage/legacy/schema.yaml‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -435,11 +435,11 @@ tables:
435435
type: { type: string, nullable: false, length: 128, options: { default: '' } }
436436
created: { type: integer, nullable: false, options: { default: '0' } }
437437
data: { type: text, nullable: true }
438-
ibexa_async_publication_job:
438+
ibexa_content_async_publication_job:
439439
uniqueConstraints:
440-
ibexa_async_publication_content: { fields: [content_id] }
440+
ibexa_content_async_publication_job_content: { fields: [content_id] }
441441
indexes:
442-
ibexa_async_publication_status: { fields: [status] }
442+
ibexa_content_async_publication_job_status: { fields: [status] }
443443
id:
444444
id: { type: integer, nullable: false, options: { autoincrement: true } }
445445
fields:

‎src/lib/Persistence/Legacy/AsyncPublication/Gateway/DoctrineDatabase.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@
1616

1717
class DoctrineDatabase extends Gateway
1818
{
19-
public const string TABLE_ASYNC_PUBLICATION = 'ibexa_async_publication';
19+
public const string TABLE_ASYNC_PUBLICATION = 'ibexa_content_async_publication_job';
2020
public const string COLUMN_ID = 'id';
2121
public const string COLUMN_CONTENT_ID = 'content_id';
2222
public const string COLUMN_VERSION_NO = 'version_no';

‎src/lib/Persistence/Legacy/AsyncPublication/Handler.php‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\CreateStruct;
1313
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\Handler as HandlerInterface;
1414
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\UpdateStruct;
15-
use Ibexa\Core\Base\Exceptions\NotFoundException;
1615

1716
class Handler implements HandlerInterface
1817
{

‎src/lib/Repository/ContentService/AsyncPublicationService.php‎

Lines changed: 182 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,182 @@
1+
<?php
2+
3+
/**
4+
* @copyright Copyright (C) Ibexa AS. All rights reserved.
5+
* @license For full copyright and license information view LICENSE file distributed with this source code.
6+
*/
7+
declare(strict_types=1);
8+
9+
namespace Ibexa\Core\Repository\ContentService;
10+
11+
use DateTime;
12+
use Ibexa\Bundle\Core\Message\PublishContentAsync;
13+
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\AsyncPublicationJob as SPIAsyncPublicationJob;
14+
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\AsyncPublicationJobStatus;
15+
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\CreateStruct;
16+
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\Handler;
17+
use Ibexa\Contracts\Core\Persistence\Content\AsyncPublication\UpdateStruct;
18+
use Ibexa\Contracts\Core\Repository\ContentService;
19+
use Ibexa\Contracts\Core\Repository\PermissionResolver;
20+
use Ibexa\Contracts\Core\Repository\Values\AsyncPublication\AsyncPublicationJob as APIAsyncPublicationJob;
21+
use Ibexa\Contracts\Core\Repository\Values\Content\Language;
22+
use Symfony\Component\Messenger\MessageBusInterface;
23+
24+
/**
25+
* Tracks background (asynchronous) content publication jobs so that AdminUI can surface
26+
* "publishing in progress" / "failed" state and operators can see in-flight and stuck jobs.
27+
*
28+
* The job store is the source of truth for the UI state; Symfony Messenger owns the actual queue.
29+
*/
30+
class AsyncPublicationService
31+
{
32+
public function __construct(
33+
private Handler $persistenceHandler,
34+
private PermissionResolver $permissionResolver,
35+
private MessageBusInterface $bus,
36+
private ContentService $contentService,
37+
) {
38+
}
39+
40+
/**
41+
* Record (or supersede) a background publication job for the given content and mark it queued.
42+
*
43+
* Enforces "one active job per content".
44+
*
45+
* @param list<string> $translations
46+
*/
47+
public function registerPublication(int $contentId, int $versionNo, array $translations = Language::ALL): void
48+
{
49+
$now = (new DateTime())->getTimestamp();
50+
51+
$createStruct = new CreateStruct();
52+
$createStruct->contentId = $contentId;
53+
$createStruct->versionNo = $versionNo;
54+
$createStruct->status = AsyncPublicationJobStatus::QUEUED;
55+
$createStruct->ownerId = $this->permissionResolver->getCurrentUserReference()->getUserId();
56+
$createStruct->created = $now;
57+
$createStruct->modified = $now;
58+
$createStruct->data = ['translations' => $translations];
59+
60+
// todo handle db transaction
61+
62+
$this->persistenceHandler->register($createStruct);
63+
64+
$this->bus->dispatch(
65+
new PublishContentAsync(
66+
$contentId,
67+
$versionNo,
68+
$translations,
69+
),
70+
);
71+
}
72+
73+
/**
74+
* @param string[] $translations
75+
*/
76+
public function processPublication(int $contentId, int $versionNo, array $translations): void
77+
{
78+
$this->markProcessing($contentId);
79+
80+
$versionInfo = $this->contentService->loadVersionInfoById($contentId, $versionNo);
81+
82+
// todo handle db transaction
83+
/** todo handle errors:
84+
* after EACH processing try:
85+
* - job stays in current processing status
86+
* - message is rejected then restarted accordingly to transport retry strategy
87+
* after LAST messenger handling try:
88+
* - job should be in failed status
89+
* - message should be rejected accordingly
90+
*/
91+
$this->contentService->publishVersion(
92+
$versionInfo,
93+
$translations,
94+
);
95+
96+
// The new published version now exists; clearing the job clears the AdminUI "in progress" indicator.
97+
// On failure the job is left in place and marked failed by PublishContentAsyncFailureSubscriber.
98+
$this->markCompleted($contentId);
99+
}
100+
101+
/**
102+
* Mark the job for the given content as being processed by a worker.
103+
*/
104+
private function markProcessing(int $contentId): void
105+
{
106+
$updateStruct = new UpdateStruct();
107+
$updateStruct->status = AsyncPublicationJobStatus::PROCESSING;
108+
$updateStruct->modified = (new DateTime())->getTimestamp();
109+
110+
$this->persistenceHandler->update($contentId, $updateStruct);
111+
}
112+
113+
/**
114+
* Clear the job for the given content once its publication has completed successfully.
115+
*/
116+
private function markCompleted(int $contentId): void
117+
{
118+
$this->persistenceHandler->remove($contentId);
119+
}
120+
121+
/**
122+
* Mark the job for the given content as failed, retaining the error details.
123+
*/
124+
// @phpstan-ignore method.unused
125+
private function markFailed(int $contentId, string $errorMessage): void
126+
{
127+
$updateStruct = new UpdateStruct();
128+
$updateStruct->status = AsyncPublicationJobStatus::FAILED;
129+
$updateStruct->errorMessage = $errorMessage;
130+
$updateStruct->modified = (new DateTime())->getTimestamp();
131+
132+
$this->persistenceHandler->update($contentId, $updateStruct);
133+
}
134+
135+
/**
136+
* Return the job tracked for the given content, or null when none is in flight.
137+
*/
138+
public function getPublicationForContent(int $contentId): ?APIAsyncPublicationJob
139+
{
140+
$spiAsyncPublication = $this->persistenceHandler->getByContentId($contentId);
141+
142+
return $spiAsyncPublication !== null
143+
? $this->buildDomainObject($spiAsyncPublication)
144+
: null;
145+
}
146+
147+
/**
148+
* Return the in-flight and failed jobs (observability surface).
149+
*
150+
* @return \Ibexa\Contracts\Core\Repository\Values\AsyncPublication\AsyncPublicationJob[]
151+
*/
152+
public function findActivePublications(int $offset = 0, int $limit = 25): array
153+
{
154+
return array_map(
155+
fn (SPIAsyncPublicationJob $spiAsyncPublication): APIAsyncPublicationJob => $this->buildDomainObject($spiAsyncPublication),
156+
$this->persistenceHandler->find($offset, $limit)
157+
);
158+
}
159+
160+
/**
161+
* @phpstan-return int<0, max>
162+
*/
163+
public function countActivePublications(): int
164+
{
165+
return $this->persistenceHandler->count();
166+
}
167+
168+
protected function buildDomainObject(SPIAsyncPublicationJob $spiAsyncPublication): APIAsyncPublicationJob
169+
{
170+
return new APIAsyncPublicationJob([
171+
'id' => $spiAsyncPublication->id,
172+
'contentId' => $spiAsyncPublication->contentId,
173+
'versionNo' => $spiAsyncPublication->versionNo,
174+
'status' => $spiAsyncPublication->status->value,
175+
'ownerId' => $spiAsyncPublication->ownerId,
176+
'created' => new DateTime("@{$spiAsyncPublication->created}"),
177+
'modified' => new DateTime("@{$spiAsyncPublication->modified}"),
178+
'errorMessage' => $spiAsyncPublication->errorMessage,
179+
'data' => $spiAsyncPublication->data,
180+
]);
181+
}
182+
}

‎src/lib/Resources/settings/async_publication.yml‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
services:
22
Ibexa\Core\Repository\ContentService\AsyncPublicationService:
3+
autowire: true
34
arguments:
45
$persistenceHandler: '@ibexa.spi.persistence.legacy.async_publication.handler'
5-
$permissionResolver: '@Ibexa\Contracts\Core\Repository\PermissionResolver'
6+
$bus: '@ibexa.messenger.bus'
7+
68
lazy: true

0 commit comments

Comments
 (0)