ProcessCacheInvalidationEventsCommand::shouldInvalidateMultiple() - Code Metrics - Inspection of "use partition" - padosoft/laravel-super-cache-invalidate - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — main (#3)

unknown

created 2024-12-31 15:51 UTC

shouldInvalidateMultiple() A

↳ Parent: ProcessCacheInvalidationEventsCommand

Complexity

Conditions	4
Paths	4

Size

Total Lines	17
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	4
eloc	9
c	2
b	0
f	0
nc	4
nop	3
dl	0
loc	17
rs	9.9666

<?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
     * @throws \Throwable
     */
    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
        $partitionCache_invalidation_events = $this->helper->getCacheInvalidationEventsPartitionName($shardId, $priority);

        $events = DB::table(DB::raw("`cache_invalidation_events` PARTITION ({$partitionCache_invalidation_events})"))
            //->from(DB::raw("`{$this->from}` PARTITION ({$partitionsString})"))
            ->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()
        ;

        //ds($partitionCache_invalidation_events . ' -> Shard (' . $shardId . ') Priority (' . $priority . ') Record = ' . $events->count());
        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
        // TODO JB 31/12/2024: per adesso commentato, da riattivare quando tutto funziona alla perfezione usando la partizione
        $associations = collect();
        /*
        $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, $shardId, $priority);

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

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

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

    /**
     * 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 \Throwable
     */
    protected function processBatch(array $batchIdentifiers, array $eventsToUpdate, int $shard, int $priority): void
    {
        $maxAttempts = 5;
        $attempts = 0;
        $updatedOk = false;

        // 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);
        }

        while ($attempts < $maxAttempts && !$updatedOk) {
            // Begin transaction for the batch
            DB::beginTransaction();

            try {
                // Mark events as processed
                $partitionCache_invalidation_events = $this->helper->getCacheInvalidationEventsPartitionName($shard, $priority);
                DB::table(DB::raw("`cache_invalidation_events` PARTITION ({$partitionCache_invalidation_events})"))
                    ->whereIn('id', $eventsToUpdate)
                    ->update(['processed' => 1])
                ;

                // Commit transaction
                DB::commit();
                $updatedOk = true;
            } catch (\Throwable $e) {
                // Rollback transaction on error
                DB::rollBack();
                $attempts++;
                $this->warn(now()->toDateTimeString() . ": Tentativo $attempts di $maxAttempts: " . $e->getMessage());
                // Logica per gestire i tentativi falliti
                if ($attempts >= $maxAttempts) {
                    // Salta il record dopo il numero massimo di tentativi
                    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, $priority, $lockTimeout, $connection_name);

        if (!$lockValue) {
            return;
        }

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

padosoft / laravel-super-cache-invalidate

Pull Request — main (#3)

shouldInvalidateMultiple() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like