mirror of
https://github.com/NNTmux/newznab-tmux.git
synced 2026-08-29 01:08:56 +00:00
DariusIII
317 lines
9.4 KiB
PHP
317 lines
9.4 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace App\Services\Binaries;
|
|
|
|
use Illuminate\Support\Facades\DB;
|
|
use Illuminate\Support\Facades\Log;
|
|
|
|
/**
|
|
* Handles binary record creation and updates during header storage.
|
|
*/
|
|
final class BinaryHandler
|
|
{
|
|
/** @var array<int, array{Size: int, Parts: int}> Pending binary updates */
|
|
private array $binariesUpdate = [];
|
|
|
|
/** @var array<int, true> IDs of binaries created in this batch */
|
|
private array $insertedBinaryIds = [];
|
|
|
|
/** @var array<string, array{CollectionID: int, BinaryID: int}> Processed articles */
|
|
private array $articles = [];
|
|
|
|
public function __construct() {}
|
|
|
|
/**
|
|
* Reset state for a new batch.
|
|
*/
|
|
public function reset(): void
|
|
{
|
|
$this->binariesUpdate = [];
|
|
$this->insertedBinaryIds = [];
|
|
$this->articles = [];
|
|
}
|
|
|
|
/**
|
|
* Get or create a binary for the given header.
|
|
*
|
|
* @param array<string, mixed> $header
|
|
* @return int|null Binary ID or null on failure
|
|
*/
|
|
public function getOrCreateBinary(
|
|
array $header,
|
|
int $collectionId,
|
|
int $groupId,
|
|
int $fileNumber
|
|
): ?int {
|
|
$articleKey = $header['matches'][1];
|
|
|
|
// Return cached if already processed
|
|
if (isset($this->articles[$articleKey])) {
|
|
$binaryId = $this->articles[$articleKey]['BinaryID'];
|
|
$this->binariesUpdate[$binaryId]['Size'] += $header['Bytes'];
|
|
$this->binariesUpdate[$binaryId]['Parts']++;
|
|
|
|
return $binaryId;
|
|
}
|
|
|
|
$hash = md5((string) ($header['matches'][1] ?? '').(string) ($header['From'] ?? '').(string) $groupId);
|
|
$driver = DB::getDriverName();
|
|
|
|
try {
|
|
$binaryId = $this->insertOrGetBinary(
|
|
$driver,
|
|
$hash,
|
|
$header,
|
|
$collectionId,
|
|
$fileNumber
|
|
);
|
|
|
|
if ($binaryId > 0) {
|
|
$this->binariesUpdate[$binaryId] = ['Size' => 0, 'Parts' => 0];
|
|
$this->articles[$articleKey] = [
|
|
'CollectionID' => $collectionId,
|
|
'BinaryID' => $binaryId,
|
|
];
|
|
|
|
return $binaryId;
|
|
}
|
|
} catch (\Throwable $e) {
|
|
if (config('app.debug') === true) {
|
|
Log::error('Binary insert failed: '.$e->getMessage());
|
|
}
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* @param array<string, mixed> $header
|
|
*/
|
|
private function insertOrGetBinary(
|
|
string $driver,
|
|
string $hash,
|
|
array $header,
|
|
int $collectionId,
|
|
int $fileNumber
|
|
): int {
|
|
$name = mb_convert_encoding($header['matches'][1], 'UTF-8', mb_list_encodings());
|
|
$totalParts = (int) $header['matches'][3];
|
|
$partSize = (int) $header['Bytes'];
|
|
|
|
if ($driver === 'sqlite') {
|
|
return $this->insertBinarySqlite($hash, $name, $collectionId, $totalParts, $fileNumber, $partSize);
|
|
}
|
|
|
|
return $this->insertBinaryMysql($hash, $name, $collectionId, $totalParts, $fileNumber, $partSize);
|
|
}
|
|
|
|
private function insertBinarySqlite(
|
|
string $hash,
|
|
string $name,
|
|
int $collectionId,
|
|
int $totalParts,
|
|
int $fileNumber,
|
|
int $partSize
|
|
): int {
|
|
// Verify collection exists before insert to avoid FK constraint violation
|
|
$collectionExists = DB::selectOne(
|
|
'SELECT id FROM collections WHERE id = ? LIMIT 1',
|
|
[$collectionId]
|
|
);
|
|
|
|
if ($collectionExists === null) {
|
|
if (config('app.debug') === true) {
|
|
Log::warning("Binary insert skipped: collection {$collectionId} no longer exists");
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
DB::statement(
|
|
'INSERT OR IGNORE INTO binaries (binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) VALUES (?, ?, ?, ?, 1, ?, ?)',
|
|
[$hash, $name, $collectionId, $totalParts, $fileNumber, $partSize]
|
|
);
|
|
|
|
$lastId = (int) DB::connection()->getPdo()->lastInsertId();
|
|
if ($lastId > 0) {
|
|
$this->insertedBinaryIds[$lastId] = true;
|
|
|
|
return $lastId;
|
|
}
|
|
|
|
$bin = DB::selectOne(
|
|
'SELECT id FROM binaries WHERE binaryhash = ? AND collections_id = ? LIMIT 1',
|
|
[$hash, $collectionId]
|
|
);
|
|
|
|
return (int) ($bin->id ?? 0);
|
|
}
|
|
|
|
private function insertBinaryMysql(
|
|
string $hash,
|
|
string $name,
|
|
int $collectionId,
|
|
int $totalParts,
|
|
int $fileNumber,
|
|
int $partSize
|
|
): int {
|
|
// Verify collection exists before insert to avoid FK constraint violation
|
|
$collectionExists = DB::selectOne(
|
|
'SELECT id FROM collections WHERE id = ? LIMIT 1',
|
|
[$collectionId]
|
|
);
|
|
|
|
if ($collectionExists === null) {
|
|
if (config('app.debug') === true) {
|
|
Log::warning("Binary insert skipped: collection {$collectionId} no longer exists");
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
$sql = 'INSERT INTO binaries '
|
|
.'(binaryhash, name, collections_id, totalparts, currentparts, filenumber, partsize) '
|
|
.'VALUES (UNHEX(?), ?, ?, ?, 1, ?, ?) '
|
|
.'ON DUPLICATE KEY UPDATE currentparts = currentparts + 1, partsize = partsize + VALUES(partsize)';
|
|
|
|
DB::statement($sql, [$hash, $name, $collectionId, $totalParts, $fileNumber, $partSize]);
|
|
|
|
$lastId = (int) DB::connection()->getPdo()->lastInsertId();
|
|
if ($lastId > 0) {
|
|
$this->insertedBinaryIds[$lastId] = true;
|
|
|
|
return $lastId;
|
|
}
|
|
|
|
$bin = DB::selectOne(
|
|
'SELECT id FROM binaries WHERE binaryhash = UNHEX(?) AND collections_id = ? LIMIT 1',
|
|
[$hash, $collectionId]
|
|
);
|
|
|
|
return (int) ($bin->id ?? 0);
|
|
}
|
|
|
|
/**
|
|
* Flush accumulated size/parts updates to the database.
|
|
*/
|
|
public function flushUpdates(int $chunkSize = 1000): bool
|
|
{
|
|
$updates = $this->getPendingUpdates();
|
|
if (empty($updates)) {
|
|
return true;
|
|
}
|
|
|
|
$driver = DB::getDriverName();
|
|
|
|
try {
|
|
if ($driver === 'sqlite') {
|
|
return $this->flushUpdatesSqlite($updates); // @phpstan-ignore argument.type
|
|
}
|
|
|
|
return $this->flushUpdatesMysql($updates, $chunkSize); // @phpstan-ignore argument.type
|
|
} catch (\Throwable $e) {
|
|
if (config('app.debug') === true) {
|
|
Log::error('Binaries aggregate update failed: '.$e->getMessage());
|
|
}
|
|
|
|
return false;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* @param array<string, mixed> $updates
|
|
*/
|
|
private function flushUpdatesSqlite(array $updates): bool
|
|
{
|
|
foreach ($updates as $row) {
|
|
DB::statement(
|
|
'UPDATE binaries SET partsize = partsize + ?, currentparts = currentparts + ? WHERE id = ?',
|
|
[$row['partsize'], $row['currentparts'], $row['id']]
|
|
);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* @param array<string, mixed> $updates
|
|
*/
|
|
private function flushUpdatesMysql(array $updates, int $chunkSize): bool
|
|
{
|
|
foreach (array_chunk($updates, $chunkSize) as $chunk) {
|
|
// Build a CASE statement for batch UPDATE instead of INSERT...ON DUPLICATE KEY
|
|
// This avoids FK constraint violations when collections have been deleted
|
|
$ids = [];
|
|
$partsizeCases = [];
|
|
$currentpartsCases = [];
|
|
$bindings = [];
|
|
|
|
foreach ($chunk as $row) {
|
|
$ids[] = $row['id'];
|
|
$partsizeCases[] = 'WHEN id = ? THEN partsize + ?';
|
|
$currentpartsCases[] = 'WHEN id = ? THEN currentparts + ?';
|
|
$bindings[] = $row['id'];
|
|
$bindings[] = $row['partsize'];
|
|
}
|
|
|
|
foreach ($chunk as $row) {
|
|
$bindings[] = $row['id'];
|
|
$bindings[] = $row['currentparts'];
|
|
}
|
|
|
|
$idPlaceholders = implode(',', array_fill(0, count($ids), '?'));
|
|
$bindings = array_merge($bindings, $ids);
|
|
|
|
$sql = 'UPDATE binaries SET '
|
|
.'partsize = CASE '.implode(' ', $partsizeCases).' ELSE partsize END, '
|
|
.'currentparts = CASE '.implode(' ', $currentpartsCases).' ELSE currentparts END '
|
|
.'WHERE id IN ('.$idPlaceholders.')';
|
|
|
|
DB::statement($sql, $bindings);
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Check if article is already processed.
|
|
*/
|
|
public function hasArticle(string $articleKey): bool
|
|
{
|
|
return isset($this->articles[$articleKey]);
|
|
}
|
|
|
|
/**
|
|
* Get IDs created in this batch.
|
|
*
|
|
* @return list<int>
|
|
*/
|
|
public function getInsertedIds(): array
|
|
{
|
|
return array_keys($this->insertedBinaryIds);
|
|
}
|
|
|
|
/**
|
|
* Get pending binary updates that haven't been flushed.
|
|
*
|
|
* @return list<array<string, int>>
|
|
*/
|
|
private function getPendingUpdates(): array
|
|
{
|
|
$rows = [];
|
|
foreach ($this->binariesUpdate as $binaryId => $binary) {
|
|
if (($binary['Size'] ?? 0) > 0 || ($binary['Parts'] ?? 0) > 0) {
|
|
$rows[] = [
|
|
'id' => $binaryId,
|
|
'partsize' => $binary['Size'],
|
|
'currentparts' => $binary['Parts'],
|
|
];
|
|
}
|
|
}
|
|
|
|
return $rows;
|
|
}
|
|
}
|