<?php
declare(strict_types=1);

/**
 * Minor Planet Center Comet Import
 * Quelle:
 * https://www.minorplanetcenter.net/iau/Ephemerides/Comets/Soft00Cmt.txt
 *
 * Zieltabelle:
 * - comets_mpc
 *
 * Nutzung:
 *   php scripts/import/import_mpc_comets.php
 *   php scripts/import/import_mpc_comets.php C:\pfad\zu\Soft00Cmt.txt
 */

error_reporting(E_ALL);
ini_set('display_errors', '1');
set_time_limit(0);
date_default_timezone_set('UTC');

$rootDir = dirname(__DIR__, 2);
$config = require $rootDir . DIRECTORY_SEPARATOR . 'config' . DIRECTORY_SEPARATOR . 'database.php';

if (!is_array($config)) {
    writeError("config/database.php muss ein Array mit DB-Zugangsdaten zurueckgeben.");
    exit(1);
}

$config = normalizeDbConfig($config);
$sourceUrl = 'https://www.minorplanetcenter.net/iau/Ephemerides/Comets/Soft00Cmt.txt';
$localSourcePath = $argv[1] ?? ($_GET['source'] ?? null);

$dsn = sprintf(
    'mysql:host=%s;port=%d;dbname=%s;charset=%s',
    $config['host'],
    (int) $config['port'],
    $config['dbname'],
    $config['charset']
);

$pdo = new PDO(
    $dsn,
    $config['user'],
    $config['pass'],
    [
        PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
        PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
    ]
);

ensureImportRunsTableExists($pdo);
ensureCometsTableExists($pdo);

$recordsReceived = 0;
$recordsInserted = 0;
$recordsUpdated = 0;
$recordsUnchanged = 0;
$recordsSkipped = 0;
$importRunId = startImportRun($pdo, 'mpc-comets', $localSourcePath !== null && trim((string) $localSourcePath) !== '' ? (string) $localSourcePath : $sourceUrl);

