Add new index commands

app/Console/Commands/NntmuxCheckIndex.php

@@ -0,0 +1,149 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Exception;
+
+class NntmuxCheckIndex extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'nntmux:check-index
+                                       {--manticore : Check ManticoreSearch}
+                                       {--elastic : Check ElasticSearch}
+                                       {--releases : Check the releases index}
+                                       {--predb : Check the predb index}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Check if data exists in search indexes';
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(): int
+    {
+        $engine = $this->getSelectedEngine();
+        $index = $this->getSelectedIndex();
+
+        if (!$engine || !$index) {
+            $this->error('You must specify both an engine (--manticore or --elastic) and an index (--releases or --predb).');
+            return Command::FAILURE;
+        }
+
+        $this->info("Checking {$engine} {$index} index...");
+
+        try {
+            if ($engine === 'elastic') {
+                $this->checkElasticIndex($index);
+            } else {
+                $this->checkManticoreIndex($index);
+            }
+
+            return Command::SUCCESS;
+        } catch (Exception $e) {
+            $this->error("Error checking index: {$e->getMessage()}");
+            return Command::FAILURE;
+        }
+    }
+
+    /**
+     * Check ElasticSearch index
+     */
+    private function checkElasticIndex(string $index): void
+    {
+        try {
+            // Check if index exists
+            $exists = \Elasticsearch::indices()->exists(['index' => $index]);
+
+            if (!$exists) {
+                $this->error("ElasticSearch index '{$index}' does not exist.");
+                return;
+            }
+
+            $this->info("ElasticSearch index '{$index}' exists.");
+
+            // Get index stats
+            $stats = \Elasticsearch::indices()->stats(['index' => $index]);
+            $docCount = $stats['indices'][$index]['total']['docs']['count'] ?? 0;
+            $storeSize = $stats['indices'][$index]['total']['store']['size_in_bytes'] ?? 0;
+
+            $this->info("Document count: " . number_format($docCount));
+            $this->info("Index size: " . $this->formatBytes($storeSize));
+
+            // Get a sample document
+            if ($docCount > 0) {
+                $sample = \Elasticsearch::search([
+                    'index' => $index,
+                    'size' => 1,
+                    'body' => [
+                        'query' => ['match_all' => (object)[]]
+                    ]
+                ]);
+
+                if (isset($sample['hits']['hits'][0])) {
+                    $this->info("Sample document:");
+                    $this->info(json_encode($sample['hits']['hits'][0]['_source'], JSON_PRETTY_PRINT));
+                }
+            }
+
+        } catch (Exception $e) {
+            $this->error("Error checking ElasticSearch index: {$e->getMessage()}");
+        }
+    }
+
+    /**
+     * Check ManticoreSearch index
+     */
+    private function checkManticoreIndex(string $index): void
+    {
+        try {
+            $indexName = $index === 'releases' ? 'releases_rt' : 'predb_rt';
+
+            // This is a basic check - you may need to adjust based on your ManticoreSearch setup
+            $this->info("Checking ManticoreSearch index '{$indexName}'...");
+            $this->warn("ManticoreSearch index checking not fully implemented yet.");
+
+        } catch (Exception $e) {
+            $this->error("Error checking ManticoreSearch index: {$e->getMessage()}");
+        }
+    }
+
+    /**
+     * Format bytes to human readable format
+     */
+    private function formatBytes(int $bytes): string
+    {
+        $units = ['B', 'KB', 'MB', 'GB', 'TB'];
+        $bytes = max($bytes, 0);
+        $pow = floor(($bytes ? log($bytes) : 0) / log(1024));
+        $pow = min($pow, count($units) - 1);
+
+        $bytes /= pow(1024, $pow);
+
+        return round($bytes, 2) . ' ' . $units[$pow];
+    }
+
+    /**
+     * Get selected engine
+     */
+    private function getSelectedEngine(): ?string
+    {
+        return $this->option('manticore') ? 'manticore' : ($this->option('elastic') ? 'elastic' : null);
+    }
+
+    /**
+     * Get selected index
+     */
+    private function getSelectedIndex(): ?string
+    {
+        return $this->option('releases') ? 'releases' : ($this->option('predb') ? 'predb' : null);
+    }
+}

app/Console/Commands/NntmuxOffsetPopulate.php

