metadata = $metadata; $this->entityManager = $entityManager; $this->config = $config; } public function process(): void { if (!$this->config->get('cleanupSubscribers')) { return; } /** @var string[] $scopeList */ $scopeList = array_keys($this->metadata->get(['scopes']) ?? []); /** @var string[] $scopeList */ $scopeList = array_values(array_filter( $scopeList, fn ($item) => (bool) $this->metadata->get(['scopes', $item, 'stream']) )); foreach ($scopeList as $scope) { $this->processEntityType($scope); } } private function processEntityType(string $entityType): void { /** @var ?array $data */ $data = $this->metadata->get(['streamDefs', $entityType, 'subscribersCleanup']); if (!($data['enabled'] ?? false)) { return; } /** @var string $dateField */ $dateField = $data['dateField'] ?? 'createdAt'; /** @var ?string[] $statusList */ $statusList = $data['statusList'] ?? null; /** @var ?string $statusField */ $statusField = $this->metadata->get(['scopes', $entityType, 'statusField']); if ($statusList === null || $statusField === null) { return; } /** @var string $period */ $period = $this->metadata->get(['streamDefs', $entityType, 'subscribersCleanup', 'period']) ?? $this->config->get('cleanupSubscribersPeriod') ?? self::PERIOD; $before = DateTime::createNow()->modify('-' . $period); $query = $this->entityManager ->getQueryBuilder() ->delete() ->from(Subscription::ENTITY_TYPE, 'subscription') ->join( $entityType, 'entity', Cond::equal( Cond::column('entity.id'), Cond::column('entityId') ) ) ->where( Cond::and( Cond::equal( Cond::column('entityType'), $entityType ), Cond::less( Cond::column('entity.' . $dateField), $before->getString() ), Cond::in( Cond::column('entity.' . $statusField), $statusList ) ) ) ->build(); $this->entityManager->getQueryExecutor()->execute($query); } }