try {
    [$lines, $resolvedSource] = resolveSourceLines($sourceUrl, $localSourcePath);
    $statement = $pdo->prepare(
        'INSERT INTO `comets_mpc` (
            `designation_packed`,
            `orbit_type`,
            `year_of_perihelion`,
            `month_of_perihelion`,
            `day_of_perihelion`,
            `perihelion_dist_au`,
            `eccentricity`,
            `arg_perihelion_deg`,
            `ascending_node_deg`,
            `inclination_deg`,
            `epoch_compact`,
            `epoch_date`,
            `absolute_magnitude_h`,
            `slope_parameter_g`,
            `designation_and_name`,
            `reference_text`,
            `raw_record`,
            `source_hash`,
            `source_url`,
            `imported_at`,
            `created_at`,
            `updated_at`
        ) VALUES (
            :designation_packed,
            :orbit_type,
            :year_of_perihelion,
            :month_of_perihelion,
            :day_of_perihelion,
            :perihelion_dist_au,
            :eccentricity,
            :arg_perihelion_deg,
            :ascending_node_deg,
            :inclination_deg,
            :epoch_compact,
            :epoch_date,
            :absolute_magnitude_h,
            :slope_parameter_g,
            :designation_and_name,
            :reference_text,
            :raw_record,
            :source_hash,
            :source_url,
            NOW(),
            NOW(),
            NOW()
        )
        ON DUPLICATE KEY UPDATE
            `orbit_type` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`orbit_type`), `orbit_type`),
            `year_of_perihelion` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`year_of_perihelion`), `year_of_perihelion`),
            `month_of_perihelion` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`month_of_perihelion`), `month_of_perihelion`),
            `day_of_perihelion` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`day_of_perihelion`), `day_of_perihelion`),
            `perihelion_dist_au` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`perihelion_dist_au`), `perihelion_dist_au`),
            `eccentricity` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`eccentricity`), `eccentricity`),
            `arg_perihelion_deg` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`arg_perihelion_deg`), `arg_perihelion_deg`),
            `ascending_node_deg` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`ascending_node_deg`), `ascending_node_deg`),
            `inclination_deg` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`inclination_deg`), `inclination_deg`),
            `epoch_compact` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`epoch_compact`), `epoch_compact`),
            `epoch_date` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`epoch_date`), `epoch_date`),
            `absolute_magnitude_h` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`absolute_magnitude_h`), `absolute_magnitude_h`),
            `slope_parameter_g` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`slope_parameter_g`), `slope_parameter_g`),
            `designation_and_name` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`designation_and_name`), `designation_and_name`),
            `reference_text` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`reference_text`), `reference_text`),
            `raw_record` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`raw_record`), `raw_record`),
            `source_hash` = VALUES(`source_hash`),
            `source_url` = IF(`source_hash` <> VALUES(`source_hash`), VALUES(`source_url`), `source_url`),
            `imported_at` = IF(`source_hash` <> VALUES(`source_hash`), NOW(), `imported_at`),
            `updated_at` = IF(`source_hash` <> VALUES(`source_hash`), NOW(), `updated_at`)'
    );

    $pdo->beginTransaction();
    $seenDesignations = [];

    foreach ($lines as $lineNumber => $line) {
        $trimmedLine = rtrim($line, "\r\n");
        if (trim($trimmedLine) === '') {
            continue;
        }

        $recordsReceived++;

        try {
            $record = parseSoft00CmtLine($trimmedLine);
            $record[':source_url'] = $resolvedSource;
            $seenDesignations[] = $record[':designation_packed'];
            $statement->execute($record);
            $affectedRows = (int) $statement->rowCount();
            if ($affectedRows === 1) {
                $recordsInserted++;
            } elseif ($affectedRows >= 2) {
                $recordsUpdated++;
            } else {
                $recordsUnchanged++;
            }
        } catch (Throwable $rowError) {
            $recordsSkipped++;
            echo 'Uebersprungen Zeile ' . ($lineNumber + 1) . ': ' . $rowError->getMessage() . "\n";
        }
    }

    if ($seenDesignations !== []) {
        $seenDesignations = array_values(array_unique($seenDesignations));
        $placeholders = implode(', ', array_fill(0, count($seenDesignations), '?'));
        $deleteStaleStatement = $pdo->prepare(sprintf(
            'DELETE FROM `comets_mpc` WHERE `designation_packed` NOT IN (%s)',
            $placeholders
        ));
        $deleteStaleStatement->execute($seenDesignations);
    }

    $pdo->commit();

    echo "Import erfolgreich\n";
    echo "Quelle: {$resolvedSource}\n";
    echo "Empfangen: {$recordsReceived}\n";
    echo "Neu: {$recordsInserted}\n";
    echo "Aktualisiert: {$recordsUpdated}\n";
    echo "Unveraendert: {$recordsUnchanged}\n";
    echo "Uebersprungen: {$recordsSkipped}\n";

    finishImportRun(
        $pdo,
        $importRunId,
        $recordsReceived,
        $recordsInserted,
        $recordsUpdated,
        'success',
        $recordsSkipped > 0 ? "Es wurden {$recordsSkipped} Datensaetze uebersprungen." : null
    );
} catch (Throwable $e) {
    if ($pdo->inTransaction()) {
        $pdo->rollBack();
    }

    finishImportRun(
        $pdo,
        $importRunId,
        $recordsReceived,
        $recordsInserted,
        $recordsUpdated,
        'error',
        substr($e->getMessage(), 0, 65000)
    );

    writeError('Fehler: ' . $e->getMessage());
    exit(1);
}

function writeError(string $message): void
{
    if (PHP_SAPI === 'cli' || PHP_SAPI === 'phpdbg') {
        fwrite(STDERR, $message . PHP_EOL);
        return;
    }

    echo htmlspecialchars($message, ENT_QUOTES, 'UTF-8') . "<br>\n";
}

function fetchRemoteLines(string $url): array
{
    if (function_exists('curl_init')) {
        return fetchRemoteLinesViaCurl($url);
    }

    try {
        return fetchRemoteLinesViaStream($url);
    } catch (Throwable $streamError) {
        return fetchRemoteLinesViaSocket($url, $streamError->getMessage());
    }
}