@@ -0,0 +1,415 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Models\Predb;
+use App\Models\Release;
+use Blacklight\ManticoreSearch;
+use Exception;
+use Illuminate\Console\Command;
+use Illuminate\Support\Arr;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Process;
+
+class NntmuxOffsetPopulate extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'nntmux:offset-populate
+                                       {--manticore : Use ManticoreSearch}
+                                       {--elastic : Use ElasticSearch}
+                                       {--releases : Populates the releases index}
+                                       {--predb : Populates the predb index}
+                                       {--parallel=8 : Number of parallel processes}
+                                       {--batch-size=10000 : Batch size for bulk operations}
+                                       {--disable-keys : Disable database keys during population}
+                                       {--memory-limit=4G : Memory limit for each process}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Populate search indexes using offset-based parallel processing for maximum performance';
+
+    private const SUPPORTED_ENGINES = ['manticore', 'elastic'];
+    private const SUPPORTED_INDEXES = ['releases', 'predb'];
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(): int
+    {
+        $engine = $this->getSelectedEngine();
+        $index = $this->getSelectedIndex();
+
+        if (!$engine || !$index) {
+            $this->error('You must specify both an engine (--manticore or --elastic) and an index (--releases or --predb).');
+            return Command::FAILURE;
+        }
+
+        $this->info("Starting offset-based parallel {$engine} {$index} population...");
+        $startTime = microtime(true);
+
+        $result = $this->populateIndexParallel($engine, $index);
+
+        $executionTime = round(microtime(true) - $startTime, 2);
+        if ($result === Command::SUCCESS) {
+            $this->info("Offset-based parallel population completed in {$executionTime} seconds.");
+        }
+
+        return $result;
+    }
+
+    /**
+     * Populate index using offset-based parallel processing
+     */
+    private function populateIndexParallel(string $engine, string $index): int
+    {
+        $total = $this->getTotalRecords($index);
+        if (!$total) {
+            $this->warn("{$index} table is empty. Nothing to do.");
+            return Command::SUCCESS;
+        }
+
+        $parallelProcesses = $this->option('parallel');
+        $recordsPerProcess = ceil($total / $parallelProcesses);
+
+        $this->info(sprintf(
+            "Processing %s records with %d parallel processes, ~%d records per process",
+            number_format($total),
+            $parallelProcesses,
+            $recordsPerProcess
+        ));
+
+        // Clear the index first
+        $this->clearIndex($engine, $index);
+
+        // Create offset-based ranges for parallel processing
+        $ranges = $this->createOffsetRanges($total, $parallelProcesses);
+        $processes = [];
+
+        // Start parallel processes
+        foreach ($ranges as $i => $range) {
+            $command = $this->buildWorkerCommand($engine, $index, $range['offset'], $range['limit'], $i);
+            $this->info("Starting worker process {$i}: processing {$range['limit']} records from offset {$range['offset']}");
+
+            $process = Process::start($command);
+            $processes[] = [
+                'process' => $process,
+                'id' => $i,
+                'range' => $range
+            ];
+        }
+
+        // Monitor processes
+        $this->monitorProcesses($processes);
+
+        // Verify final count
+        $this->verifyIndexPopulation($engine, $index, $total);
+
+        $this->info('All parallel processes completed successfully!');
+        return Command::SUCCESS;
+    }
+
+    /**
+     * Create offset-based ranges for parallel execution
+     */
+    private function createOffsetRanges(int $total, int $processes): array
+    {
+        $ranges = [];
+        $recordsPerProcess = ceil($total / $processes);
+
+        for ($i = 0; $i < $processes; $i++) {
+            $offset = $i * $recordsPerProcess;
+            $limit = min($recordsPerProcess, $total - $offset);
+
+            if ($limit > 0) {
+                $ranges[] = [
+                    'offset' => $offset,
+                    'limit' => $limit,
+                    'worker_id' => $i
+                ];
+            }
+        }
+
+        return $ranges;
+    }
+
+    /**
+     * Build worker command for offset-based parallel processing
+     */
+    private function buildWorkerCommand(string $engine, string $index, int $offset, int $limit, int $workerId): string
+    {
+        $memoryLimit = $this->option('memory-limit') ?? '4G';
+        $batchSize = $this->option('batch-size');
+
+        $artisanPath = base_path('artisan');
+
+        $options = [
+            "--{$engine}",
+            "--{$index}",
+            "--offset={$offset}",
+            "--limit={$limit}",
+            "--worker-id={$workerId}",
+            "--batch-size={$batchSize}",
+            "--memory-limit={$memoryLimit}"
+        ];
+
+        if ($this->option('disable-keys')) {
+            $options[] = '--disable-keys';
+        }
+
+        return sprintf(
+            'php -d memory_limit=%s "%s" nntmux:offset-worker %s 2>&1',
+            $memoryLimit,
+            $artisanPath,
+            implode(' ', $options)
+        );
+    }
+
+    /**
+     * Monitor parallel processes
+     */
+    private function monitorProcesses(array $processes): void
+    {
+        $bar = $this->output->createProgressBar(count($processes));
+        $bar->setFormat('verbose');
+        $bar->start();
+
+        $completed = 0;
+        $completedProcesses = [];
+        $failedProcesses = [];
+
+        while ($completed < count($processes)) {
+            foreach ($processes as $index => $processInfo) {
+                $process = $processInfo['process'];
+                $processId = $processInfo['id'];
+
+                if (in_array($processId, $completedProcesses)) {
+                    continue;
+                }
+
+                if (!$process->running()) {
+                    $completedProcesses[] = $processId;
+                    $completed++;
+                    $bar->advance();
+
+                    try {
+                        $result = $process->wait();
+                        $exitCode = $result->exitCode();
+
+                        if ($exitCode === 0) {
+                            $this->info("Worker {$processId} completed successfully");
+                        } else {
+                            $failedProcesses[] = $processId;
+                            $this->error("Worker {$processId} failed with exit code: {$exitCode}");
+
+                            $output = $result->output();
+                            $errorOutput = $result->errorOutput();
+
+                            if ($output) {
+                                $this->error("Worker {$processId} output: " . substr($output, -500));
+                            }
+                            if ($errorOutput) {
+                                $this->error("Worker {$processId} error: " . substr($errorOutput, -500));
+                            }
+                        }
+                    } catch (Exception $e) {
+                        $failedProcesses[] = $processId;
+                        $this->error("Worker {$processId} exception: {$e->getMessage()}");
+                    }
+                }
+            }
+
+            usleep(500000); // 0.5 second delay
+        }
+
+        $bar->finish();
+        $this->newLine();
+
+        if (!empty($failedProcesses)) {
+            $this->error("Failed workers: " . implode(', ', $failedProcesses));
+        }
+    }
+
+    /**
+     * Verify index population
+     */
+    private function verifyIndexPopulation(string $engine, string $index, int $expectedTotal): void
+    {
+        $this->info("Verifying index population...");
+
+        if ($engine === 'elastic') {
+            try {
+                // Wait a moment for ElasticSearch to refresh
+                sleep(2);
+
+                $stats = \Elasticsearch::indices()->stats(['index' => $index]);
+                $actualCount = $stats['indices'][$index]['total']['docs']['count'] ?? 0;
+
+                $this->info("Expected: " . number_format($expectedTotal) . " records");
+                $this->info("Actual: " . number_format($actualCount) . " records");
+
+                if ($actualCount >= $expectedTotal * 0.95) { // Allow for 5% tolerance
+                    $this->info("✓ Index population successful!");
+                } else {
+                    $this->warn("⚠ Index population may be incomplete");
+                }
+            } catch (Exception $e) {
+                $this->warn("Could not verify index population: {$e->getMessage()}");
+            }
+        } else {
+            $this->info("ManticoreSearch index verification not implemented yet");
+        }
+    }
+
+    /**
+     * Clear the search index
+     */
+    private function clearIndex(string $engine, string $index): void
+    {
+        if ($engine === 'manticore') {
+            $manticore = new ManticoreSearch;
+            $indexName = $index === 'releases' ? 'releases_rt' : 'predb_rt';
+            $manticore->truncateRTIndex(Arr::wrap($indexName));
+            $this->info("Truncated ManticoreSearch index: {$indexName}");
+        } else {
+            // For ElasticSearch, just clear the data instead of recreating the index
+            try {
+                $exists = \Elasticsearch::indices()->exists(['index' => $index]);
+
+                if ($exists) {
+                    // Get current document count
+                    $stats = \Elasticsearch::indices()->stats(['index' => $index]);
+                    $currentCount = $stats['indices'][$index]['total']['docs']['count'] ?? 0;
+
+                    if ($currentCount > 0) {
+                        $this->info("ElasticSearch index '{$index}' exists with {$currentCount} documents. Clearing data...");
+
+                        // Delete all documents but keep the index structure
+                        \Elasticsearch::deleteByQuery([
+                            'index' => $index,
+                            'body' => [
+                                'query' => ['match_all' => (object)[]]
+                            ]
+                        ]);
+
+                        // Force refresh to ensure deletions are visible
+                        \Elasticsearch::indices()->refresh(['index' => $index]);
+                        $this->info("Cleared all documents from ElasticSearch index: {$index}");
+                    } else {
+                        $this->info("ElasticSearch index '{$index}' exists but is already empty");
+                    }
+                } else {
+                    $this->info("ElasticSearch index '{$index}' does not exist. Creating optimized index...");
+                    $this->createOptimizedElasticIndex($index);
+                }
+            } catch (Exception $e) {
+                $this->warn("Could not clear ElasticSearch index: {$e->getMessage()}");
+                $this->info("Attempting to recreate index...");
+
+                // Fallback: delete and recreate
+                try {
+                    \Elasticsearch::indices()->delete(['index' => $index]);
+                } catch (Exception $e) {
+                    // Index might not exist, that's okay
+                }
+                $this->createOptimizedElasticIndex($index);
+            }
+        }
+    }
+
+    /**
+     * Create optimized ElasticSearch index
+     */
+    private function createOptimizedElasticIndex(string $indexName): void
+    {
+        $settings = [
+            'index' => $indexName,
+            'body' => [
+                'settings' => [
+                    'number_of_shards' => 1,
+                    'number_of_replicas' => 0,
+                    'refresh_interval' => '30s',
+                    'translog' => [
+                        'durability' => 'async',
+                        'sync_interval' => '30s',
+                        'flush_threshold_size' => '1gb'
+                    ]
+                ],
+                'mappings' => $this->getIndexMappings($indexName)
+            ]
+        ];
+
+        \Elasticsearch::indices()->create($settings);
+        $this->info("Created optimized ElasticSearch index: {$indexName}");
+    }
+
+    /**
+     * Get index mappings
+     */
+    private function getIndexMappings(string $indexName): array
+    {
+        if ($indexName === 'releases') {
+            return [
+                'properties' => [
+                    'id' => ['type' => 'long'],
+                    'name' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'searchname' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'plainsearchname' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'fromname' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'categories_id' => ['type' => 'integer'],
+                    'filename' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'postdate' => ['type' => 'date']
+                ]
+            ];
+        } else {
+            return [
+                'properties' => [
+                    'id' => ['type' => 'long'],
+                    'title' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'filename' => ['type' => 'text', 'analyzer' => 'standard'],
+                    'source' => ['type' => 'keyword']
+                ]
+            ];
+        }
+    }
+
+    /**
+     * Get total records
+     */
+    private function getTotalRecords(string $index): int
+    {
+        return $index === 'releases' ? Release::count() : Predb::count();
+    }
+
+    /**
+     * Get selected engine
+     */
+    private function getSelectedEngine(): ?string
+    {
+        foreach (self::SUPPORTED_ENGINES as $engine) {
+            if ($this->option($engine)) {
+                return $engine;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Get selected index
+     */
+    private function getSelectedIndex(): ?string
+    {
+        foreach (self::SUPPORTED_INDEXES as $index) {
+            if ($this->option($index)) {
+                return $index;
+            }
+        }
+        return null;
+    }
+}

app/Console/Commands/NntmuxOffsetWorker.php

@@ -0,0 +1,343 @@
+<?php
+
+namespace App\Console\Commands;
+
+use App\Models\Predb;
+use App\Models\Release;
+use Blacklight\ManticoreSearch;
+use Exception;
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\DB;
+
+class NntmuxOffsetWorker extends Command
+{
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'nntmux:offset-worker
+                                       {--manticore : Use ManticoreSearch}
+                                       {--elastic : Use ElasticSearch}
+                                       {--releases : Populates the releases index}
+                                       {--predb : Populates the predb index}
+                                       {--offset=0 : Start offset}
+                                       {--limit=0 : Number of records to process}
+                                       {--worker-id=0 : Worker ID}
+                                       {--batch-size=10000 : Batch size for bulk operations}
+                                       {--disable-keys : Disable database keys}
+                                       {--memory-limit=4G : Memory limit}';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Offset-based worker process for parallel index population';
+
+    private const GROUP_CONCAT_MAX_LEN = 65535;
+
+    /**
+     * Execute the console command.
+     */
+    public function handle(): int
+    {
+        $workerId = $this->option('worker-id');
+        $offset = (int) $this->option('offset');
+        $limit = (int) $this->option('limit');
+
+        $this->info("Worker {$workerId} starting: processing {$limit} records from offset {$offset}");
+
+        try {
+            $this->optimizeWorkerSettings();
+
+            $engine = $this->getSelectedEngine();
+            $index = $this->getSelectedIndex();
+
+            if (!$engine || !$index) {
+                $this->error('Engine and index must be specified');
+                return Command::FAILURE;
+            }
+
+            $result = $this->processOffset($engine, $index, $offset, $limit, $workerId);
+
+            if ($result === Command::SUCCESS) {
+                $this->info("Worker {$workerId} completed successfully");
+            }
+
+            return $result;
+
+        } catch (Exception $e) {
+            $this->error("Worker {$workerId} failed: {$e->getMessage()}");
+            return Command::FAILURE;
+        }
+    }
+
+    /**
+     * Process a specific offset range of records
+     */
+    private function processOffset(string $engine, string $index, int $offset, int $limit, int $workerId): int
+    {
+        $batchSize = (int) $this->option('batch-size');
+        $query = $this->buildOffsetQuery($index, $offset, $limit);
+        $transformer = $this->getTransformer($engine, $index);
+
+        $indexName = $this->getIndexName($engine, $index);
+        $processed = 0;
+        $errors = 0;
+
+        $this->info("Worker {$workerId}: Processing {$limit} records with batch size {$batchSize}");
+        $startTime = microtime(true);
+
+        if ($engine === 'manticore') {
+            $manticore = new ManticoreSearch;
+            $batchData = [];
+
+            $query->chunk($batchSize, function ($items) use ($manticore, $indexName, $transformer, &$processed, &$errors, &$batchData, $batchSize, $workerId) {
+                $this->info("Worker {$workerId}: Processing chunk of {$items->count()} items");
+
+                foreach ($items as $item) {
+                    try {
+                        $batchData[] = $transformer($item);
+                        $processed++;
+
+                        if (count($batchData) >= $batchSize) {
+                            $this->processManticoreBatch($manticore, $indexName, $batchData, $workerId);
+                            $this->info("Worker {$workerId}: Inserted batch of " . count($batchData) . " records");
+                            $batchData = [];
+                        }
+                    } catch (Exception $e) {
+                        $errors++;
+                        $this->error("Worker {$workerId}: Error processing item {$item->id}: {$e->getMessage()}");
+                    }
+                }
+            });
+
+            // Process remaining items
+            if (!empty($batchData)) {
+                $this->processManticoreBatch($manticore, $indexName, $batchData, $workerId);
+                $this->info("Worker {$workerId}: Inserted final batch of " . count($batchData) . " records");
+            }
+
+        } else { // ElasticSearch
+            $query->chunk($batchSize, function ($items) use ($indexName, $transformer, &$processed, &$errors, $batchSize, $workerId) {
+                $this->info("Worker {$workerId}: Processing chunk of {$items->count()} items");
+                $data = ['body' => []];
+
+                foreach ($items as $item) {
+                    try {
+                        $transformedData = $transformer($item);
+
+                        $data['body'][] = [
+                            'index' => [
+                                '_index' => $indexName,
+                                '_id' => $item->id,
+                            ],
+                        ];
+                        $data['body'][] = $transformedData;
+
+                        $processed++;
+                    } catch (Exception $e) {
+                        $errors++;
+                        $this->error("Worker {$workerId}: Error processing item {$item->id}: {$e->getMessage()}");
+                    }
+                }
+
+                if (!empty($data['body'])) {
+                    $this->processElasticBatch($data, $workerId);
+                    $this->info("Worker {$workerId}: Inserted batch of " . (count($data['body']) / 2) . " records");
+                }
+            });
+        }
+
+        $executionTime = round(microtime(true) - $startTime, 2);
+        $this->info("Worker {$workerId}: Completed processing {$processed} records with {$errors} errors in {$executionTime} seconds");
+        return Command::SUCCESS;
+    }
+
+    /**
+     * Build offset-based query
+     */
+    private function buildOffsetQuery(string $index, int $offset, int $limit)
+    {
+        if ($index === 'releases') {
+            return Release::query()
+                ->select([
+                    'releases.id',
+                    'releases.name',
+                    'releases.searchname',
+                    'releases.fromname',
+                    'releases.categories_id',
+                    'releases.postdate',
+                ])
+                ->selectRaw('(SELECT GROUP_CONCAT(rf.name SEPARATOR " ") FROM release_files rf WHERE rf.releases_id = releases.id) AS filename')
+                ->orderBy('releases.id')
+                ->offset($offset)
+                ->limit($limit);
+        } else {
+            return Predb::query()
+                ->select(['id', 'title', 'filename', 'source'])
+                ->orderBy('id')
+                ->offset($offset)
+                ->limit($limit);
+        }
+    }
+
+    /**
+     * Get transformer function
+     */
+    private function getTransformer(string $engine, string $index): callable
+    {
+        if ($index === 'releases') {
+            if ($engine === 'manticore') {
+                return function ($item) {
+                    return [
+                        'id' => (string) $item->id,
+                        'name' => (string) ($item->name ?: ''),
+                        'searchname' => (string) ($item->searchname ?: ''),
+                        'fromname' => (string) ($item->fromname ?: ''),
+                        'categories_id' => (string) ($item->categories_id ?: '0'),
+                        'filename' => (string) ($item->filename ?: ''),
+                        'dummy' => '1',
+                    ];
+                };
+            } else {
+                return function ($item) {
+                    $searchName = str_replace(['.', '-'], ' ', $item->searchname ?? '');
+                    return [
+                        'id' => $item->id,
+                        'name' => $item->name,
+                        'searchname' => $item->searchname,
+                        'plainsearchname' => $searchName,
+                        'fromname' => $item->fromname,
+                        'categories_id' => $item->categories_id,
+                        'filename' => $item->filename ?? '',
+                        'postdate' => $item->postdate,
+                    ];
+                };
+            }
+        } else { // predb
+            return function ($item) {
+                return [
+                    'id' => $item->id,
+                    'title' => (string) ($item->title ?? ''),
+                    'filename' => (string) ($item->filename ?? ''),
+                    'source' => (string) ($item->source ?? ''),
+                    'dummy' => 1,
+                ];
+            };
+        }
+    }
+
+    /**
+     * Get index name
+     */
+    private function getIndexName(string $engine, string $index): string
+    {
+        if ($engine === 'manticore') {
+            return $index === 'releases' ? 'releases_rt' : 'predb_rt';
+        } else {
+            return $index;
+        }
+    }
+
+    /**
+     * Process ManticoreSearch batch
+     */
+    private function processManticoreBatch(ManticoreSearch $manticore, string $indexName, array $data, int $workerId): void
+    {
+        $retries = 3;
+        $attempt = 0;
+
+        while ($attempt < $retries) {
+            try {
+                $manticore->manticoreSearch->table($indexName)->replaceDocuments($data);
+                break;
+            } catch (Exception $e) {
+                $attempt++;
+                if ($attempt >= $retries) {
+                    throw new Exception("Worker {$workerId}: Failed to process ManticoreSearch batch after {$retries} attempts: {$e->getMessage()}");
+                }
+                usleep(200000); // 200ms delay before retry
+            }
+        }
+    }
+
+    /**
+     * Process ElasticSearch batch
+     */
+    private function processElasticBatch(array $data, int $workerId): void
+    {
+        $retries = 3;
+        $attempt = 0;
+
+        while ($attempt < $retries) {
+            try {
+                $response = \Elasticsearch::bulk($data);
+
+                if (isset($response['errors']) && $response['errors']) {
+                    $errorCount = 0;
+                    foreach ($response['items'] as $item) {
+                        if (isset($item['index']['error'])) {
+                            $errorCount++;
+                        }
+                    }
+                    if ($errorCount > 0) {
+                        $this->warn("Worker {$workerId}: ElasticSearch batch had {$errorCount} errors");
+                    }
+                }
+                break;
+            } catch (Exception $e) {
+                $attempt++;
+                if ($attempt >= $retries) {
+                    throw new Exception("Worker {$workerId}: Failed to process ElasticSearch batch after {$retries} attempts: {$e->getMessage()}");
+                }
+                usleep(200000); // 200ms delay before retry
+            }
+        }
+    }
+
+    /**
+     * Optimize worker settings
+     */
+    private function optimizeWorkerSettings(): void
+    {
+        // Set memory limit
+        ini_set('memory_limit', $this->option('memory-limit'));
+
+        // Database optimizations
+        if ($this->option('disable-keys')) {
+            try {
+                DB::statement('SET SESSION FOREIGN_KEY_CHECKS = 0');
+                DB::statement('SET SESSION UNIQUE_CHECKS = 0');
+                DB::statement('SET SESSION AUTOCOMMIT = 0');
+                DB::statement('SET SESSION read_buffer_size = 2097152');
+                DB::statement('SET SESSION sort_buffer_size = 16777216');
+            } catch (Exception $e) {
+                $this->warn("Could not optimize database settings: {$e->getMessage()}");
+            }
+        }
+
+        // Set GROUP_CONCAT max length
+        DB::statement('SET SESSION group_concat_max_len = ?', [self::GROUP_CONCAT_MAX_LEN]);
+
+        // PHP optimizations
+        gc_enable();
+    }
+
+    /**
+     * Get selected engine
+     */
+    private function getSelectedEngine(): ?string
+    {
+        return $this->option('manticore') ? 'manticore' : ($this->option('elastic') ? 'elastic' : null);
+    }
+
+    /**
+     * Get selected index
+     */
+    private function getSelectedIndex(): ?string
+    {
+        return $this->option('releases') ? 'releases' : ($this->option('predb') ? 'predb' : null);
+    }
+}

app/Console/Commands/NntmuxPopulateSearchIndexes.php

@@ -22,7 +22,10 @@ class NntmuxPopulateSearchIndexes extends Command
                                        {--elastic : Use ElasticSearch}
                                        {--releases : Populates the releases index}
                                        {--predb : Populates the predb index}
-                                       {--count=20000 : Sets the chunk size}
+                                       {--count=50000 : Sets the chunk size}
+                                       {--parallel=4 : Number of parallel processes}
+                                       {--batch-size=5000 : Batch size for bulk operations}
+                                       {--disable-keys : Disable database keys during population}
                                        {--optimize : Optimize ManticoreSearch indexes}';
 
     /**
@@ -38,7 +41,11 @@ class NntmuxPopulateSearchIndexes extends Command
 
     private const GROUP_CONCAT_MAX_LEN = 16384;
 
-    private const DEFAULT_CHUNK_SIZE = 20000;
+    private const DEFAULT_CHUNK_SIZE = 50000;
+
+    private const DEFAULT_PARALLEL_PROCESSES = 4;
+
+    private const DEFAULT_BATCH_SIZE = 5000;
 
     /**
      * Execute the console command.
@@ -235,20 +242,23 @@ class NntmuxPopulateSearchIndexes extends Command
     }
 
     /**
-     * Process data for ManticoreSearch
+     * Process data for ManticoreSearch with optimizations
      */
     private function processManticoreData(string $indexName, int $total, $query, callable $transformer): int
     {
         $manticore = new ManticoreSearch;
         $chunkSize = $this->getChunkSize();
+        $batchSize = $this->getBatchSize();
 
+        $this->optimizeDatabase();
         $this->setGroupConcatMaxLen();
 
         $this->info(sprintf(
-            "Populating ManticoreSearch index '%s' with %s rows using chunks of %s.",
+            "Populating ManticoreSearch index '%s' with %s rows using chunks of %s and batch size of %s.",
             $indexName,
             number_format($total),
-            number_format($chunkSize)
+            number_format($chunkSize),
+            number_format($batchSize)
         ));
 
         $bar = $this->output->createProgressBar($total);
@@ -257,15 +267,20 @@ class NntmuxPopulateSearchIndexes extends Command
 
         $processedCount = 0;
         $errorCount = 0;
+        $batchData = [];
 
         try {
-            $query->chunk($chunkSize, function ($items) use ($manticore, $indexName, $transformer, $bar, &$processedCount, &$errorCount) {
-                $data = [];
-
+            $query->chunk($chunkSize, function ($items) use ($manticore, $indexName, $transformer, $bar, &$processedCount, &$errorCount, $batchSize, &$batchData) {
                 foreach ($items as $item) {
                     try {
-                        $data[] = $transformer($item);
+                        $batchData[] = $transformer($item);
                         $processedCount++;
+
+                        // Process in optimized batch sizes
+                        if (count($batchData) >= $batchSize) {
+                            $this->processBatch($manticore, $indexName, $batchData);
+                            $batchData = [];
+                        }
                     } catch (Exception $e) {
                         $errorCount++;
                         if ($this->output->isVerbose()) {
@@ -274,12 +289,13 @@ class NntmuxPopulateSearchIndexes extends Command
                     }
                     $bar->advance();
                 }
-
-                if (! empty($data)) {
-                    $manticore->manticoreSearch->table($indexName)->replaceDocuments($data);
-                }
             });
 
+            // Process remaining items
+            if (!empty($batchData)) {
+                $this->processBatch($manticore, $indexName, $batchData);
+            }
+
             $bar->finish();
             $this->newLine();
 
@@ -297,6 +313,8 @@ class NntmuxPopulateSearchIndexes extends Command
             $this->error("Failed to populate ManticoreSearch: {$e->getMessage()}");
 
             return Command::FAILURE;
+        } finally {
+            $this->restoreDatabase();
         }
     }
 
@@ -386,19 +404,22 @@ class NntmuxPopulateSearchIndexes extends Command
     }
 
     /**
-     * Process data for ElasticSearch
+     * Process data for ElasticSearch with optimizations
      */
     private function processElasticData(string $indexName, int $total, $query, callable $transformer): int
     {
         $chunkSize = $this->getChunkSize();
+        $batchSize = $this->getBatchSize();
 
+        $this->optimizeDatabase();
         $this->setGroupConcatMaxLen();
 
         $this->info(sprintf(
-            "Populating ElasticSearch index '%s' with %s rows using chunks of %s.",
+            "Populating ElasticSearch index '%s' with %s rows using chunks of %s and batch size of %s.",
             $indexName,
             number_format($total),
-            number_format($chunkSize)
+            number_format($chunkSize),
+            number_format($batchSize)
         ));
 
         $bar = $this->output->createProgressBar($total);
@@ -407,11 +428,10 @@ class NntmuxPopulateSearchIndexes extends Command
 
         $processedCount = 0;
         $errorCount = 0;
-        $batchSize = min($chunkSize, 1000); // ElasticSearch performs better with smaller bulk sizes
 
         try {
             $query->chunk($chunkSize, function ($items) use ($indexName, $transformer, $bar, &$processedCount, &$errorCount, $batchSize) {
-                // Process in smaller batches for ElasticSearch
+                // Process in optimized batches for ElasticSearch
                 foreach ($items->chunk($batchSize) as $batch) {
                     $data = ['body' => []];
 
@@ -438,20 +458,8 @@ class NntmuxPopulateSearchIndexes extends Command
                         $bar->advance();
                     }
 
-                    if (! empty($data['body'])) {
-                        $response = \Elasticsearch::bulk($data);
-
-                        // Check for errors in bulk response
-                        if (isset($response['errors']) && $response['errors']) {
-                            foreach ($response['items'] as $item) {
-                                if (isset($item['index']['error'])) {
-                                    $errorCount++;
-                                    if ($this->output->isVerbose()) {
-                                        $this->error('ElasticSearch error: '.json_encode($item['index']['error']));
-                                    }
-                                }
-                            }
-                        }
+                    if (!empty($data['body'])) {
+                        $this->processElasticBatch($data, $errorCount);
                     }
                 }
             });
@@ -473,6 +481,109 @@ class NntmuxPopulateSearchIndexes extends Command
             $this->error("Failed to populate ElasticSearch: {$e->getMessage()}");
 
             return Command::FAILURE;
+        } finally {
+            $this->restoreDatabase();
+        }
+    }
+
+    /**
+     * Process ManticoreSearch batch with retry logic
+     */
+    private function processBatch(ManticoreSearch $manticore, string $indexName, array $data): void
+    {
+        $retries = 3;
+        $attempt = 0;
+
+        while ($attempt < $retries) {
+            try {
+                $manticore->manticoreSearch->table($indexName)->replaceDocuments($data);
+                break;
+            } catch (Exception $e) {
+                $attempt++;
+                if ($attempt >= $retries) {
+                    throw $e;
+                }
+                usleep(100000); // 100ms delay before retry
+            }
+        }
+    }
+
+    /**
+     * Process ElasticSearch batch with retry logic
+     */
+    private function processElasticBatch(array $data, int &$errorCount): void
+    {
+        $retries = 3;
+        $attempt = 0;
+
+        while ($attempt < $retries) {
+            try {
+                $response = \Elasticsearch::bulk($data);
+
+                // Check for errors in bulk response
+                if (isset($response['errors']) && $response['errors']) {
+                    foreach ($response['items'] as $item) {
+                        if (isset($item['index']['error'])) {
+                            $errorCount++;
+                            if ($this->output->isVerbose()) {
+                                $this->error('ElasticSearch error: '.json_encode($item['index']['error']));
+                            }
+                        }
+                    }
+                }
+                break;
+            } catch (Exception $e) {
+                $attempt++;
+                if ($attempt >= $retries) {
+                    throw $e;
+                }
+                usleep(100000); // 100ms delay before retry
+            }
+        }
+    }
+
+    /**
+     * Optimize database settings for bulk operations
+     */
+    private function optimizeDatabase(): void
+    {
+        if ($this->option('disable-keys')) {
+            $this->info('Disabling database keys for faster bulk operations...');
+
+            try {
+                // Disable foreign key checks
+                DB::statement('SET FOREIGN_KEY_CHECKS = 0');
+                DB::statement('SET UNIQUE_CHECKS = 0');
+                DB::statement('SET AUTOCOMMIT = 0');
+
+                // Increase buffer sizes
+                DB::statement('SET SESSION innodb_buffer_pool_size = 1073741824'); // 1GB
+                DB::statement('SET SESSION bulk_insert_buffer_size = 268435456'); // 256MB
+                DB::statement('SET SESSION read_buffer_size = 2097152'); // 2MB
+                DB::statement('SET SESSION sort_buffer_size = 16777216'); // 16MB
+
+            } catch (Exception $e) {
+                $this->warn("Could not optimize database settings: {$e->getMessage()}");
+            }
+        }
+    }
+
+    /**
+     * Restore database settings after bulk operations
+     */
+    private function restoreDatabase(): void
+    {
+        if ($this->option('disable-keys')) {
+            $this->info('Restoring database settings...');
+
+            try {
+                DB::statement('SET FOREIGN_KEY_CHECKS = 1');
+                DB::statement('SET UNIQUE_CHECKS = 1');
+                DB::statement('SET AUTOCOMMIT = 1');
+                DB::statement('COMMIT');
+            } catch (Exception $e) {
+                $this->warn("Could not restore database settings: {$e->getMessage()}");
+            }
         }
     }
 
@@ -486,6 +597,15 @@ class NntmuxPopulateSearchIndexes extends Command
         return $chunkSize > 0 ? $chunkSize : self::DEFAULT_CHUNK_SIZE;
     }
 
+    /**
+     * Get the batch size from options
+     */
+    private function getBatchSize(): int
+    {
+        $batchSize = (int) $this->option('batch-size');
+        return $batchSize > 0 ? $batchSize : self::DEFAULT_BATCH_SIZE;
+    }
+
     /**
      * Set the GROUP_CONCAT max length for the session
      */