ProcessCacheInvalidationEventsCommand::__construct() - Code Metrics - Inspection of "prima versione super cache invalidate" - padosoft/laravel-super-cache-invalidate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#1)

unknown

created 2024-12-04 07:35 UTC

__construct() A

↳ Parent: ProcessCacheInvalidationEventsCommand

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
eloc	2
c	1
b	0
f	0
nc	1
nop	1
dl	0
loc	4
rs	10

<?php

namespace Padosoft\SuperCacheInvalidate\Console;

use Illuminate\Console\Command;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Carbon\Carbon;
use Padosoft\SuperCacheInvalidate\Helpers\SuperCacheInvalidationHelper;

class ProcessCacheInvalidationEventsCommand extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'supercache:process-invalidation
                            {--shard= : The shard number to process}
                            {--priority= : The priority level}
                            {--limit= : The maximum number of events to fetch per batch}
                            {--tag-batch-size= : The number of identifiers to process per invalidation batch}
                            {--connection_name= : The Redis connection name}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = 'Process cache invalidation events for a given shard and priority';

    /**
     * Cache invalidation helper instance.
     */
    protected SuperCacheInvalidationHelper $helper;

    /**
     * Create a new command instance.
     */
    public function __construct(SuperCacheInvalidationHelper $helper)
    {
        parent::__construct();
        $this->helper = $helper;
    }

    private function getStoreFromConnectionName(string $connection_name): ?string
    {
        // Cerca il nome dello store associato alla connessione Redis
        foreach (config('cache.stores') as $storeName => $storeConfig) {
            if (
                isset($storeConfig['driver'], $storeConfig['connection']) &&
                $storeConfig['driver'] === 'redis' &&
                $storeConfig['connection'] === $connection_name
            ) {
                return $storeName;
            }
        }

        return null;
    }

    /**
     * Process cache invalidation events.
     *
     * @param int $shardId      The shard number to process
     * @param int $priority     The priority level
     * @param int $limit        Maximum number of events to fetch per batch
     * @param int $tagBatchSize Number of identifiers to process per batch
     *
     * @throws \Exception
     */
    protected function processEvents(int $shardId, int $priority, int $limit, int $tagBatchSize, string $connection_name): void
    {
        $processingStartTime = now();
        $invalidationWindow = config('super_cache_invalidate.invalidation_window');

        // Fetch a batch of unprocessed events
        $events = DB::table('cache_invalidation_events')
            ->where('processed', '=', 0)
            ->where('shard', '=', $shardId)
            ->where('priority', '=', $priority)
            ->where('event_time', '<', $processingStartTime)
            // Cerco tutte le chiavi/tag da invalidare per questo database redis
            ->where('connection_name', '=', $connection_name)
            ->orderBy('event_time')
            ->limit($limit)
            ->get()
        ;

        if ($events->isEmpty()) {
            // No more events to process
            return;
        }

        // Group events by type and identifier
        $eventsByIdentifier = $events->groupBy(function ($event) {
            return $event->type . ':' . $event->identifier;
        });

        $batchIdentifiers = [];
        $eventsToUpdate = [];
        $counter = 0;

        // Fetch associated identifiers for the events
        $eventIds = $events->pluck('id')->all();

        //retrive associated identifiers related to fetched event id
        // Per le chiavi/tag associati non filtro per connection_name, potrebbero esserci associazioni anche in altri database
        $associations = DB::table('cache_invalidation_event_associations')
            ->whereIn('event_id', $eventIds)
            ->get()
            ->groupBy('event_id')
        ;

        // Prepare list of all identifiers to fetch last invalidation times
        $allIdentifiers = [];
        foreach ($eventsByIdentifier as $key => $eventsGroup) {
            $allIdentifiers[] = $key;
            foreach ($eventsGroup as $event) {
                $eventAssociations = $associations->where('event_id', '=', $event->id);
                foreach ($eventAssociations as $assoc) {
                    $assocKey = $assoc->associated_type . ':' . $assoc->associated_identifier;
                    $allIdentifiers[] = $assocKey;
                }
            }
        }
        // Fetch last invalidation times in bulk
        $lastInvalidationTimes = $this->getLastInvalidationTimes(array_unique($allIdentifiers));

        foreach ($eventsByIdentifier as $key => $eventsGroup) {
            // Extract type and identifier
            [$type, $identifier] = explode(':', $key, 2);

            // Get associated identifiers for the events
            $associatedIdentifiers = [];
            foreach ($eventsGroup as $event) {
                $eventAssociations = $associations->where('event_id', '=', $event->id);
                foreach ($eventAssociations as $assoc) {
                    $assocKey = $assoc->associated_type . ':' . $assoc->associated_identifier;
                    $associatedIdentifiers[$assocKey] = [
                        'type' => $assoc->associated_type,
                        'identifier' => $assoc->associated_identifier,
                        'connection_name' => $assoc->connection_name,
                    ];
                }
            }

            // Build a list of all identifiers to check
            $identifiersToCheck = [$key];
            $identifiersToCheck = array_merge($identifiersToCheck, array_keys($associatedIdentifiers));
            $lastInvalidationTimesSubset = array_intersect_key($lastInvalidationTimes, array_flip($identifiersToCheck));

            $shouldInvalidate = $this->shouldInvalidateMultiple($identifiersToCheck, $lastInvalidationTimesSubset, $invalidationWindow);

            if ($shouldInvalidate) {
                // Proceed to invalidate
                $latestEvent = $eventsGroup->last();

                // Accumulate identifiers and events
                $batchIdentifiers[] = [
                    'type' => $type,
                    'identifier' => $identifier,
                    'event' => $latestEvent,
                    'connection_name' => $connection_name,
                    'associated' => array_values($associatedIdentifiers),
                ];

                // Update last invalidation times for all identifiers
                $this->updateLastInvalidationTimes($identifiersToCheck);

                // Mark all events in the group as processed
                foreach ($eventsGroup as $event) {
                    $eventsToUpdate[] = $event->id;
                }
            } else {
                // Within the invalidation window, skip invalidation
                // Mark all events except the last one as processed
                $eventsToProcess = $eventsGroup->slice(0, -1);
                foreach ($eventsToProcess as $event) {
                    $eventsToUpdate[] = $event->id;
                }
                // The last event remains unprocessed
            }

            $counter++;

            // When we reach the batch size, process the accumulated identifiers
            if ($counter % $tagBatchSize === 0) {
                $this->processBatch($batchIdentifiers, $eventsToUpdate);

                // Reset the accumulators
                $batchIdentifiers = [];
                $eventsToUpdate = [];
            }
        }

        if (empty($batchIdentifiers)) {
            return;
        }

        // Process any remaining identifiers in the batch
        $this->processBatch($batchIdentifiers, $eventsToUpdate);
    }

    /**
     * Fetch last invalidation times for identifiers in bulk.
     *
     * @param  array $identifiers Array of 'type:identifier' strings
     * @return array Associative array of last invalidation times
     */
    protected function getLastInvalidationTimes(array $identifiers): array
    {
        // Extract types and identifiers into tuples
        $tuples = array_map(static function ($key) {
            return explode(':', $key, 2);
        }, $identifiers);

        if (empty($tuples)) {
            return [];
        }

        $records = $this->getRecordsFromDb($tuples);

        // Build associative array
        $lastInvalidationTimes = [];
        foreach ($records as $record) {
            $key = $record->identifier_type . ':' . $record->identifier;
            $lastInvalidationTimes[$key] = Carbon::parse($record->last_invalidated);
        }

        return $lastInvalidationTimes;
    }

    /**
     * Execute Query to get records from DB
     */
    protected function getRecordsFromDb(array $tuples): array
    {
        // Prepare placeholders and parameters
        $placeholders = implode(',', array_fill(0, count($tuples), '(?, ?)'));
        $params = [];
        foreach ($tuples as [$type, $identifier]) {
            $params[] = $type;
            $params[] = $identifier;
        }

        $sql = "SELECT identifier_type,
                        identifier,
                        last_invalidated
                FROM cache_invalidation_timestamps
                WHERE (identifier_type, identifier) IN ($placeholders)
                ";

        return DB::select($sql, $params);
    }

    /**
     * Determine whether to invalidate based on last invalidation times for multiple identifiers.
     *
     * @param  array $identifiers           Array of 'type:identifier' strings
     * @param  array $lastInvalidationTimes Associative array of last invalidation times
     * @param  int   $invalidationWindow    Invalidation window in seconds
     * @return bool  True if should invalidate, false otherwise
     */
    protected function shouldInvalidateMultiple(array $identifiers, array $lastInvalidationTimes, int $invalidationWindow): bool
    {
        $now = now();
        foreach ($identifiers as $key) {
            $lastInvalidated = $lastInvalidationTimes[$key] ?? null;
            if (!$lastInvalidated) {
                continue;
            }
            $elapsed = $now->diffInSeconds($lastInvalidated);
            if ($elapsed < $invalidationWindow) {
                // At least one identifier is within the invalidation window
                return false;
            }
        }

        // All identifiers are outside the invalidation window or have no record
        return true;
    }

    /**
     * Update the last invalidation times for multiple identifiers.
     *
     * @param array $identifiers Array of 'type:identifier' strings
     */
    protected function updateLastInvalidationTimes(array $identifiers): void
    {
        $now = now();

        foreach ($identifiers as $key) {
            [$type, $identifier] = explode(':', $key, 2);
            DB::table('cache_invalidation_timestamps')
                ->updateOrInsert(
                    ['identifier_type' => $type, 'identifier' => $identifier],
                    ['last_invalidated' => $now]
                )
            ;
        }
    }

    /**
     * Process a batch of identifiers and update events.
     *
     * @param array $batchIdentifiers Array of identifiers to invalidate
     * @param array $eventsToUpdate   Array of event IDs to mark as processed
     *
     * @throws \Exception
     */
    protected function processBatch(array $batchIdentifiers, array $eventsToUpdate): void
    {

        // Begin transaction for the batch
        DB::beginTransaction();

        try {
            // Separate keys and tags
            $keys = [];
            $tags = [];

            foreach ($batchIdentifiers as $item) {
                switch ($item['type']) {
                    case 'key':
                        $keys[] = $item['identifier'] . '§' . $item['connection_name'];
                        break;
                    case 'tag':
                        $tags[] = $item['identifier'] . '§' . $item['connection_name'];
                        break;
                }

                if (empty($item['associated'])) {
                    continue;
                }

                // Include associated identifiers
                foreach ($item['associated'] as $assoc) {
                    switch ($assoc['type']) {
                        case 'key':
                            $keys[] = $assoc['identifier'] . '§' . $assoc['connection_name'];
                            break;
                        case 'tag':
                            $tags[] = $assoc['identifier'] . '§' . $assoc['connection_name'];
                            break;
                    }
                }
            }

            // Remove duplicates
            $keys = array_unique($keys);
            $tags = array_unique($tags);

            // Invalidate cache for keys
            if (!empty($keys)) {
                $this->invalidateKeys($keys);
            }

            // Invalidate cache for tags
            if (!empty($tags)) {
                $this->invalidateTags($tags);
            }

            // Mark events as processed
            DB::table('cache_invalidation_events')
                ->whereIn('id', $eventsToUpdate)
                ->update(['processed' => 1])
            ;

            // Commit transaction
            DB::commit();
        } catch (\Exception $e) {
            // Rollback transaction on error
            DB::rollBack();
            throw $e;
        }
    }

    /**
     * Invalidate cache keys.
     *
     * @param array $keys Array of cache keys to invalidate
     */
    protected function invalidateKeys(array $keys): void
    {
        $callback = config('super_cache_invalidate.key_invalidation_callback');


        // Anche in questo caso va fatto un loop perchè le chiavi potrebbero stare in database diversi
        foreach ($keys as $keyAndConnectionName) {
            [$key, $connection_name] = explode('§', $keyAndConnectionName);

            // Metodo del progetto
            if (is_callable($callback)) {
                $callback($key, $connection_name);
                return;
            }

            // oppure di default uso Laravel
            $storeName =  $this->getStoreFromConnectionName($connection_name);

            if ($storeName === null) {
                return;
            }
            Cache::store($storeName)->forget($key);
        }
    }

    /**
     * Invalidate cache tags.
     *
     * @param array $tags Array of cache tags to invalidate
     */
    protected function invalidateTags(array $tags): void
    {
        $callback = config('super_cache_invalidate.tag_invalidation_callback');

        $groupByConnection = [];

        // Anche in questo caso va fatto un loop perchè i tags potrebbero stare in database diversi,
        // ma per ottimizzare possiamo raggruppare le operazioni per connessione
        foreach ($tags as $tagAndConnectionName) {
            // chiave e connessione
            [$key, $connection] = explode('§', $tagAndConnectionName);

            // Aggiungo la chiave alla connessione appropriata
            $groupByConnection[$connection][] = $key;
        }
        if (is_callable($callback)) {
            foreach ($groupByConnection as $connection_name => $arrTags) {
                $callback($arrTags, $connection_name);
            }
            return;
        }
        foreach ($groupByConnection as $connection_name => $arrTags) {
            $storeName =  $this->getStoreFromConnectionName($connection_name);
            if ($storeName === null) {
                return;
            }
            Cache::store($storeName)->tags($arrTags)->flush();
        }
    }

    /**
     * Execute the console command.
     */
    public function handle(): void
    {
        $shardId = (int) $this->option('shard');
        $priority = (int) $this->option('priority');
        $limit = $this->option('limit') ?? config('super_cache_invalidate.processing_limit');
        $limit = (int)$limit;
        $tagBatchSize = $this->option('tag-batch-size') ?? config('super_cache_invalidate.tag_batch_size');
        $tagBatchSize = (int)$tagBatchSize;
        $lockTimeout = (int) config('super_cache_invalidate.lock_timeout');
        $connection_name = $this->option('connection_name') ?? config('super_cache_invalidate.default_connection_name');
        /*
        if ($shardId === 0 && $priority === 0) {
            $this->error('Shard and priority are required and must be non-zero integers.');

            return;
        }
        */
        $lockValue = $this->helper->acquireShardLock($shardId, $lockTimeout, $connection_name);

        if (!$lockValue) {
            return;
        }

        try {
            $this->processEvents($shardId, $priority, $limit, $tagBatchSize, $connection_name);
        } catch (\Exception $e) {
            $this->error('An error occourred in ' . __METHOD__ . ': ' . $e->getTraceAsString());
        } finally {
            $this->helper->releaseShardLock($shardId, $lockValue, $connection_name);
        }
    }
}