function resolveSourceLines(string $remoteUrl, ?string $localSourcePath = null): array
{
    if ($localSourcePath !== null && trim($localSourcePath) !== '') {
        $lines = fetchLocalLines($localSourcePath);
        return [$lines, realpath($localSourcePath) ?: $localSourcePath];
    }

    try {
        $lines = fetchRemoteLines($remoteUrl);
        return [$lines, $remoteUrl];
    } catch (Throwable $remoteError) {
        $defaultLocalPath = dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Soft00Cmt.txt';
        if (is_file($defaultLocalPath)) {
            $lines = fetchLocalLines($defaultLocalPath);
            return [$lines, realpath($defaultLocalPath) ?: $defaultLocalPath];
        }

        throw new RuntimeException(
            $remoteError->getMessage()
            . ' | Lege alternativ eine lokale Datei unter scripts/import/Soft00Cmt.txt ab '
            . 'oder uebergib ihren Pfad als Argument bzw. ?source='
        );
    }
}

function fetchLocalLines(string $path): array
{
    if (!is_file($path)) {
        throw new RuntimeException('Lokale Datei nicht gefunden: ' . $path);
    }

    $contents = @file_get_contents($path);
    if ($contents === false) {
        $error = error_get_last();
        throw new RuntimeException('Lokale Datei konnte nicht gelesen werden: ' . ($error['message'] ?? $path));
    }

    return preg_split("/\r\n|\n|\r/", $contents) ?: [];
}

function fetchRemoteLinesViaCurl(string $url): array
{
    $ch = curl_init($url);

    curl_setopt_array($ch, [
        CURLOPT_RETURNTRANSFER => true,
        CURLOPT_FOLLOWLOCATION => true,
        CURLOPT_CONNECTTIMEOUT => 20,
        CURLOPT_TIMEOUT => 300,
        CURLOPT_USERAGENT => 'AstroMINT MPC Comet Importer/2.0',
        CURLOPT_HTTPHEADER => [
            'Accept: text/plain',
        ],
    ]);

    $response = curl_exec($ch);

    if ($response === false) {
        throw new RuntimeException('Download fehlgeschlagen: ' . curl_error($ch));
    }

    $httpCode = (int) curl_getinfo($ch, CURLINFO_HTTP_CODE);
    if ($httpCode !== 200) {
        throw new RuntimeException('HTTP-Fehler beim Abruf: ' . $httpCode);
    }

    return preg_split("/\r\n|\n|\r/", $response) ?: [];
}

function fetchRemoteLinesViaStream(string $url): array
{
    $context = stream_context_create([
        'http' => [
            'method' => 'GET',
            'timeout' => 300,
            'ignore_errors' => true,
            'header' => implode("\r\n", [
                'User-Agent: AstroMINT MPC Comet Importer/2.0',
                'Accept: text/plain',
            ]),
        ],
        'ssl' => [
            'verify_peer' => true,
            'verify_peer_name' => true,
        ],
    ]);

    $response = @file_get_contents($url, false, $context);
    if ($response === false) {
        $error = error_get_last();
        throw new RuntimeException('Download fehlgeschlagen: ' . ($error['message'] ?? 'unbekannter Fehler'));
    }

    $statusLine = null;
    foreach (($http_response_header ?? []) as $headerLine) {
        if (stripos($headerLine, 'HTTP/') === 0) {
            $statusLine = $headerLine;
            break;
        }
    }

    if ($statusLine !== null && preg_match('/\s(\d{3})\s/', $statusLine, $matches) === 1) {
        $httpCode = (int) $matches[1];
        if ($httpCode !== 200) {
            throw new RuntimeException('HTTP-Fehler beim Abruf: ' . $httpCode);
        }
    }

    return preg_split("/\r\n|\n|\r/", $response) ?: [];
}

