Files
mail_manager/lib/Manager.php
Sebastian Krupinski 935743963f
Some checks failed
Build Test / test (pull_request) Successful in 32s
JS Unit Tests / test (pull_request) Failing after 31s
PHP Unit Tests / test (pull_request) Successful in 1m1s
feat: mail entity download
Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>
2026-05-28 23:22:02 -04:00

1306 lines
60 KiB
PHP

<?php
declare(strict_types=1);
namespace KTXM\MailManager;
use InvalidArgumentException;
use KTXC\Resource\ProviderManager;
use KTXF\Mail\Collection\CollectionBaseInterface;
use KTXF\Mail\Collection\CollectionPropertiesMutableInterface;
use KTXF\Mail\Collection\ICollectionBase;
use KTXF\Mail\Entity\IMessageBase;
use KTXF\Mail\Object\MessagePropertiesMutableInterface;
use KTXF\Mail\Provider\ProviderBaseInterface;
use KTXF\Mail\Provider\ProviderServiceDiscoverInterface;
use KTXF\Mail\Provider\ProviderServiceMutateInterface;
use KTXF\Mail\Provider\ProviderServiceTestInterface;
use KTXF\Mail\Service\ServiceBaseInterface;
use KTXF\Mail\Service\ServiceCollectionMutableInterface;
use KTXF\Mail\Service\ServiceConfigurableInterface;
use KTXF\Mail\Service\ServiceEntityMutableInterface;
use KTXF\Mail\Service\ServiceMutableInterface;
use KTXF\Resource\BinaryResource;
use KTXF\Resource\Filter\IFilter;
use KTXF\Resource\Identifier\CollectionIdentifier;
use KTXF\Resource\Identifier\EntityIdentifier;
use KTXF\Resource\Identifier\ResourceIdentifiers;
use KTXF\Resource\Provider\ResourceServiceIdentityInterface;
use KTXF\Resource\Provider\ResourceServiceLocationInterface;
use KTXF\Resource\Range\RangeAnchorType;
use KTXF\Resource\Range\RangeType;
use KTXF\Resource\Selector\CollectionSelector;
use KTXF\Resource\Selector\EntitySelector;
use KTXF\Resource\Selector\ServiceSelector;
use KTXF\Resource\Selector\SourceSelector;
use KTXF\Resource\Sort\ISort;
use Psr\Log\LoggerInterface;
/**
* Mail Manager
*
* Provides unified mail sending across multiple providers
*/
class Manager {
public function __construct(
private LoggerInterface $logger,
private ProviderManager $providerManager,
) { }
/**
* Retrieve available providers
*
* @param SourceSelector|null $sources collection of provider identifiers
*
* @return array<string,ProviderBaseInterface> collection of available providers e.g. ['provider1' => IProvider, 'provider2' => IProvider]
*/
public function providerList(string $tenantId, string $userId, ?SourceSelector $sources = null): array {
// determine filter from sources
$filter = ($sources !== null && $sources->identifiers() !== []) ? $sources->identifiers() : null;
// retrieve providers from provider manager
return $this->providerManager->providers(ProviderBaseInterface::TYPE_MAIL, $filter);
}
/**
* Retrieve specific provider for specific user
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param string $provider provider identifier
*
* @return ProviderBaseInterface
* @throws InvalidArgumentException
*/
public function providerFetch(string $tenantId, string $userId, string $provider): ProviderBaseInterface {
// retrieve provider
$providers = $this->providerList($tenantId, $userId, new SourceSelector([$provider => true]));
if (!isset($providers[$provider])) {
throw new InvalidArgumentException("Provider '$provider' not found");
}
return $providers[$provider];
}
/**
* Confirm which providers are available
*
* @param SourceSelector|null $sources collection of provider identifiers to confirm
*
* @return array<string,bool> collection of providers and their availability status e.g. ['provider1' => true, 'provider2' => false]
*/
public function providerExtant(string $tenantId, string $userId, SourceSelector $sources): array {
// determine which providers are available
$providersResolved = $this->providerList($tenantId, $userId, $sources);
$providersAvailable = array_keys($providersResolved);
$providersUnavailable = array_diff($sources->identifiers(), $providersAvailable);
// construct response data
$responseData = array_merge(
array_fill_keys($providersAvailable, true),
array_fill_keys($providersUnavailable, false)
);
return $responseData;
}
/**
* Retrieve available services for specific user
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param SourceSelector|null $sources list of provider and service identifiers
*
* @return array<string,<string,ServiceBaseInterface>> collections of available services e.g. ['provider1' => ['service1' => IServiceBase], 'provider2' => ['service2' => IServiceBase]]
*/
public function serviceList(string $tenantId, string $userId, ?SourceSelector $sources = null): array {
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
// retrieve services for each provider
$responseData = [];
foreach ($providers as $provider) {
$serviceFilter = $sources[$provider->identifier()] instanceof ServiceSelector ? $sources[$provider->identifier()]->identifiers() : [];
$services = $provider->serviceList($tenantId, $userId, $serviceFilter);
$responseData[$provider->identifier()] = $services;
}
return $responseData;
}
/**
* Retrieve service for specific user
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param string $providerId provider identifier
* @param string|int $serviceId service identifier
*
* @return ServiceBaseInterface
* @throws InvalidArgumentException
*/
public function serviceFetch(string $tenantId, string $userId, string $providerId, string|int $serviceId): ServiceBaseInterface {
// retrieve provider and service
$service = $this->providerFetch($tenantId, $userId, $providerId)->serviceFetch($tenantId, $userId, $serviceId);
if ($service === null) {
throw new InvalidArgumentException("Service '$serviceId' not found for provider '$providerId'");
}
// retrieve services
return $service;
}
/**
* Confirm which services are available
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param SourceSelector|null $sources collection of provider and service identifiers to confirm
*
* @return array<string,bool> collection of providers and their availability status e.g. ['provider1' => ['service1' => false], 'provider2' => ['service2' => true, 'service3' => true]]
*/
public function serviceExtant(string $tenantId, string $userId, SourceSelector $sources): array {
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
$providersRequested = $sources->identifiers();
$providersUnavailable = array_diff($providersRequested, array_keys($providers));
// initialize response with unavailable providers
$responseData = array_fill_keys($providersUnavailable, false);
// retrieve services for each available provider
foreach ($providers as $provider) {
$serviceSelector = $sources[$provider->identifier()];
$serviceAvailability = $provider->serviceExtant($tenantId, $userId, ...$serviceSelector->identifiers());
$responseData[$provider->identifier()] = $serviceAvailability;
}
return $responseData;
}
/**
* Find a service that handles a specific email address
*
* Searches all providers for a service that handles the given address,
* respecting the user context for scope filtering.
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier
* @param string $address Email address to find service for
*
* @return ?ServiceBaseInterface
*/
public function serviceFindByAddress(string $tenantId, string $userId, string $address): ?ServiceBaseInterface {
// retrieve providers
$providers = $this->providerList($tenantId, $userId);
foreach ($providers as $providerId => $provider) {
$service = $provider->serviceFindByAddress($tenantId, $userId, $address);
if ($service !== null) {
return $service;
}
}
return null;
}
/**
* Create a new service
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param string $providerId Provider identifier
* @param array $data Service configuration data
*
* @return ServiceBaseInterface Created service
*
* @throws InvalidArgumentException If provider doesn't support service creation
*/
public function serviceCreate(string $tenantId, ?string $userId, string $providerId, array $data): ServiceBaseInterface {
// retrieve provider and service
$provider = $this->providerFetch($tenantId, $userId, $providerId);
if ($provider instanceof ProviderServiceMutateInterface === false) {
throw new InvalidArgumentException("Provider '$providerId' does not support service creation");
}
if ($provider->capable(ProviderServiceMutateInterface::CAPABILITY_SERVICE_CREATE) === false) {
throw new InvalidArgumentException("Provider '$providerId' is not capable of creating services");
}
// Create a fresh service instance
$service = $provider->serviceFresh();
// Deserialize the data into the service
$service->jsonDeserialize($data);
// Create the service
$serviceId = $provider->serviceCreate($tenantId, $userId, $service);
// Fetch and return the created service
return $provider->serviceFetch($tenantId, $userId, $serviceId);
}
/**
* Update an existing service
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier for context
* @param string $providerId Provider identifier
* @param string|int $serviceId Service identifier
* @param array $data Updated service configuration data
* @param bool $delta Whether the update is a delta (partial) update or a full replacement
*
* @return ServiceBaseInterface Updated service
*
* @throws InvalidArgumentException If provider doesn't support service modification or service not found
*/
public function serviceUpdate(string $tenantId, string $userId, string $providerId, string|int $serviceId, array $data, bool $delta = false): ServiceBaseInterface {
// retrieve provider and service
$provider = $this->providerFetch($tenantId, $userId, $providerId);
if ($provider instanceof ProviderServiceMutateInterface === false) {
throw new InvalidArgumentException("Provider '$providerId' does not support service modification");
}
if ($provider->capable(ProviderServiceMutateInterface::CAPABILITY_SERVICE_MODIFY) === false) {
throw new InvalidArgumentException("Provider '$providerId' is not capable of modifying services");
}
// Fetch existing service
$service = $provider->serviceFetch($tenantId, $userId, $serviceId);
if ($service === null) {
throw new InvalidArgumentException("Service '$serviceId' not found");
}
if ($service instanceof ServiceMutableInterface === false) {
throw new InvalidArgumentException("Service '$serviceId' is not mutable and cannot be updated");
}
// Update with new data
$service->jsonDeserialize($data, $delta);
// Modify the service
$provider->serviceModify($tenantId, $userId, $service);
// Fetch and return the updated service
return $provider->serviceFetch($tenantId, $userId, $serviceId);
}
/**
* Delete a service
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier for context
* @param string $providerId Provider identifier
* @param string|int $serviceId Service identifier
*
* @return bool True if service was deleted
*
* @throws InvalidArgumentException If provider doesn't support service deletion or service not found
*/
public function serviceDelete(string $tenantId, string $userId, string $providerId, string|int $serviceId): bool {
// retrieve provider and service
$provider = $this->providerFetch($tenantId, $userId, $providerId);
if ($provider instanceof ProviderServiceMutateInterface === false) {
throw new InvalidArgumentException("Provider '$providerId' does not support service deletion");
}
if ($provider->capable(ProviderServiceMutateInterface::CAPABILITY_SERVICE_DESTROY) === false) {
throw new InvalidArgumentException("Provider '$providerId' is not capable of deleting services");
}
// Fetch existing service
$service = $provider->serviceFetch($tenantId, $userId, $serviceId);
if ($service === null) {
throw new InvalidArgumentException("Service '$serviceId' not found");
}
// Delete the service
return $provider->serviceDestroy($tenantId, $userId, $service);
}
/**
* Discover mail service settings from identity, yielding results as each provider completes
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier
* @param string|null $providerId Specific provider to use for discovery (or null for all)
* @param string $identity Identity to discover configuration for (e.g., email address)
* @param string|null $location Optional hostname to test directly (bypasses DNS SRV lookup)
* @param string|null $secret Optional password/token to validate discovered service
*
* @return \Generator Yields providerId => ResourceServiceLocationInterface pairs as each provider completes
*/
public function serviceDiscover(
string $tenantId,
string $userId,
string|null $providerId,
string $identity,
string|null $location = null,
string|null $secret = null
): \Generator {
$providers = $this->providerList($tenantId, $userId, $providerId !== null ? new SourceSelector([$providerId => true]) : null);
foreach ($providers as $currentProviderId => $provider) {
if ($provider instanceof ProviderServiceDiscoverInterface === false) {
continue;
}
try {
$result = $provider->serviceDiscover($tenantId, $userId, $identity, $location, $secret);
if ($result !== null) {
yield $currentProviderId => $result;
}
} catch (\Throwable $e) {
$this->logger->warning('Provider autodiscovery failed', [
'provider' => $currentProviderId,
'identity' => $identity,
'error' => $e->getMessage(),
]);
}
}
}
/**
* Test a mail service connection
*
* Tests connectivity and authentication for either an existing service
* or a fresh configuration. Delegates to the appropriate provider.
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier for context
* @param string $providerId Provider ID (for existing service or targeted test)
* @param string|int|null $serviceId Service ID (for existing service test)
* @param ResourceServiceLocationInterface|array|null $location Service location (for fresh config test)
* @param ResourceServiceIdentityInterface|array|null $identity Service credentials (for fresh config test)
*
* @return array Test results
*
* @throws InvalidArgumentException If invalid parameters
*/
public function serviceTest(
string $tenantId,
string $userId,
string $providerId,
string|int|null $serviceId = null,
ResourceServiceLocationInterface|array|null $location = null,
ResourceServiceIdentityInterface|array|null $identity = null
): array {
// retrieve provider
$provider = $this->providerFetch($tenantId, $userId, $providerId);
if ($provider instanceof ProviderServiceTestInterface === false) {
throw new InvalidArgumentException("Provider '$providerId' does not support service testing");
}
// Testing existing service
if ($providerId !== null && $serviceId !== null) {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
if ($service === null) {
throw new InvalidArgumentException("Service not found: $providerId/$serviceId");
}
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service $providerId/$serviceId is disabled");
}
// test service
try {
return $provider->serviceTest($service);
} catch (\Throwable $e) {
throw new InvalidArgumentException('Service test failed: ' . $e->getMessage());
}
}
// Testing fresh configuration
if ($location !== null && $identity !== null) {
if ($provider instanceof ProviderServiceMutateInterface === false) {
throw new InvalidArgumentException("Provider '$providerId' does not support fresh service configuration testing");
}
if (empty($location['type'])) {
throw new InvalidArgumentException('Service location not valid');
}
if (empty($identity['type'])) {
throw new InvalidArgumentException('Service identity not valid');
}
/** @var ServiceConfigurableInterface|ServiceMutableInterface $service */
$service = $provider->serviceFresh();
if ($location instanceof ResourceServiceLocationInterface === false) {
$location = $service->freshLocation($location['type'], (array)$location);
$service->setLocation($location);
}
if ($identity instanceof ResourceServiceIdentityInterface === false) {
$identity = $service->freshIdentity($identity['type'], (array)$identity);
$service->setIdentity($identity);
}
return $provider->serviceTest($service);
}
throw new InvalidArgumentException(
'Either (provider + service) or (provider + location + identity) must be provided'
);
}
// ==================== Collection Operations ====================
/**
* List collections across services
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param SourceSelector|null $sources Provider/service sources
* @param IFilter|null $filter Collection filter
* @param ISort|null $sort Collection sort
*
* @return array<string, array<string|int, array<string|int, ICollectionBase>>> Collections grouped by provider/service
*/
public function collectionList(string $tenantId, ?string $userId, ?SourceSelector $sources = null, ?IFilter $filter = null, ?ISort $sort = null): array {
// confirm that sources are provided
if ($sources === null) {
$sources = new SourceSelector([]);
}
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
// retrieve services for each provider
$responseData = [];
foreach ($providers as $provider) {
$serviceFilter = $sources[$provider->identifier()] instanceof ServiceSelector ? $sources[$provider->identifier()]->identifiers() : [];
/** @var ServiceBaseInterface[] $services */
$services = $provider->serviceList($tenantId, $userId, $serviceFilter);
// retrieve collections for each service
foreach ($services as $service) {
if ($service->getEnabled() === false) {
continue;
}
// construct filter for collections
$collectionFilter = null;
if ($filter !== null && $filter !== []) {
$collectionFilter = $service->collectionListFilter();
foreach ($filter as $attribute => $value) {
$collectionFilter->condition($attribute, $value);
}
}
// construct sort for collections
$collectionSort = null;
if ($sort !== null && $sort !== []) {
$collectionSort = $service->collectionListSort();
foreach ($sort as $attribute => $direction) {
$collectionSort->condition($attribute, $direction);
}
}
$collections = $service->collectionList('', $collectionFilter, $collectionSort);
if ($collections !== []) {
$responseData[$provider->identifier()][$service->identifier()] = $collections;
}
}
}
return $responseData;
}
/**
* Check if collections exist
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param SourceSelector $sources Collection sources with identifiers
*
* @return array<string, array<string|int, array<string|int, bool>>> Existence map grouped by provider/service
*/
public function collectionExtant(string $tenantId, ?string $userId, SourceSelector $sources): array {
// retrieve available providers
$providers = $this->providerList($tenantId, $userId, $sources);
$providersRequested = $sources->identifiers();
$providersUnavailable = array_diff($providersRequested, array_keys($providers));
// initialize response with unavailable providers
$responseData = array_fill_keys($providersUnavailable, false);
// check services and collections for each available provider
foreach ($providers as $provider) {
// extract services for this provider
$serviceSelector = $sources[$provider->identifier()];
$servicesRequested = $serviceSelector->identifiers();
/** @var ServiceBaseInterface[] $servicesAvailable */
$servicesAvailable = $provider->serviceList($tenantId, $userId, $servicesRequested);
$servicesUnavailable = array_diff($servicesRequested, array_keys($servicesAvailable));
// mark unavailable services as false
if ($servicesUnavailable !== []) {
$responseData[$provider->identifier()] = array_fill_keys($servicesUnavailable, false);
}
// confirm collections for each available service
foreach ($servicesAvailable as $service) {
// omit disabled services
if ($service->getEnabled() === false) {
$responseData[$provider->identifier()][$service->identifier()] = false;
continue;
}
// extract collections requested for this service
$collectionSelector = $serviceSelector[$service->identifier()];
$collectionsRequested = $collectionSelector instanceof CollectionSelector ? $collectionSelector->identifiers() : [];
if ($collectionsRequested === []) {
continue;
}
// check each requested collection
$collectionsAvailable = $service->collectionExtant(...$collectionsRequested);
$collectionsUnavailable = array_diff($collectionsRequested, array_keys($collectionsAvailable));
$responseData[$provider->identifier()][$service->identifier()] = array_merge(
$collectionsAvailable,
array_fill_keys($collectionsUnavailable, false)
);
}
}
return $responseData;
}
/**
* Fetch a specific collection
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param string $providerId Provider identifier
* @param string|int $serviceId Service identifier
* @param string|int $collectionId Collection identifier
*
* @return CollectionBaseInterface|null
*/
public function collectionFetch(string $tenantId, ?string $userId, string $providerId, string|int $serviceId, string|int $collectionId): ?CollectionBaseInterface {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
if ($service === null || $service->getEnabled() === false) {
return null;
}
// retrieve collection
return $service->collectionFetch($collectionId);
}
/**
* Create a new collection for a specific user
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param CollectionIdentifier $target target parent collection identifier
* @param CollectionPropertiesMutableInterface|array $properties properties for the new collection
* @param array $options additional options for creation
*
* @return CollectionBaseInterface
* @throws InvalidArgumentException
*/
public function collectionCreate(string $tenantId, string $userId, string $provider, string|int $service, CollectionIdentifier|null $target, CollectionPropertiesMutableInterface|array $properties, array $options = []): CollectionBaseInterface {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $provider, $service);
// Check if service supports collection creation
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' not found or is disabled");
}
if ($service instanceof ServiceCollectionMutableInterface === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' does not support collection mutations");
}
if (!$service->capable(ServiceCollectionMutableInterface::CAPABILITY_COLLECTION_CREATE)) {
throw new InvalidArgumentException("Service '{$service->identifier()}' is not capable of creating collections");
}
// convert properties if necessary
if ($properties instanceof CollectionPropertiesMutableInterface === false) {
$properties = $service->collectionFresh()->getProperties()->jsonDeserialize($properties);
}
// Create collection
return $service->collectionCreate($target, $properties, $options);
}
/**
* Modify an existing collection for a specific user
*
* @param string $tenantId tenant identifier
* @param string $userId user identifier
* @param CollectionIdentifier $target target collection identifier
* @param CollectionPropertiesMutableInterface|array $properties properties to modify
*
* @return CollectionBaseInterface
* @throws InvalidArgumentException
*/
public function collectionUpdate(string $tenantId, string $userId, CollectionIdentifier $target, CollectionPropertiesMutableInterface|array $properties): CollectionBaseInterface {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $target->provider(), $target->service());
// Check if service supports collection creation
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' not found or is disabled");
}
if ($service instanceof ServiceCollectionMutableInterface === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' does not support collection mutations");
}
if (!$service->capable(ServiceCollectionMutableInterface::CAPABILITY_COLLECTION_UPDATE)) {
throw new InvalidArgumentException("Service '{$service->identifier()}' is not capable of updating collections");
}
// convert properties if necessary
if ($properties instanceof CollectionPropertiesMutableInterface === false) {
$properties = $service->collectionFresh()->getProperties()->jsonDeserialize($properties);
}
// Update collection
return $service->collectionUpdate($target, $properties);
}
/**
* Delete a specific collection
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param CollectionIdentifier $target Target collection identifier
* @param array $options Additional options for deletion (e.g., 'force' => true to force delete even if not empty)
*
* @return CollectionBaseInterface|null
*/
public function collectionDelete(string $tenantId, ?string $userId, CollectionIdentifier $target, array $options = []): CollectionBaseInterface | bool {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $target->provider(), $target->service());
// Check if service supports collection deletion
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' not found or is disabled");
}
if ($service instanceof ServiceCollectionMutableInterface === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' does not support collection mutations");
}
if (!$service->capable(ServiceCollectionMutableInterface::CAPABILITY_COLLECTION_DELETE)) {
throw new InvalidArgumentException("Service '{$service->identifier()}' is not capable of deleting collections");
}
// convert options
$force = $options['force'] ?? false;
// delete collection
return $service->collectionDelete($target, $force);
}
/**
* Move a specific collection to a new parent collection
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param CollectionIdentifier $target Target collection identifier (new parent)
* @param CollectionIdentifier $source Source collection identifier (collection to move)
*
* @return CollectionBaseInterface Moved collection
*/
public function collectionMove(string $tenantId, ?string $userId, CollectionIdentifier $target, CollectionIdentifier $source): CollectionBaseInterface {
// validate that source and target are the same provider and service
if ($source->provider() !== $target->provider() || $source->service() !== $target->service()) {
throw new InvalidArgumentException("Source '{$source->service()}' and target '{$target->service()}' collections must belong to the same provider and service");
}
// Validate that source and target are not the same
if ($source->collection() === $target->collection()) {
throw new InvalidArgumentException("Source '{$source->collection()}' and target '{$target->collection()}' collections are the same");
}
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $source->provider(), $source->service());
// Check if service supports collection move
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' not found or is disabled");
}
if ($service instanceof ServiceCollectionMutableInterface === false) {
throw new InvalidArgumentException("Service '{$service->identifier()}' does not support collection mutations");
}
if (!$service->capable(ServiceCollectionMutableInterface::CAPABILITY_COLLECTION_MOVE)) {
throw new InvalidArgumentException("Service '{$service->identifier()}' is not capable of moving collections");
}
// move collection
return $service->collectionMove($target, $source);
}
// ==================== Message Operations ====================
/**
* List messages in a collection
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier
* @param SourceSelector $sources Message sources with collection identifiers
* @param array|null $filter Message filter
* @param array|null $sort Message sort
* @param array|null $range Message range/pagination
*
* @return array<string, array<string|int, array<string|int, array<string|int, IMessageBase>>>> Messages grouped by provider/service/collection
*/
public function entityListBulk(string $tenantId, string $userId, SourceSelector $sources, array|null $filter = null, array|null $sort = null, array|null $range = null): array {
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
// retrieve services for each provider
$responseData = [];
foreach ($providers as $provider) {
// retrieve services for each provider
$serviceSelector = $sources[$provider->identifier()];
$servicesSelected = $provider->serviceList($tenantId,$userId, $serviceSelector->identifiers());
/** @var ServiceBaseInterface $service */
foreach ($servicesSelected as $service) {
// omit disabled services
if ($service->getEnabled() === false) {
continue;
}
// retrieve collections for each service
$collectionSelector = $serviceSelector[$service->identifier()];
$collectionSelected = $collectionSelector instanceof CollectionSelector ? $collectionSelector->identifiers() : [];
if ($collectionSelected === []) {
$collections = $service->collectionList('');
$collectionSelected = array_map(
fn($collection) => $collection->identifier(),
$collections
);
}
if ($collectionSelected === []) {
continue;
}
// construct filter for entities
$entityFilter = null;
if ($filter !== null && $filter !== []) {
$entityFilter = $service->entityListFilter();
foreach ($filter as $attribute => $value) {
$entityFilter->condition($attribute, $value);
}
}
// construct sort for entities
$entitySort = null;
if ($sort !== null && $sort !== []) {
$entitySort = $service->entityListSort();
foreach ($sort as $attribute => $direction) {
$entitySort->condition($attribute, $direction);
}
}
// construct range for entities
$entityRange = null;
if ($range !== null && $range !== [] && isset($range['type'])) {
$entityRange = $service->entityListRange(RangeType::from($range['type']));
// Cast to IRangeTally if the range type is TALLY
if ($entityRange->type() === RangeType::TALLY) {
/** @var IRangeTally $entityRange */
if (isset($range['anchor'])) {
$entityRange->setAnchor(RangeAnchorType::from($range['anchor']));
}
if (isset($range['position'])) {
$entityRange->setPosition($range['position']);
}
if (isset($range['tally'])) {
$entityRange->setTally($range['tally']);
}
}
}
// retrieve entities for each collection
foreach ($collectionSelected as $collectionId) {
$entities = $service->entityList($collectionId, $entityFilter, $entitySort, $entityRange, null);
// skip collections with no entities
if ($entities === []) {
continue;
}
$responseData[$provider->identifier()][$service->identifier()][$collectionId] = $entities;
}
}
}
return $responseData;
}
/**
* Stream entities
*
* @since 2026.02.01
*
* @param string $tenantId Tenant identifier
* @param string $userId User identifier
* @param SourceSelector $sources Message sources with collection identifiers
* @param array|null $filter Message filter
* @param array|null $sort Message sort
* @param array|null $range Message range/pagination
*
* @return \Generator<EntityBaseInterface> Yields each entity as it is retrieved
*/
public function entityListStream(string $tenantId, string $userId, SourceSelector $sources, array|null $filter = null, array|null $sort = null, array|null $range = null): \Generator {
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
// retrieve services for each provider
foreach ($providers as $provider) {
$serviceSelector = $sources[$provider->identifier()];
$servicesSelected = $provider->serviceList($tenantId, $userId, $serviceSelector->identifiers());
/** @var ServiceBaseInterface $service */
foreach ($servicesSelected as $service) {
// omit disabled services
if ($service->getEnabled() === false) {
continue;
}
// retrieve collections for each service
$collectionSelector = $serviceSelector[$service->identifier()];
$collectionSelected = $collectionSelector instanceof CollectionSelector ? $collectionSelector->identifiers() : [];
if ($collectionSelected === []) {
$collections = $service->collectionList('');
$collectionSelected = array_map(
fn($collection) => $collection->identifier(),
$collections
);
}
if ($collectionSelected === []) {
continue;
}
// construct filter for entities
$entityFilter = null;
if ($filter !== null && $filter !== []) {
$entityFilter = $service->entityListFilter();
foreach ($filter as $attribute => $value) {
$entityFilter->condition($attribute, $value);
}
}
// construct sort for entities
$entitySort = null;
if ($sort !== null && $sort !== []) {
$entitySort = $service->entityListSort();
foreach ($sort as $attribute => $direction) {
$entitySort->condition($attribute, $direction);
}
}
// construct range for entities
$entityRange = null;
if ($range !== null && $range !== [] && isset($range['type'])) {
$entityRange = $service->entityListRange(RangeType::from($range['type']));
if ($entityRange->type() === RangeType::TALLY) {
/** @var IRangeTally $entityRange */
if (isset($range['anchor'])) {
$entityRange->setAnchor(RangeAnchorType::from($range['anchor']));
}
if (isset($range['position'])) {
$entityRange->setPosition($range['position']);
}
if (isset($range['tally'])) {
$entityRange->setTally($range['tally']);
}
}
}
// yield entities for each collection individually
foreach ($collectionSelected as $collectionId) {
yield from $service->entityListStream($collectionId, $entityFilter, $entitySort, $entityRange, null);
}
}
}
}
/**
* Fetch specific messages in bulk
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param EntityIdentifier ...$identifiers Specific message identifiers to fetch
*
* @return array<int,IMessageBase>
*/
public function entityFetchBulk(string $tenantId, ?string $userId, EntityIdentifier ...$identifiers): array {
// group identifiers by provider/service
$groupedIdentifiers = [];
foreach ($identifiers as $identifier) {
$groupedIdentifiers[$identifier->provider()][$identifier->service()][] = $identifier->entity();
}
// retrieve each service and fetch entities
$list = [];
foreach ($groupedIdentifiers as $providerId => $services) {
foreach ($services as $serviceId => $entities) {
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$providerId}:{$serviceId}' not found or is disabled");
}
// retrieve entities and merge into list
$list = array_merge($list, $service->entityFetchBulk(...$entities));
}
}
return $list;
}
/**
* Fetch specific messages as a stream
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param EntityIdentifier ...$identifiers Specific message identifiers to fetch
*
* @return \Generator<string,IMessageBase>
*/
public function entityFetchStream(string $tenantId, ?string $userId, EntityIdentifier ...$identifiers): \Generator {
// group identifiers by provider/service
$groupedIdentifiers = [];
foreach ($identifiers as $identifier) {
$groupedIdentifiers[$identifier->provider()][$identifier->service()][] = $identifier->entity();
}
// retrieve each service and fetch entities
foreach ($groupedIdentifiers as $providerId => $services) {
foreach ($services as $serviceId => $entities) {
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
if ($service->getEnabled() === false) {
throw new InvalidArgumentException("Service '{$providerId}:{$serviceId}' not found or is disabled");
}
// retrieve entities and yield each one
yield from $service->entityFetchStream(...$entities);
}
}
}
public function entityDownload(string $tenantId, string $userId, EntityIdentifier $targetEntity, array|null $targetPart): BinaryResource {
// retrieve service
$service = $this->serviceFetch($tenantId, $userId, $targetEntity->provider(), $targetEntity->service());
// download entity
return $service->entityDownload($targetEntity, $targetPart);
}
/**
* Check if messages exist
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param SourceSelector $sources Message sources with identifiers
*
* @return array<string, array<string|int, array<string|int, array<string|int, bool>>>> Existence map grouped by provider/service/collection
*/
public function entityExtant(string $tenantId, string $userId, SourceSelector $sources): array {
// confirm that sources are provided
if ($sources === null) {
$sources = new SourceSelector([]);
}
// retrieve available providers
$providers = $this->providerList($tenantId, $userId, $sources);
$providersRequested = $sources->identifiers();
$providersUnavailable = array_diff($providersRequested, array_keys($providers));
// initialize response with unavailable providers
$responseData = array_fill_keys($providersUnavailable, false);
// check services, collections, and entities for each available provider
foreach ($providers as $provider) {
// extract services requested for this provider
$serviceSelector = $sources[$provider->identifier()];
$servicesRequested = $serviceSelector->identifiers();
/** @var ServiceBaseInterface[] $servicesAvailable */
$servicesAvailable = $provider->serviceList($tenantId, $userId, $servicesRequested);
$servicesUnavailable = array_diff($servicesRequested, array_keys($servicesAvailable));
// mark unavailable services as false
if ($servicesUnavailable !== []) {
$responseData[$provider->identifier()] = array_fill_keys($servicesUnavailable, false);
}
// check collections and entities for each available service
foreach ($servicesAvailable as $service) {
// omit disabled services
if ($service->getEnabled() === false) {
$responseData[$provider->identifier()][$service->identifier()] = false;
continue;
}
// extract collections requested for this service
$collectionSelector = $serviceSelector[$service->identifier()];
$collectionsRequested = $collectionSelector instanceof CollectionSelector ? $collectionSelector->identifiers() : [];
if ($collectionsRequested === []) {
continue;
}
// check entities for each requested collection
foreach ($collectionsRequested as $collectionId) {
// first check if collection exists
$collectionExists = $service->collectionExtant((string)$collectionId);
if (!$collectionExists) {
// collection doesn't exist, mark as false
$responseData[$provider->identifier()][$service->identifier()][$collectionId] = false;
continue;
}
// extract entity identifiers from collection selector
$entitySelector = $collectionSelector[$collectionId];
// handle both array of entity IDs and boolean true (meaning check if collection exists)
if ($entitySelector instanceof EntitySelector) {
// check specific entities within the collection
$responseData[$provider->identifier()][$service->identifier()][$collectionId] = $service->entityExtant($collectionId, ...$entitySelector->identifiers());
} elseif ($entitySelector === true) {
// just checking if collection exists (already confirmed above)
$responseData[$provider->identifier()][$service->identifier()][$collectionId] = true;
}
}
}
}
return $responseData;
}
/**
* Get message delta/changes
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param SourceSelector $sources Message sources with signatures
*
* @return array<string, array<string|int, array<string|int, array>>> Delta grouped by provider/service/collection
*/
public function entityDelta(string $tenantId, string $userId, SourceSelector $sources): array {
// confirm that sources are provided
if ($sources === null) {
$sources = new SourceSelector([]);
}
// retrieve providers
$providers = $this->providerList($tenantId, $userId, $sources);
$providersRequested = $sources->identifiers();
$providersUnavailable = array_diff($providersRequested, array_keys($providers));
// initialize response with unavailable providers
$responseData = array_fill_keys($providersUnavailable, false);
// iterate through available providers
foreach ($providers as $provider) {
// extract services requested for this provider
$serviceSelector = $sources[$provider->identifier()];
$servicesRequested = $serviceSelector instanceof ServiceSelector ? $serviceSelector->identifiers() : [];
/** @var ServiceBaseInterface[] $services */
$services = $provider->serviceList($tenantId, $userId, $servicesRequested);
$servicesUnavailable = array_diff($servicesRequested, array_keys($services));
if ($servicesUnavailable !== []) {
$responseData[$provider->identifier()] = array_fill_keys($servicesUnavailable, false);
}
// iterate through available services
foreach ($services as $service) {
// omit disabled services
if ($service->getEnabled() === false) {
$responseData[$provider->identifier()][$service->identifier()] = false;
continue;
}
// extract collections requested for this service
$collectionSelector = $serviceSelector[$service->identifier()];
$collectionsRequested = $collectionSelector instanceof CollectionSelector ? $collectionSelector->identifiers() : [];
if ($collectionsRequested === []) {
$responseData[$provider->identifier()][$service->identifier()] = false;
continue;
}
// check delta for each requested collection
foreach ($collectionsRequested as $collection) {
$entitySelector = $collectionSelector[$collection] ?? null;
$responseData[$provider->identifier()][$service->identifier()][$collection] = $service->entityDelta($collection, $entitySelector);
}
}
}
return $responseData;
}
/**
* Patches entities
*
* @since 2026.04.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param MessagePropertiesMutableInterface|array $properties Properties to patch
* @param EntityIdentifier ...$targets Target entities to patch
*
* @return array<string, array{
* disposition: 'patched'|'error',
* error?: string
* }> Results keyed by source entity identifier
*/
public function entityPatch(string $tenantId, string $userId, MessagePropertiesMutableInterface|array $properties, EntityIdentifier ...$targets): array {
$operationOutcome = [];
$targetIdentifiers = new ResourceIdentifiers();
foreach ($targets as $target) {
$targetIdentifiers->add($target);
}
// process targets grouped by provider
foreach ($targetIdentifiers->providers() as $providerId) {
// retrieve provider and validate
$provider = $this->providerFetch($tenantId, $userId, $providerId);
$providerTargets = $targetIdentifiers->byProvider($providerId);
// provider not found, mark all identifiers as failed and continue to next provider
if ($provider === null) {
foreach ($providerTargets->services() as $serviceId) {
$serviceTargets = $providerTargets->byService($serviceId);
foreach ($serviceTargets as $identifier) {
$operationOutcome[$identifier] = ['disposition' => 'error', 'error' => 'provider not found'];
}
}
continue;
}
// process targets grouped by service for this provider
foreach ($providerTargets->services() as $serviceId) {
// extract services requested for this provider
$serviceTargets = $providerTargets->byService($serviceId);
// retrieve and validate service
$service = null;
try {
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
} catch (\Throwable $e) {
$error = "Service $serviceId not found";
}
if ($service instanceof ServiceEntityMutableInterface === false && !isset($error)) {
$error = "Service $serviceId does not support entity mutation";
}
if (!isset($error) && !$service->capable(ServiceEntityMutableInterface::CAPABILITY_ENTITY_PATCH)) {
$error = "Service $serviceId does not support entity patching";
}
// on error, mark all identifiers for this service as failed and continue to next service
if (isset($error)) {
foreach ($serviceTargets as $identifier) {
$operationOutcome[(string)$identifier] = ['disposition' => 'error','error' => $error];
}
continue;
}
/** @var ServiceEntityMutableInterface $service */
if ($properties instanceof MessagePropertiesMutableInterface === false) {
$properties = $service->entityFresh()->getProperties()->jsonDeserialize($properties);
}
$operationOutcome = array_merge($operationOutcome, $service->entityPatch($properties, ...$serviceTargets->all()));
}
}
return $operationOutcome;
}
/**
* Deletes entities
*
* @since 2026.04.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param EntityIdentifier $targets Source entities to delete
*
* @return array<string, array{
* disposition: 'moved'|'deleted'|'error',
* destination: ?CollectionIdentifier,
* mutation: EntityIdentifier
* }> Results keyed by source entity identifier
*/
public function entityDelete(string $tenantId, string $userId, EntityIdentifier ...$targets): array {
$operationOutcome = [];
$targetIdentifiers = new ResourceIdentifiers();
foreach ($targets as $target) {
$targetIdentifiers->add($target);
}
// process targets grouped by provider
foreach ($targetIdentifiers->providers() as $providerId) {
// retrieve provider and validate
$provider = $this->providerFetch($tenantId, $userId, $providerId);
$providerTargets = $targetIdentifiers->byProvider($providerId);
// provider not found, mark all identifiers as failed and continue to next provider
if ($provider === null) {
foreach ($providerTargets->services() as $serviceId) {
$serviceTargets = $providerTargets->byService($serviceId);
foreach ($serviceTargets as $identifier) {
$operationOutcome[$identifier] = ['disposition' => 'error', 'error' => 'provider not found'];
}
}
continue;
}
// process targets grouped by service for this provider
foreach ($providerTargets->services() as $serviceId) {
// extract services requested for this provider
$serviceTargets = $providerTargets->byService($serviceId);
// retrieve and validate service
$service = null;
try {
$service = $this->serviceFetch($tenantId, $userId, $providerId, $serviceId);
} catch (\Throwable $e) {
$error = "Service $serviceId not found";
}
if ($service instanceof ServiceEntityMutableInterface === false && !isset($error)) {
$error = "Service $serviceId does not support entity mutation";
}
if (!isset($error) && !$service->capable(ServiceEntityMutableInterface::CAPABILITY_ENTITY_DELETE)) {
$error = "Service $serviceId does not support entity deletion";
}
// on error, mark all identifiers for this service as failed and continue to next service
if (isset($error)) {
foreach ($serviceTargets as $identifier) {
$operationOutcome[(string)$identifier] = ['disposition' => 'error', 'error' => $error];
}
continue;
}
/** @var ServiceEntityMutableInterface $service */
$operationOutcome = array_merge($operationOutcome, $service->entityDelete(...$serviceTargets->all()));
}
}
return $operationOutcome;
}
/**
* Moves entities to another collection
*
* @since 2025.05.01
*
* @param string $tenantId Tenant identifier
* @param string|null $userId User identifier for context
* @param CollectionIdentifier $target Target collection identifier
* @param EntityIdentifier ...$sources Source entities to move
*
* @return array<string, array{
* disposition: 'moved'|'error',
* destination: ?CollectionIdentifier,
* mutation: EntityIdentifier
* }> Results keyed by source entity identifier
*/
public function entityMove(string $tenantId, string $userId, CollectionIdentifier $target, EntityIdentifier ...$sources): array {
$operationOutcome = [];
// retrieve and validate service
$targetService = null;
try {
$targetService = $this->serviceFetch($tenantId, $userId, $target->provider(), $target->service());
} catch (\Throwable $e) {
// do nothing here, error will be handled in validation below
}
if ($targetService === null || $targetService->getEnabled() === false) {
$error = "Service {$target->service()} not found or is disabled";
}
/** @var ServiceEntityBaseInterface $targetService */
if ($targetService instanceof ServiceEntityMutableInterface === false && !isset($error)) {
$error = "Service {$targetService->identifier()} does not support entity mutation";
}
if (!isset($error) && !$targetService->capable(ServiceEntityMutableInterface::CAPABILITY_ENTITY_MOVE)) {
$error = "Service {$targetService->identifier()} does not support entity moving";
}
// on error, mark all identifiers for this service as failed and continue to next service
if (isset($error)) {
foreach ($sources as $identifier) {
$operationOutcome[(string)$identifier] = ['disposition' => 'error','error' => $error];
}
return $operationOutcome;
}
// validate that sources and target are the same service and group sources by service for processing
$groupedSources = [];
foreach ($sources as $source) {
if ($source->provider() !== $target->provider() || $source->service() !== $target->service()) {
$operationOutcome[(string)$source] = [
'disposition' => 'error',
'error' => "Source '{$source}' and target '{$target}' collections must belong to the same provider and service"
];
continue;
}
$groupedSources[$source->provider()][$source->service()][] = $source;
}
if ($groupedSources === []) {
return $operationOutcome;
}
// perform operation for entities on the same service as the target
$operationOutcome = array_merge(
$operationOutcome,
$targetService->entityMove($target, ...$groupedSources[$target->provider()][$target->service()])
);
// TODO: Handle moving entities across different services/providers by fetching each entity and re-creating it in the target collection,
// then deleting the original if the move is successful. This will require additional logic to handle potential failures and ensure data integrity.
return $operationOutcome;
}
}