1			<?php
2
3			namespace Padosoft\SuperCacheInvalidate\Console;
4
5			use Illuminate\Console\Command;
6			use Illuminate\Support\Facades\Cache;
7			use Illuminate\Support\Facades\DB;
8			use Carbon\Carbon;
9			use Padosoft\SuperCacheInvalidate\Helpers\SuperCacheInvalidationHelper;
10
11			class ProcessCacheInvalidationEventsCommand extends Command
12			{
13			/**
14			* The name and signature of the console command.
15			*
16			* @var string
17			*/
18			protected $signature = 'supercache:process-invalidation
19			{--shard= : The shard number to process}
20			{--priority= : The priority level}
21			{--limit= : The maximum number of events to fetch per batch}
22			{--tag-batch-size= : The number of identifiers to process per invalidation batch}
23			{--connection_name= : The Redis connection name}';
24
25			/**
26			* The console command description.
27			*
28			* @var string
29			*/
30			protected $description = 'Process cache invalidation events for a given shard and priority';
31
32			/**
33			* Cache invalidation helper instance.
34			*/
35			protected SuperCacheInvalidationHelper $helper;
36
37			/**
38			* Create a new command instance.
39			*/
40			public function __construct(SuperCacheInvalidationHelper $helper)
41			{
42			parent::__construct();
43			$this->helper = $helper;
44			}
45
46			private function getStoreFromConnectionName(string $connection_name): ?string
47			{
48			// Cerca il nome dello store associato alla connessione Redis
49			foreach (config('cache.stores') as $storeName => $storeConfig) {
50			if (
51			isset($storeConfig['driver'], $storeConfig['connection']) &&
52			$storeConfig['driver'] === 'redis' &&
53			$storeConfig['connection'] === $connection_name
54			) {
55			return $storeName;
56			}
57			}
58
59			return null;
60			}
61
62			/**
63			* Process cache invalidation events.
64			*
65			* @param int $shardId The shard number to process
66			* @param int $priority The priority level
67			* @param int $limit Maximum number of events to fetch per batch
68			* @param int $tagBatchSize Number of identifiers to process per batch
69			*
70			* @throws \Exception
71			*/
72			protected function processEvents(int $shardId, int $priority, int $limit, int $tagBatchSize, string $connection_name): void
73			{
74			$processingStartTime = now();
75			$invalidationWindow = config('super_cache_invalidate.invalidation_window');
76
77			// Fetch a batch of unprocessed events
78			$events = DB::table('cache_invalidation_events')
79			->where('processed', '=', 0)
80			->where('shard', '=', $shardId)
81			->where('priority', '=', $priority)
82			->where('event_time', '<', $processingStartTime)
83			// Cerco tutte le chiavi/tag da invalidare per questo database redis
84			->where('connection_name', '=', $connection_name)
85			->orderBy('event_time')
86			->limit($limit)
87			->get()
88			;
89
90			if ($events->isEmpty()) {
91			// No more events to process
92			return;
93			}
94
95			// Group events by type and identifier
96			$eventsByIdentifier = $events->groupBy(function ($event) {
97			return $event->type . ':' . $event->identifier;
98			});
99
100			$batchIdentifiers = [];
101			$eventsToUpdate = [];
102			$counter = 0;
103
104			// Fetch associated identifiers for the events
105			$eventIds = $events->pluck('id')->all();
106
107			//retrive associated identifiers related to fetched event id
108			// Per le chiavi/tag associati non filtro per connection_name, potrebbero esserci associazioni anche in altri database
109			$associations = DB::table('cache_invalidation_event_associations')
110			->whereIn('event_id', $eventIds)
111			->get()
112			->groupBy('event_id')
113			;
114
115			// Prepare list of all identifiers to fetch last invalidation times
116			$allIdentifiers = [];
117			foreach ($eventsByIdentifier as $key => $eventsGroup) {
118			$allIdentifiers[] = $key;
119			foreach ($eventsGroup as $event) {
120			$eventAssociations = $associations->where('event_id', '=', $event->id);
121			foreach ($eventAssociations as $assoc) {
122			$assocKey = $assoc->associated_type . ':' . $assoc->associated_identifier;
123			$allIdentifiers[] = $assocKey;
124			}
125			}
126			}
127			// Fetch last invalidation times in bulk
128			$lastInvalidationTimes = $this->getLastInvalidationTimes(array_unique($allIdentifiers));
129
130			foreach ($eventsByIdentifier as $key => $eventsGroup) {
131			// Extract type and identifier
132			[$type, $identifier] = explode(':', $key, 2);
133
134			// Get associated identifiers for the events
135			$associatedIdentifiers = [];
136			foreach ($eventsGroup as $event) {
137			$eventAssociations = $associations->where('event_id', '=', $event->id);
138			foreach ($eventAssociations as $assoc) {
139			$assocKey = $assoc->associated_type . ':' . $assoc->associated_identifier;
140			$associatedIdentifiers[$assocKey] = [
141			'type' => $assoc->associated_type,
142			'identifier' => $assoc->associated_identifier,
143			'connection_name' => $assoc->connection_name,
144			];
145			}
146			}
147
148			// Build a list of all identifiers to check
149			$identifiersToCheck = [$key];
150			$identifiersToCheck = array_merge($identifiersToCheck, array_keys($associatedIdentifiers));
151			$lastInvalidationTimesSubset = array_intersect_key($lastInvalidationTimes, array_flip($identifiersToCheck));
152
153			$shouldInvalidate = $this->shouldInvalidateMultiple($identifiersToCheck, $lastInvalidationTimesSubset, $invalidationWindow);
154
155			if ($shouldInvalidate) {
156			// Proceed to invalidate
157			$latestEvent = $eventsGroup->last();
158
159			// Accumulate identifiers and events
160			$batchIdentifiers[] = [
161			'type' => $type,
162			'identifier' => $identifier,
163			'event' => $latestEvent,
164			'connection_name' => $connection_name,
165			'associated' => array_values($associatedIdentifiers),
166			];
167
168			// Update last invalidation times for all identifiers
169			$this->updateLastInvalidationTimes($identifiersToCheck);
170
171			// Mark all events in the group as processed
172			foreach ($eventsGroup as $event) {
173			$eventsToUpdate[] = $event->id;
174			}
175			} else {
176			// Within the invalidation window, skip invalidation
177			// Mark all events except the last one as processed
178			$eventsToProcess = $eventsGroup->slice(0, -1);
179			foreach ($eventsToProcess as $event) {
180			$eventsToUpdate[] = $event->id;
181			}
182			// The last event remains unprocessed
183			}
184
185			$counter++;
186
187			// When we reach the batch size, process the accumulated identifiers
188			if ($counter % $tagBatchSize === 0) {
189			$this->processBatch($batchIdentifiers, $eventsToUpdate);
190
191			// Reset the accumulators
192			$batchIdentifiers = [];
193			$eventsToUpdate = [];
194			}
195			}
196
197			if (empty($batchIdentifiers)) {
198			return;
199			}
200
201			// Process any remaining identifiers in the batch
202			$this->processBatch($batchIdentifiers, $eventsToUpdate);
203			}
204
205			/**
206			* Fetch last invalidation times for identifiers in bulk.
207			*
208			* @param array $identifiers Array of 'type:identifier' strings
209			* @return array Associative array of last invalidation times
210			*/
211			protected function getLastInvalidationTimes(array $identifiers): array
212			{
213			// Extract types and identifiers into tuples
214			$tuples = array_map(static function ($key) {
215			return explode(':', $key, 2);
216			}, $identifiers);
217
218			if (empty($tuples)) {
219			return [];
220			}
221
222			$records = $this->getRecordsFromDb($tuples);
223
224			// Build associative array
225			$lastInvalidationTimes = [];
226			foreach ($records as $record) {
227			$key = $record->identifier_type . ':' . $record->identifier;
228			$lastInvalidationTimes[$key] = Carbon::parse($record->last_invalidated);
229			}
230
231			return $lastInvalidationTimes;
232			}
233
234			/**
235			* Execute Query to get records from DB
236			*/
237			protected function getRecordsFromDb(array $tuples): array
238			{
239			// Prepare placeholders and parameters
240			$placeholders = implode(',', array_fill(0, count($tuples), '(?, ?)'));
241			$params = [];
242			foreach ($tuples as [$type, $identifier]) {
243			$params[] = $type;
244			$params[] = $identifier;
245			}
246
247			$sql = "SELECT identifier_type,
248			identifier,
249			last_invalidated
250			FROM cache_invalidation_timestamps
251			WHERE (identifier_type, identifier) IN ($placeholders)
252			";
253
254			return DB::select($sql, $params);
255			}
256
257			/**
258			* Determine whether to invalidate based on last invalidation times for multiple identifiers.
259			*
260			* @param array $identifiers Array of 'type:identifier' strings
261			* @param array $lastInvalidationTimes Associative array of last invalidation times
262			* @param int $invalidationWindow Invalidation window in seconds
263			* @return bool True if should invalidate, false otherwise
264			*/
265			protected function shouldInvalidateMultiple(array $identifiers, array $lastInvalidationTimes, int $invalidationWindow): bool
266			{
267			$now = now();
268			foreach ($identifiers as $key) {
269			$lastInvalidated = $lastInvalidationTimes[$key] ?? null;
270			if (!$lastInvalidated) {
271			continue;
272			}
273			$elapsed = $now->diffInSeconds($lastInvalidated);
274			if ($elapsed < $invalidationWindow) {
275			// At least one identifier is within the invalidation window
276			return false;
277			}
278			}
279
280			// All identifiers are outside the invalidation window or have no record
281			return true;
282			}
283
284			/**
285			* Update the last invalidation times for multiple identifiers.
286			*
287			* @param array $identifiers Array of 'type:identifier' strings
288			*/
289			protected function updateLastInvalidationTimes(array $identifiers): void
290			{
291			$now = now();
292
293			foreach ($identifiers as $key) {
294			[$type, $identifier] = explode(':', $key, 2);
295			DB::table('cache_invalidation_timestamps')
296			->updateOrInsert(
297			['identifier_type' => $type, 'identifier' => $identifier],
298			['last_invalidated' => $now]
299			)
300			;
301			}
302			}
303
304			/**
305			* Process a batch of identifiers and update events.
306			*
307			* @param array $batchIdentifiers Array of identifiers to invalidate
308			* @param array $eventsToUpdate Array of event IDs to mark as processed
309			*
310			* @throws \Exception
311			*/
312			protected function processBatch(array $batchIdentifiers, array $eventsToUpdate): void
313			{
314
315			// Begin transaction for the batch
316			DB::beginTransaction();
317
318			try {
319			// Separate keys and tags
320			$keys = [];
321			$tags = [];
322
323			foreach ($batchIdentifiers as $item) {
324			switch ($item['type']) {
325			case 'key':
326			$keys[] = $item['identifier'] . '§' . $item['connection_name'];
327			break;
328			case 'tag':
329			$tags[] = $item['identifier'] . '§' . $item['connection_name'];
330			break;
331			}
332
333			if (empty($item['associated'])) {
334			continue;
335			}
336
337			// Include associated identifiers
338			foreach ($item['associated'] as $assoc) {
339			switch ($assoc['type']) {
340			case 'key':
341			$keys[] = $assoc['identifier'] . '§' . $assoc['connection_name'];
342			break;
343			case 'tag':
344			$tags[] = $assoc['identifier'] . '§' . $assoc['connection_name'];
345			break;
346			}
347			}
348			}
349
350			// Remove duplicates
351			$keys = array_unique($keys);
352			$tags = array_unique($tags);
353
354			// Invalidate cache for keys
355			if (!empty($keys)) {
356			$this->invalidateKeys($keys);
357			}
358
359			// Invalidate cache for tags
360			if (!empty($tags)) {
361			$this->invalidateTags($tags);
362			}
363
364			// Mark events as processed
365			DB::table('cache_invalidation_events')
366			->whereIn('id', $eventsToUpdate)
367			->update(['processed' => 1])
368			;
369
370			// Commit transaction
371			DB::commit();
372			} catch (\Exception $e) {
373			// Rollback transaction on error
374			DB::rollBack();
375			throw $e;
376			}
377			}
378
379			/**
380			* Invalidate cache keys.
381			*
382			* @param array $keys Array of cache keys to invalidate
383			*/
384			protected function invalidateKeys(array $keys): void
385			{
386			$callback = config('super_cache_invalidate.key_invalidation_callback');
387
388
389			// Anche in questo caso va fatto un loop perchè le chiavi potrebbero stare in database diversi
390			foreach ($keys as $keyAndConnectionName) {
391			[$key, $connection_name] = explode('§', $keyAndConnectionName);
392
393			// Metodo del progetto
394			if (is_callable($callback)) {
395			$callback($key, $connection_name);
396			return;
397			}
398
399			// oppure di default uso Laravel
400			$storeName = $this->getStoreFromConnectionName($connection_name);
401
402			if ($storeName === null) {
403			return;
404			}
405			Cache::store($storeName)->forget($key);
406			}
407			}
408
409			/**
410			* Invalidate cache tags.
411			*
412			* @param array $tags Array of cache tags to invalidate
413			*/
414			protected function invalidateTags(array $tags): void
415			{
416			$callback = config('super_cache_invalidate.tag_invalidation_callback');
417
418			$groupByConnection = [];
419
420			// Anche in questo caso va fatto un loop perchè i tags potrebbero stare in database diversi,
421			// ma per ottimizzare possiamo raggruppare le operazioni per connessione
422			foreach ($tags as $tagAndConnectionName) {
423			// chiave e connessione
424			[$key, $connection] = explode('§', $tagAndConnectionName);
425
426			// Aggiungo la chiave alla connessione appropriata
427			$groupByConnection[$connection][] = $key;
428			}
429			if (is_callable($callback)) {
430			foreach ($groupByConnection as $connection_name => $arrTags) {
431			$callback($arrTags, $connection_name);
432			}
433			return;
434			}
435			foreach ($groupByConnection as $connection_name => $arrTags) {
436			$storeName = $this->getStoreFromConnectionName($connection_name);
437			if ($storeName === null) {
438			return;
439			}
440			Cache::store($storeName)->tags($arrTags)->flush();
441			}
442			}
443
444			/**
445			* Execute the console command.
446			*/
447			public function handle(): void
448			{
449			$shardId = (int) $this->option('shard');
450			$priority = (int) $this->option('priority');
451			$limit = $this->option('limit') ?? config('super_cache_invalidate.processing_limit');
452			$limit = (int)$limit;
453			$tagBatchSize = $this->option('tag-batch-size') ?? config('super_cache_invalidate.tag_batch_size');
454			$tagBatchSize = (int)$tagBatchSize;
455			$lockTimeout = (int) config('super_cache_invalidate.lock_timeout');
456			$connection_name = $this->option('connection_name') ?? config('super_cache_invalidate.default_connection_name');
457			/*
458			if ($shardId === 0 && $priority === 0) {
459			$this->error('Shard and priority are required and must be non-zero integers.');
460
461			return;
462			}
463			*/
464			$lockValue = $this->helper->acquireShardLock($shardId, $lockTimeout, $connection_name);
465
466			if (!$lockValue) {
467			return;
468			}
469
470			try {
471			$this->processEvents($shardId, $priority, $limit, $tagBatchSize, $connection_name);
472			} catch (\Exception $e) {
473			$this->error('An error occourred in ' . __METHOD__ . ': ' . $e->getTraceAsString());
474			} finally {
475			$this->helper->releaseShardLock($shardId, $lockValue, $connection_name);
476			}
477			}
478			}
479

padosoft / laravel-super-cache-invalidate

Pull Request — main (#1)

__construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like