function fetchRemoteLinesViaSocket(string $url, ?string $previousError = null): array
{
    $parts = parse_url($url);
    if (!is_array($parts) || ($parts['scheme'] ?? '') !== 'https' || !isset($parts['host'])) {
        throw new RuntimeException('Socket-Fallback unterstuetzt nur HTTPS-URLs.');
    }

    $host = (string) $parts['host'];
    $path = (string) ($parts['path'] ?? '/');
    $query = isset($parts['query']) ? '?' . $parts['query'] : '';
    $target = 'ssl://' . $host . ':443';

    $context = stream_context_create([
        'ssl' => [
            'verify_peer' => true,
            'verify_peer_name' => true,
            'SNI_enabled' => true,
            'peer_name' => $host,
        ],
    ]);

    $errno = 0;
    $errstr = '';
    $socket = @stream_socket_client($target, $errno, $errstr, 30, STREAM_CLIENT_CONNECT, $context);

    if ($socket === false) {
        $detail = $errstr !== '' ? $errstr : 'Verbindung konnte nicht aufgebaut werden';
        if ($previousError !== null && $previousError !== '') {
            $detail .= ' | file_get_contents: ' . $previousError;
        }
        throw new RuntimeException('Download fehlgeschlagen: ' . $detail);
    }

    stream_set_timeout($socket, 300);

    $request = implode("\r\n", [
        'GET ' . $path . $query . ' HTTP/1.1',
        'Host: ' . $host,
        'User-Agent: AstroMINT MPC Comet Importer/2.0',
        'Accept: text/plain',
        'Connection: close',
        '',
        '',
    ]);

    fwrite($socket, $request);
    $response = stream_get_contents($socket);
    fclose($socket);

    if ($response === false || $response === '') {
        throw new RuntimeException('Download fehlgeschlagen: leere Antwort vom Server');
    }

    $separatorPos = strpos($response, "\r\n\r\n");
    if ($separatorPos === false) {
        throw new RuntimeException('Download fehlgeschlagen: ungueltige HTTP-Antwort');
    }

    $headerText = substr($response, 0, $separatorPos);
    $body = substr($response, $separatorPos + 4);
    $headerLines = explode("\r\n", $headerText);
    $statusLine = $headerLines[0] ?? '';

    if (preg_match('/\s(\d{3})\s/', $statusLine, $matches) !== 1) {
        throw new RuntimeException('Download fehlgeschlagen: HTTP-Status unbekannt');
    }

    $httpCode = (int) $matches[1];
    if ($httpCode !== 200) {
        throw new RuntimeException('HTTP-Fehler beim Abruf: ' . $httpCode);
    }

    $headers = [];
    foreach (array_slice($headerLines, 1) as $headerLine) {
        $parts = explode(':', $headerLine, 2);
        if (count($parts) === 2) {
            $headers[strtolower(trim($parts[0]))] = trim($parts[1]);
        }
    }

    if (($headers['transfer-encoding'] ?? '') === 'chunked') {
        $body = decodeChunkedHttpBody($body);
    }

    return preg_split("/\r\n|\n|\r/", $body) ?: [];
}

function decodeChunkedHttpBody(string $body): string
{
    $decoded = '';
    $offset = 0;
    $length = strlen($body);

    while ($offset < $length) {
        $lineEnd = strpos($body, "\r\n", $offset);
        if ($lineEnd === false) {
            break;
        }

        $hexLength = trim(substr($body, $offset, $lineEnd - $offset));
        if ($hexLength === '') {
            break;
        }

        $chunkLength = hexdec($hexLength);
        $offset = $lineEnd + 2;

        if ($chunkLength === 0) {
            break;
        }

        $decoded .= substr($body, $offset, $chunkLength);
        $offset += $chunkLength + 2;
    }

    return $decoded;
}

