ProcessCacheInvalidationEventsCommand::handle() - 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 08:15 UTC

ProcessCacheInvalidationEventsCommand::handle() A

↳ Parent: ProcessCacheInvalidationEventsCommand

Complexity

Conditions	3
Paths	5

Size

Total Lines	29
Code Lines	17

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	3
eloc	17
c	1
b	0
f	0
nc	5
nop	0
dl	0
loc	29
rs	9.7

<?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;
    }

    protected function getCache_invalidation_eventsPartitionName(int $shardId, int $priority, int $processed, Carbon $processingStartTime): string
    {
        if ($processed === 0) {
            return "p_unprocessed_s{$shardId}_p{$priority}";
        }
        $year = $processingStartTime->year;
        $week = $processingStartTime->weekOfYear;
        return "p_s{$shardId}_p{$priority}_{$year}w{$week}";
    }

    /**
     * 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
        $partitionCache_invalidation_events = $this->getCache_invalidation_eventsPartitionName($shardId, $priority, 0, $processingStartTime);
        $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()
        ;

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

padosoft / laravel-super-cache-invalidate

Pull Request — main (#3)

ProcessCacheInvalidationEventsCommand::handle() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like