feat: import people
Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>
This commit is contained in:
@@ -23,7 +23,10 @@ use KTXF\Resource\Identifier\ResourceIdentifier;
|
||||
use KTXF\Resource\Identifier\ResourceIdentifiers;
|
||||
use KTXF\Resource\Identifier\ServiceIdentifier;
|
||||
use KTXF\Routing\Attributes\AuthenticatedRoute;
|
||||
use KTXM\PeopleManager\Import\ImportOptions;
|
||||
use KTXM\PeopleManager\Import\ImportService;
|
||||
use KTXM\PeopleManager\Manager;
|
||||
use KTXM\PeopleManager\Stream\ExpectedTotal;
|
||||
use Psr\Log\LoggerInterface;
|
||||
use Throwable;
|
||||
|
||||
@@ -50,6 +53,7 @@ class DefaultController extends ControllerAbstract {
|
||||
private readonly SessionTenant $tenantIdentity,
|
||||
private readonly SessionIdentity $userIdentity,
|
||||
private readonly Manager $manager,
|
||||
private readonly ImportService $importService,
|
||||
private readonly LoggerInterface $logger,
|
||||
) {}
|
||||
|
||||
@@ -153,7 +157,8 @@ class DefaultController extends ControllerAbstract {
|
||||
'entity.delete' => $this->entityDelete($tenantId, $userId, $data),
|
||||
'entity.move' => $this->entityMove($tenantId, $userId, $data),
|
||||
'entity.copy' => $this->entityCopy($tenantId, $userId, $data),
|
||||
|
||||
'entity.import' => $this->entityImport($tenantId, $userId, $data, $version, $transaction),
|
||||
|
||||
default => throw new InvalidArgumentException(self::ERR_INVALID_OPERATION . $operation)
|
||||
};
|
||||
}
|
||||
@@ -540,39 +545,117 @@ class DefaultController extends ControllerAbstract {
|
||||
$sources = null;
|
||||
}
|
||||
|
||||
|
||||
$filter = $data['filter'] ?? null;
|
||||
$sort = $data['sort'] ?? null;
|
||||
$range = $data['range'] ?? null;
|
||||
|
||||
$entityGenerator = $this->manager->entityListStream($tenantId, $userId, $sources, $filter, $sort, $range);
|
||||
$logger = $this->logger;
|
||||
$entities = $this->manager->entityListStream($tenantId, $userId, $sources, $filter, $sort, $range);
|
||||
|
||||
$responseGenerator = (function () use ($entityGenerator, $version, $transaction, $logger): \Generator {
|
||||
yield ['type' => 'control', 'status' => 'start', 'version' => $version, 'transaction' => $transaction];
|
||||
return new StreamedNdJsonResponse(
|
||||
$this->streamEnvelope($entities, $version, $transaction),
|
||||
1,
|
||||
200,
|
||||
['Content-Type' => 'application/json'],
|
||||
);
|
||||
}
|
||||
|
||||
$total = 0;
|
||||
/**
|
||||
* Import vCards into a collection, streaming one NDJSON event per contact.
|
||||
*
|
||||
* Request data: { target: "provider:service:collection", data: "<raw vCard string>", options?: {...} }
|
||||
*/
|
||||
private function entityImport(string $tenantId, string $userId, array $data, int $version, string $transaction): StreamedNdJsonResponse {
|
||||
|
||||
if (!isset($data['target'])) {
|
||||
throw new InvalidArgumentException(self::ERR_MISSING_TARGET);
|
||||
}
|
||||
if (!is_string($data['target'])) {
|
||||
throw new InvalidArgumentException(self::ERR_INVALID_TARGET);
|
||||
}
|
||||
if (!isset($data['data']) || !is_string($data['data']) || trim($data['data']) === '') {
|
||||
throw new InvalidArgumentException('Invalid parameter: data must be a non-empty string');
|
||||
}
|
||||
|
||||
$target = ResourceIdentifier::fromString($data['target']);
|
||||
if (!$target instanceof CollectionIdentifier) {
|
||||
throw new InvalidArgumentException('Invalid parameter: target must be provider:service:collection');
|
||||
}
|
||||
|
||||
$options = ImportOptions::fromArray($data['options'] ?? []);
|
||||
|
||||
// Spill the payload to a temp file and release the in-memory copy before the
|
||||
// (potentially long) streaming parse, so peak memory stays at one contact object.
|
||||
$tempFile = tmpfile();
|
||||
if ($tempFile === false) {
|
||||
throw new \RuntimeException('Unable to allocate temporary file for import');
|
||||
}
|
||||
fwrite($tempFile, $data['data']);
|
||||
unset($data);
|
||||
rewind($tempFile);
|
||||
|
||||
$events = $this->importService->import($tempFile, $target, $options, $tenantId, $userId);
|
||||
$frames = $this->streamEnvelope($events, $version, $transaction);
|
||||
|
||||
// Stream the envelope, releasing the spilled temp file once it is fully
|
||||
// drained (or the client disconnects) so it never outlives its stream.
|
||||
$response = (function () use ($frames, $tempFile): \Generator {
|
||||
try {
|
||||
foreach ($entityGenerator as $entity) {
|
||||
if (!$entity instanceof JsonSerializable) {
|
||||
continue;
|
||||
}
|
||||
yield [
|
||||
'type' => 'data',
|
||||
'data' => $entity->jsonSerialize()
|
||||
];
|
||||
$total++;
|
||||
}
|
||||
} catch (\Throwable $t) {
|
||||
$logger->error('Error streaming entities', ['exception' => $t]);
|
||||
yield ['type' => 'error', 'message' => $t->getMessage()];
|
||||
return;
|
||||
yield from $frames;
|
||||
} finally {
|
||||
fclose($tempFile);
|
||||
}
|
||||
|
||||
yield ['type' => 'control', 'status' => 'end', 'total' => $total];
|
||||
})();
|
||||
|
||||
return new StreamedNdJsonResponse($responseGenerator, 1, 200, ['Content-Type' => 'application/json']);
|
||||
return new StreamedNdJsonResponse($response, 1, 200);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wrap a generator of JsonSerializable domain objects in the canonical NDJSON
|
||||
* stream envelope shared by every streaming operation:
|
||||
*
|
||||
* control:start {version, transaction, total?} — total? = expected count
|
||||
* data {data} — one per domain object
|
||||
* error {message} — on failure, then stop
|
||||
* control:end {total} — total = objects emitted
|
||||
*
|
||||
* If the generator leads with an {@see ExpectedTotal} event, its value is
|
||||
* folded into the start frame's `total` (the progress denominator) rather
|
||||
* than emitted as a data frame.
|
||||
*
|
||||
* @param \Generator<\JsonSerializable> $items
|
||||
*/
|
||||
private function streamEnvelope(\Generator $items, int $version, string $transaction): \Generator {
|
||||
// Peek the first event: an expected-total marker rides on the start frame.
|
||||
$expected = null;
|
||||
$items->rewind();
|
||||
if ($items->valid() && $items->current() instanceof ExpectedTotal) {
|
||||
$expected = $items->current()->expectedTotal();
|
||||
$items->next();
|
||||
}
|
||||
|
||||
$start = ['type' => 'control', 'status' => 'start', 'version' => $version, 'transaction' => $transaction];
|
||||
if ($expected !== null) {
|
||||
$start['total'] = $expected;
|
||||
}
|
||||
yield $start;
|
||||
|
||||
$total = 0;
|
||||
try {
|
||||
for (; $items->valid(); $items->next()) {
|
||||
$item = $items->current();
|
||||
if (!$item instanceof \JsonSerializable) {
|
||||
continue;
|
||||
}
|
||||
yield ['type' => 'data', 'data' => $item->jsonSerialize()];
|
||||
$total++;
|
||||
}
|
||||
} catch (\Throwable $t) {
|
||||
$this->logger->error('Error streaming response', ['exception' => $t]);
|
||||
yield ['type' => 'error', 'message' => $t->getMessage()];
|
||||
return;
|
||||
}
|
||||
|
||||
yield ['type' => 'control', 'status' => 'end', 'total' => $total];
|
||||
}
|
||||
|
||||
private function entityFetch(string $tenantId, string $userId, array $data): mixed {
|
||||
|
||||
Reference in New Issue
Block a user