function parseSoft00CmtLine(string $line): array
{
    $padded = str_pad($line, 172, ' ');

    $designationPacked = requireStringField(readFixedField($padded, 1, 12), 'designation_packed');
    $designationAndName = requireStringField(readFixedField($padded, 103, 158), 'designation_and_name');
    $epochCompact = toNullableCompactDate(readFixedField($padded, 82, 89));
    $orbitType = deriveOrbitType($designationAndName, $designationPacked);
    $yearOfPerihelion = requireIntField(readFixedField($padded, 15, 18), 'year_of_perihelion');
    $monthOfPerihelion = requireIntField(readFixedField($padded, 20, 21), 'month_of_perihelion');
    $dayOfPerihelion = requireFloatField(readFixedField($padded, 23, 29), 'day_of_perihelion');
    $perihelionDistAu = requireFloatField(readFixedField($padded, 32, 39), 'perihelion_dist_au');
    $eccentricity = requireFloatField(readFixedField($padded, 42, 49), 'eccentricity');
    $argPerihelionDeg = requireFloatField(readFixedField($padded, 52, 59), 'arg_perihelion_deg');
    $ascendingNodeDeg = requireFloatField(readFixedField($padded, 62, 69), 'ascending_node_deg');
    $inclinationDeg = requireFloatField(readFixedField($padded, 72, 79), 'inclination_deg');
    $absoluteMagnitudeH = toNullableFloat(readFixedField($padded, 91, 95));
    $slopeParameterG = toNullableFloat(readFixedField($padded, 97, 101));
    $referenceText = toNullableString(readFixedField($padded, 160, 200));

    return [
        ':designation_packed' => $designationPacked,
        ':orbit_type' => $orbitType,
        ':year_of_perihelion' => $yearOfPerihelion,
        ':month_of_perihelion' => $monthOfPerihelion,
        ':day_of_perihelion' => $dayOfPerihelion,
        ':perihelion_dist_au' => $perihelionDistAu,
        ':eccentricity' => $eccentricity,
        ':arg_perihelion_deg' => $argPerihelionDeg,
        ':ascending_node_deg' => $ascendingNodeDeg,
        ':inclination_deg' => $inclinationDeg,
        ':epoch_compact' => $epochCompact,
        ':epoch_date' => $epochCompact !== null ? compactDateToSql($epochCompact) : null,
        ':absolute_magnitude_h' => $absoluteMagnitudeH,
        ':slope_parameter_g' => $slopeParameterG,
        ':designation_and_name' => $designationAndName,
        ':reference_text' => $referenceText,
        ':raw_record' => $line,
        ':source_hash' => buildSourceHash([
            $designationPacked,
            $orbitType,
            $yearOfPerihelion,
            $monthOfPerihelion,
            $dayOfPerihelion,
            $perihelionDistAu,
            $eccentricity,
            $argPerihelionDeg,
            $ascendingNodeDeg,
            $inclinationDeg,
            $epochCompact,
            $absoluteMagnitudeH,
            $slopeParameterG,
            $designationAndName,
            $referenceText,
        ]),
    ];
}

function buildSourceHash(array $parts): string
{
    $normalized = array_map(
        static fn(mixed $value): string => $value === null ? '' : trim((string) $value),
        $parts
    );

    return sha1(implode('|', $normalized));
}

function deriveOrbitType(string $designationAndName, string $designationPacked): string
{
    if (preg_match('/^\d+([A-Z])\//', $designationAndName, $matches) === 1) {
        return $matches[1];
    }

    if (preg_match('/^([A-Z])\//', $designationAndName, $matches) === 1) {
        return $matches[1];
    }

    if (preg_match('/([A-Z])/', ltrim($designationPacked), $matches) === 1) {
        return $matches[1];
    }

    throw new RuntimeException('orbit_type konnte nicht abgeleitet werden');
}

function readFixedField(string $line, int $startColumn, int $endColumn): ?string
{
    $offset = $startColumn - 1;
    $length = $endColumn - $startColumn + 1;
    $value = substr($line, $offset, $length);

    if ($value === false) {
        return null;
    }

    $value = trim($value);
    return $value === '' ? null : $value;
}

function requireStringField(?string $value, string $fieldName): string
{
    if ($value === null || $value === '') {
        throw new RuntimeException("Pflichtfeld fehlt oder ist ungueltig: {$fieldName}");
    }

    return $value;
}

function requireIntField(?string $value, string $fieldName): int
{
    $value = trim((string) $value);
    if ($value === '' || preg_match('/^-?\d+$/', $value) !== 1) {
        throw new RuntimeException("Pflichtfeld fehlt oder ist ungueltig: {$fieldName}");
    }

    return (int) $value;
}

function requireFloatField(?string $value, string $fieldName): float
{
    $value = trim((string) $value);
    if ($value === '' || !is_numeric($value)) {
        throw new RuntimeException("Pflichtfeld fehlt oder ist ungueltig: {$fieldName}");
    }

    return (float) $value;
}

function toNullableCompactDate(?string $value): ?string
{
    $value = trim((string) $value);
    if ($value === '') {
        return null;
    }

    if (preg_match('/^\d{8}$/', $value) !== 1) {
        throw new RuntimeException('epoch_compact ist ungueltig');
    }

    return $value;
}

function toNullableString(?string $value): ?string
{
    if ($value === null) {
        return null;
    }

    $value = trim($value);
    return $value === '' ? null : $value;
}

function toNullableFloat(?string $value): ?float
{
    if ($value === null) {
        return null;
    }

    $value = trim($value);
    if ($value === '' || !is_numeric($value)) {
        return null;
    }

    return (float) $value;
}

function compactDateToSql(string $value): ?string
{
    if (preg_match('/^\d{8}$/', $value) !== 1) {
        return null;
    }

    $year = (int) substr($value, 0, 4);
    $month = (int) substr($value, 4, 2);
    $day = (int) substr($value, 6, 2);

    if (!checkdate($month, $day, $year)) {
        return null;
    }

    return sprintf('%04d-%02d-%02d', $year, $month, $day);
}

function normalizeDbConfig(array $config): array
{
    return [
        'host' => (string) ($config['host'] ?? '127.0.0.1'),
        'port' => (int) ($config['port'] ?? 3306),
        'dbname' => (string) ($config['dbname'] ?? ''),
        'user' => (string) ($config['user'] ?? ''),
        'pass' => (string) ($config['pass'] ?? ''),
        'charset' => (string) ($config['charset'] ?? 'utf8mb4'),
    ];
}

function ensureImportRunsTableExists(PDO $pdo): void
{
    $pdo->exec("
        CREATE TABLE IF NOT EXISTS `app_import_runs` (
          `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
          `source` varchar(32) NOT NULL,
          `source_url` varchar(500) DEFAULT NULL,
          `started_at` datetime NOT NULL,
          `finished_at` datetime DEFAULT NULL,
          `records_received` int(11) NOT NULL DEFAULT 0,
          `records_inserted` int(11) NOT NULL DEFAULT 0,
          `records_updated` int(11) NOT NULL DEFAULT 0,
          `status` varchar(32) NOT NULL DEFAULT 'success',
          `message` text DEFAULT NULL,
          PRIMARY KEY (`id`),
          KEY `idx_app_import_runs_source` (`source`),
          KEY `idx_app_import_runs_started_at` (`started_at`)
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
    ");
}

function startImportRun(PDO $pdo, string $source, ?string $sourceUrl): int
{
    $stmt = $pdo->prepare("
        INSERT INTO `app_import_runs` (
            `source`,
            `source_url`,
            `started_at`,
            `status`,
            `message`
        ) VALUES (
            :source,
            :source_url,
            NOW(),
            'running',
            NULL
        )
    ");
    $stmt->execute([
        ':source' => $source,
        ':source_url' => $sourceUrl,
    ]);

    return (int) $pdo->lastInsertId();
}

function finishImportRun(
    PDO $pdo,
    int $runId,
    int $recordsReceived,
    int $recordsInserted,
    int $recordsUpdated,
    string $status,
    ?string $message
): void {
    $stmt = $pdo->prepare("
        UPDATE `app_import_runs`
        SET
            `finished_at` = NOW(),
            `records_received` = :records_received,
            `records_inserted` = :records_inserted,
            `records_updated` = :records_updated,
            `status` = :status,
            `message` = :message
        WHERE `id` = :id
        LIMIT 1
    ");
    $stmt->execute([
        ':records_received' => $recordsReceived,
        ':records_inserted' => $recordsInserted,
        ':records_updated' => $recordsUpdated,
        ':status' => $status,
        ':message' => $message,
        ':id' => $runId,
    ]);
}

function ensureCometsTableExists(PDO $pdo): void
{
    $pdo->exec(<<<'SQL'
CREATE TABLE IF NOT EXISTS `comets_mpc` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT,
  `designation_packed` varchar(12) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL COMMENT 'Fixed-width MPC-Packbezeichnung aus Soft00Cmt.txt, inkl. Fragmentkennung',
  `orbit_type` char(1) NOT NULL COMMENT 'Abgeleiteter Kometentyp, z. B. C, P oder I',
  `year_of_perihelion` smallint(6) NOT NULL COMMENT 'Jahr des Periheldurchgangs',
  `month_of_perihelion` tinyint(3) unsigned NOT NULL COMMENT 'Monat des Periheldurchgangs',
  `day_of_perihelion` decimal(8,4) NOT NULL COMMENT 'Tag des Periheldurchgangs inklusive Tagesbruchteil',
  `perihelion_dist_au` decimal(12,6) NOT NULL COMMENT 'Periheldistanz q in AE',
  `eccentricity` decimal(12,6) NOT NULL COMMENT 'Exzentrizitaet e',
  `arg_perihelion_deg` decimal(9,4) NOT NULL COMMENT 'Argument des Perihels in Grad',
  `ascending_node_deg` decimal(9,4) NOT NULL COMMENT 'Laenge des aufsteigenden Knotens in Grad',
  `inclination_deg` decimal(9,4) NOT NULL COMMENT 'Bahnneigung in Grad',
  `epoch_compact` char(8) DEFAULT NULL COMMENT 'Epoche des Elementsatzes als YYYYMMDD; kann bei Fragmenten fehlen',
  `epoch_date` date DEFAULT NULL COMMENT 'Epoche des Elementsatzes als SQL-Datum',
  `absolute_magnitude_h` decimal(5,1) DEFAULT NULL COMMENT 'Absolute Helligkeit H laut MPC',
  `slope_parameter_g` decimal(4,1) DEFAULT NULL COMMENT 'Steigungsparameter G laut MPC',
  `designation_and_name` varchar(255) NOT NULL COMMENT 'Lesbare Bezeichnung, z. B. C/1995 O1 (Hale-Bopp)',
  `reference_text` varchar(64) DEFAULT NULL COMMENT 'Referenz aus der MPC-Datei, z. B. MPEC 2026-FC3',
  `raw_record` varchar(255) DEFAULT NULL COMMENT 'Originale Fixed-width-Zeile aus Soft00Cmt.txt',
  `source_hash` char(40) DEFAULT NULL COMMENT 'Hash ueber die relevanten MPC-Quelldaten zur Aenderungserkennung',
  `source_url` varchar(255) DEFAULT NULL COMMENT 'Quelle des letzten Imports',
  `imported_at` datetime NOT NULL DEFAULT current_timestamp() COMMENT 'Zeitpunkt des letzten Imports',
  `created_at` datetime NOT NULL DEFAULT current_timestamp(),
  `updated_at` datetime NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
  PRIMARY KEY (`id`),
  UNIQUE KEY `uq_mpc_comets_designation_packed` (`designation_packed`),
  KEY `idx_mpc_comets_orbit_type` (`orbit_type`),
  KEY `idx_mpc_comets_perihelion_year_month` (`year_of_perihelion`,`month_of_perihelion`),
  KEY `idx_mpc_comets_epoch_compact` (`epoch_compact`),
  KEY `idx_mpc_comets_designation_and_name` (`designation_and_name`),
  KEY `idx_mpc_comets_source_hash` (`source_hash`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='Kometenelemente aus Soft00Cmt.txt des Minor Planet Center'
SQL
    );

    ensureColumnExists($pdo, 'comets_mpc', 'source_hash', "ALTER TABLE `comets_mpc` ADD COLUMN `source_hash` char(40) DEFAULT NULL COMMENT 'Hash ueber die relevanten MPC-Quelldaten zur Aenderungserkennung' AFTER `raw_record`");
    ensureIndexExists($pdo, 'comets_mpc', 'idx_mpc_comets_source_hash', 'ALTER TABLE `comets_mpc` ADD KEY `idx_mpc_comets_source_hash` (`source_hash`)');
}

function ensureColumnExists(PDO $pdo, string $tableName, string $columnName, string $alterSql): void
{
    $stmt = $pdo->prepare('
        SELECT COUNT(*)
        FROM information_schema.COLUMNS
        WHERE TABLE_SCHEMA = DATABASE()
          AND TABLE_NAME = :table_name
          AND COLUMN_NAME = :column_name
    ');
    $stmt->execute([
        ':table_name' => $tableName,
        ':column_name' => $columnName,
    ]);

    if ((int) $stmt->fetchColumn() === 0) {
        $pdo->exec($alterSql);
    }
}

function ensureIndexExists(PDO $pdo, string $tableName, string $indexName, string $alterSql): void
{
    $stmt = $pdo->prepare('
        SELECT COUNT(*)
        FROM information_schema.STATISTICS
        WHERE TABLE_SCHEMA = DATABASE()
          AND TABLE_NAME = :table_name
          AND INDEX_NAME = :index_name
    ');
    $stmt->execute([
        ':table_name' => $tableName,
        ':index_name' => $indexName,
    ]);

    if ((int) $stmt->fetchColumn() === 0) {
        $pdo->exec($alterSql);
    }
}